如何将Replit上的Dialogflow CX Flask Webhook迁移至Google Cloud Functions

迁移方案 #

Dialogflow CX的Webhook请求/响应格式和ES存在差异，你现有代码已经适配CX规范，迁移到Cloud Functions无需修改业务逻辑，仅需调整外层服务适配Cloud Functions的入口规则即可，具体操作如下：

一、代码改造 #

Cloud Functions为无服务器运行环境，无需启动完整的Flask服务监听端口，仅需要编写入口函数处理请求即可，改造后代码如下：

import functions_framework
from flask import jsonify, request

def create_json_response(response_text, updated_params):
    jsonResponse = {
        "fulfillment_response": {
            "messages": [
                {
                    "text": {
                        "text": [
                            str(response_text)
                        ]
                    }
                }
            ]
        }, 
        "sessionInfo": {
            "parameters": updated_params
        }
    }
    return jsonResponse

@functions_framework.http
def dialogflow_cx_webhook(request):
    # 可选：处理跨域预检请求，遇到跨域报错时再开启即可
    if request.method == 'OPTIONS':
        headers = {
            'Access-Control-Allow-Origin': '*',
            'Access-Control-Allow-Methods': 'POST',
            'Access-Control-Allow-Headers': 'Content-Type',
            'Access-Control-Max-Age': '3600'
        }
        return ('', 204, headers)
    
    # 以下业务逻辑和你原有代码完全一致，无需修改
    req = request.get_json(silent=True, force=True)
    fulfillmentInfo = req.get('fulfillmentInfo')
    
    if fulfillmentInfo['tag'] == 'add':
        updated_params = {}
        num1 = int(req.get('sessionInfo').get('parameters').get('number'))
        num2 = int(req.get('sessionInfo').get('parameters').get('number1'))
        total = str(num1 + num2)
        updated_params['previous1'] = num1
        updated_params['previous2'] = num2
        updated_params['number'] = None
        updated_params['number1'] = None
        jsonResponse = create_json_response(total, updated_params)
    elif fulfillmentInfo['tag'] == 'multiply':
        updated_params = {}
        num1 = int(req.get('sessionInfo').get('parameters').get('number'))
        num2 = int(req.get('sessionInfo').get('parameters').get('number1'))
        total = str(num1 * num2)
        updated_params['previous1'] = num1
        updated_params['previous2'] = num2
        updated_params['number'] = None
        updated_params['number1'] = None
        jsonResponse = create_json_response(total, updated_params)
    else:
        jsonResponse = create_json_response("未匹配到对应操作标签", {})
    
    # 可选：配置跨域头，生产环境建议替换*为Dialogflow官方域名
    headers = {
        'Access-Control-Allow-Origin': '*'
    }
    return (jsonify(jsonResponse), 200, headers)

依赖配置（requirements.txt）：

functions-framework==3.*
flask==2.*

二、部署步骤 #

• 打开Google Cloud Console进入Cloud Functions页面，点击创建函数
• 环境选择第2代，自定义函数名称，选择就近区域，触发类型选HTTP，测试阶段可勾选「允许未通过身份验证的调用」，生产环境建议配置专属权限
• 点击下一步，运行时选择Python 3.9及以上版本，入口函数填写代码里的入口名 dialogflow_cx_webhook
• 将改造后的main.py和requirements.txt内容分别粘贴到对应编辑器窗口
• 点击部署，等待部署完成后复制触发器页面的HTTP URL
• 进入Dialogflow CX控制台对应Agent的Webhook管理页面，新建Webhook并粘贴刚才复制的URL，保存后即可在对话流中调用该Webhook

注意事项 #

• 若需要保留原有根路径返回Hello World的功能，可在入口函数开头判断request.path为/时直接返回对应字符串即可
• 生产环境请关闭未认证调用，给Dialogflow服务账号分配Cloud Functions调用者权限，同时收紧跨域权限避免被恶意调用

内容的提问来源于stack exchange，提问作者buttermilk