我正在尝试使用流程 B 中的数据存储页面生成流程 A 中问题的答案。为此,我将使用 Webhook。我从 Google 的 qwiklabs 资源中获取了此代码。但是,我似乎无法让它运行。这是我的代码:
import re
import json
import logging
from dfcx_scrapi.tools.webhook_util import WebhookUtil
from dfcx_scrapi.core.conversation import DialogflowConversation
# logging config
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)-8s %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
DATA_STORE_PAGE = 'projects/PROJECT-ID/locations/LOCATION-ID/agents/AGENT-ID/flows/FLOW-ID/pages/PAGE-ID'
def get_session_id(req):
"""Get the current Session ID."""
return req['sessionInfo']['session']
def call_data_store(user_query: str) -> str:
"""Calls Data Store route and retrieves response."""
agent_id = '/'.join(DATA_STORE_PAGE.split('/')[:6])
dc = DialogflowConversation(agent_id=agent_id)
res = dc.reply(
send_obj={'text': user_query},
current_page=DATA_STORE_PAGE
)
return res
def _get_uri(sources: str) -> str:
"""Extracts the URI from the top returned source."""
uri = None
pattern = r'"uri":"(.*?)"'
res = re.search(pattern, sources)
if res:
uri = res.group(1)
uri = uri.replace("\\u003d", "=")
return uri
def _get_title(sources: str) -> str:
"""Extracts the Title from the top returned source."""
title = None
pattern = r'"title":"(.*?)"'
res = re.search(pattern, sources)
if res:
title = res.group(1)
return title
def _get_snippet(sources: str) -> str:
"""Extracts the info card snippet from the top returned source."""
snippet = None
pattern = r'"snippet":"(.*?)"'
res = re.search(pattern, sources)
if res:
snippet = res.group(1)
return snippet
def main(request, debugging=False):
"""Main entrypoint"""
logging.info('###### [cx_webhook] - CFX Triggered! ######')
wu = WebhookUtil()
if debugging:
req = request
else:
req = request.get_json()
logging.info('* [cx_webhook] Incoming Request: %s', req)
webhook_tag = wu.get_tag(req)
if webhook_tag == 'get_data_store_response':
logging.info('* [cx_webhook] DATA STORE TAG RECEIVED.')
text = wu.get_user_utterance(req)
res = call_data_store(text)
logging.info('* INFOBOT RESPONSE: %s', res)
response_text = res['response_messages'][0]['text']
response_params = res['params']
title = _get_title(response_params['sources'])
uri = _get_uri(response_params['sources'])
snippet = _get_snippet(response_params['sources'])
session_info = wu.build_session_info({"title": title, "uri": uri, "snippet": snippet})
message = wu.build_response(response_text, session_info=session_info)
else:
logging.info('* [cx_webhook] No Webhook Tag Received.')
res = json.dumps(message)
logging.info(res)
return res
我的要求.txt如下:
functions-framework==3.*
dfcx-scrapi
错误信息: 来自 Google Cloud 的错误消息
我尝试为其分配更多内存,更改Python版本,以及在requirements.txt中添加包,但似乎没有任何效果。
尝试在requirement.txt中添加wheel和pandas。这应该可以解决问题。