基于Google Cloud Functions构建OpenAI-Dialogflow CX Webhook部署问题

核心问题分析 #

你的部署失败及功能异常主要来自以下几个代码逻辑和格式错误：

1. Dialogflow CX 请求解析错误 #

Dialogflow CX 发送给Webhook的请求中，用户查询文本不在根节点的text字段，而是嵌套在queryInfo.text路径下。你的代码直接判断'text' in request_json会触发400错误，导致Webhook无法正常接收请求。

2. 无状态环境下的全局会话变量滥用 #

Cloud Functions是无状态服务，全局变量convo会在多个用户请求、实例重启时共享或丢失数据，完全无法实现正确的会话上下文管理。

3. 缺失关键错误处理 #

• 调用OpenAI API后未判断响应是否成功，若返回错误（如密钥无效、配额不足），response.get('choices')[0]会直接抛出异常导致函数崩溃。
• Secret Manager获取密钥失败时未中断流程，会导致后续OpenAI请求因无有效令牌失败。

4. 潜在权限问题 #

部署Cloud Functions的服务账号需要具备Secret Manager Secret Accessor权限，否则无法读取OpenAI密钥，直接导致API调用失败。

修复后的代码 #

import openai
import json
import requests
from google.cloud import secretmanager

# 初始化Secret Manager客户端
client = secretmanager.SecretManagerServiceClient()

def get_secret(secret_name, project_id, version_id='latest'):
    """从Google Cloud Secret Manager获取密钥"""
    resource_name = f"projects/{project_id}/secrets/{secret_name}/versions/{version_id}"
    try:
        response = client.access_secret_version(request={"name": resource_name})
        return response.payload.data.decode("UTF-8")
    except Exception as e:
        print(f"获取密钥失败: {e}")
        raise  # 抛出异常中断流程，避免后续无效请求

def query_gpt(prompt, api_key):
    """调用OpenAI Completion API"""
    body = {
        "model": "text-davinci-003",
        "prompt": prompt,
        "max_tokens": 200,
        "temperature": 0.9,
        "top_p": 1,
        "n": 1,
        "frequency_penalty": 0,
        "presence_penalty": 0.6  
    }
    headers = {"Authorization": f"Bearer {api_key}"}
    res = requests.post('https://api.openai.com/v1/completions', json=body, headers=headers)
    res.raise_for_status()  # 检查HTTP请求是否成功
    return res.json()

def webhook(request):
    """Cloud Functions入口函数"""
    if request.method != 'POST':
        return ('仅接受POST请求', 405)

    request_json = request.get_json(silent=True)
    if not request_json:
        return ('请求格式错误', 400)
    
    # 正确解析Dialogflow CX的用户查询文本
    try:
        user_query = request_json['queryInfo']['text']
    except KeyError:
        return ('无法获取用户查询', 400)
    
    # 从Dialogflow CX的session参数中读取会话历史（替代全局变量）
    session_params = request_json.get('sessionInfo', {}).get('parameters', {})
    convo = session_params.get('conversation_history', [])
    convo.append(f'User: {user_query}')
    convo.append("Addie:")
    prompt = "\n".join(convo)

    # 获取OpenAI密钥并调用API
    try:
        openai_key = get_secret('openai-api-key', 'my-project-id')
        gpt_response = query_gpt(prompt, openai_key)
        result = gpt_response['choices'][0]['text'].strip('\n')
        convo.append(result)
    except Exception as e:
        print(f"处理请求失败: {e}")
        result = "抱歉，我现在无法为你提供服务，请稍后再试。"
        convo = []  # 出错时重置会话历史
    
    # 构造返回给Dialogflow CX的响应，同时更新会话参数
    return json.dumps({
        'fulfillment_response': {
            'messages': [{
                'text': {
                    'text': [result]
                },
                'responseType': 'TEXT',
                'source': 'VIRTUAL_AGENT'
            }]
        },
        'sessionInfo': {
            'parameters': {
                'conversation_history': convo
            }
        }
    })

额外配置步骤 #

1. 权限配置：给Cloud Functions的服务账号添加roles/secretmanager.secretAccessor角色，确保能读取Secret Manager中的密钥。
2. Dialogflow CX配置：在Dialogflow CX的Webhook设置中，确保启用会话参数传递，并允许Webhook更新会话参数。
3. 测试验证：部署后使用Dialogflow CX的测试工具发送请求，查看Cloud Functions的日志排查错误。

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