如何扩展Dialogflow Webhook超时时间以适配OpenAI API调用？

我尝试扩展Google Dialogflow Webhook请求的最大超时时间（默认5秒），因为调用OpenAI API时常超过5秒。我的思路是通过触发Dialogflow的followupEventInput重置超时（最多2次），同时后台调用OpenAI API，成功后返回结果至Dialogflow，但线程功能使用有误，目前无法实现需求。

现有代码 #

import os
import openai
import time
import backoff
from datetime import datetime, timedelta
from flask import Flask, request, render_template
from threading import Thread
import asyncio

app = Flask(__name__)

conversation_History = ""
user_Input = ""
reply='' 
answer = ""

@app.route('/') 
def Default(): 
    return render_template('index.html')

@backoff.on_exception(backoff.expo, openai.error.RateLimitError)
def ask(question):
  global conversation_History
  global answer
  global reply
  openai.api_key = os.getenv("gtp_Secret_Key")  
  #start_sequence = "\nAI:"
  #restart_sequence = "\nHuman: "
  response = openai.Completion.create(
    model="text-davinci-003",
    prompt="I am a chatbot from OpenAI. I'm happy to answer your questions.\nHuman:" + conversation_History + " "+ question +"\nAI: ",    
    temperature=0.9,
    max_tokens=500,
    top_p=1,
    frequency_penalty=0,
    presence_penalty=0.6,
    stop=[" Human:", " AI:"]
    )
  conversation_History = conversation_History + question + "\nAI" + answer + "\nHuman:"
  answer = response.choices[0].text  

def broadbridge_webhook_results():
  global answer
  
  now = datetime.now()
  current_time = now.strftime("%H:%M:%S")
  print("Current Time =", current_time)

  extended_time = now + timedelta(seconds=3)
  print("extended Time =", extended_time.time())

  req = request.get_json(force=True)

  action = req.get('queryResult').get('action')
  reply='' 

  if action=='input.unknown' or action=='input.welcome':
    time.sleep(3.5)

    if now<=extended_time and not len(answer) == 0:
      reply={
              "fulfillmentText": answer,
              "source": "webhookdata"
            }

    reply={      
      "followupEventInput": {        
        "name": "extent_webhook_deadline",        
        "languageCode": "en-US"
        }
      }

  if action=='followupevent':
    print("enter into first followup event")
    time.sleep(3.5)

    if now<=extended_time and not len(answer) == 0:
      reply={
        "fulfillmentText": answer,
        "source": "webhookdata"
      }

    reply={      
      "followupEventInput": {        
        "name": "extent_webhook_deadline_2",        
        "languageCode": "en-US"
        }
      }
    
  if action=='followupevent_2':
    print("enter into second followup event")
    time.sleep(3.5)

    reply={
      "fulfillmentText": answer,
      "source": "webhookdata"
    }
        
    print("Final time of execution:=>", now.strftime("%H:%M:%S"))

@app.route('/webhook', methods=['GET', 'POST'])
def webhook():
  global answer
  global reply

  answer=""
  req = request.get_json(silent=True, force=True)
  user_Input = req.get('queryResult').get('queryText')
  Thread(target=broadbridge_webhook_results()).start()
  Thread(target=ask(user_Input)).start()
  
  return reply

#conversation_History = conversation_History + user_Input + "\nAI" + answer + "\nHuman:"
#if now<=extended_time and not len(answer) == 0:
  
if __name__ == '__main__':
  app.run(host='0.0.0.0', port=8080)

核心问题分析 #

1. 线程启动错误：启动线程时直接执行了函数（Thread(target=broadbridge_webhook_results()).start()），导致函数在主线程中同步执行，而非子线程异步执行，主线程无法及时返回Dialogflow响应。
2. Request对象线程不安全：子线程中调用request.get_json()，Flask的Request对象绑定到主线程，子线程无法访问，会抛出异常。
3. 全局变量冲突：conversation_History、answer等全局变量无会话隔离，多用户请求会互相覆盖数据。
4. 逻辑判断无效：now<=extended_time的判断无意义，now是函数启动时间，extended_time为now+3秒，3.5秒sleep后now必然小于extended_time。
5. 对话历史更新错误：ask函数中先更新对话历史再赋值answer，导致历史中写入空值。

修复后的代码（会话隔离版） #

import os
import openai
import time
import backoff
from flask import Flask, request
from threading import Thread
from collections import defaultdict

app = Flask(__name__)

# 按会话ID存储用户数据，避免多请求冲突
session_data = defaultdict(lambda: {"history": "", "answer": None})
openai.api_key = os.getenv("gtp_Secret_Key")  

@app.route('/') 
def Default(): 
    return "Dialogflow Webhook Running"

@backoff.on_exception(backoff.expo, openai.error.RateLimitError)
def ask(session_id, question):
    """后台异步调用OpenAI，结果存入会话数据"""
    data = session_data[session_id]
    prompt = f"I am a chatbot from OpenAI. I'm happy to answer your questions.\nHuman:{data['history']} {question}\nAI: "
    response = openai.Completion.create(
        model="text-davinci-003",
        prompt=prompt,
        temperature=0.9,
        max_tokens=500,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0.6,
        stop=[" Human:", " AI:"]
    )
    answer = response.choices[0].text.strip()
    # 先赋值结果再更新对话历史
    data["answer"] = answer
    data["history"] = f"{data['history']}{question}\nAI: {answer}\nHuman:"

@app.route('/webhook', methods=['POST'])
def webhook():
    req = request.get_json(silent=True, force=True)
    session_id = req.get('session')
    query_text = req.get('queryResult').get('queryText')
    action = req.get('queryResult').get('action')
    data = session_data[session_id]

    if action in ('input.unknown', 'input.welcome'):
        # 重置当前会话的AI结果
        data["answer"] = None
        # 异步启动OpenAI调用线程
        Thread(target=ask, args=(session_id, query_text)).start()
        # 立即返回followup事件，重置Dialogflow超时
        return {      
            "followupEventInput": {        
                "name": "extent_webhook_deadline",        
                "languageCode": "en-US"
            }
        }
    elif action == 'followupevent':
        # 检查AI是否返回结果
        if data["answer"]:
            return {
                "fulfillmentText": data["answer"],
                "source": "webhookdata"
            }
        # 未返回结果，继续触发下一次超时重置
        return {      
            "followupEventInput": {        
                "name": "extent_webhook_deadline_2",        
                "languageCode": "en-US"
            }
        }
    elif action == 'followupevent_2':
        # 最后一次请求，返回结果或超时提示
        final_answer = data["answer"] if data["answer"] else "抱歉，请求超时，请稍后重试。"
        return {
            "fulfillmentText": final_answer,
            "source": "webhookdata"
        }
    # 默认错误响应
    return {"fulfillmentText": "抱歉，出现错误，请重试。"}

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

关键修复说明 #

1. 线程启动修正：使用Thread(target=ask, args=(session_id, query_text)).start()传递函数对象和参数，确保异步执行。
2. 会话隔离：通过session_id区分用户，用session_data存储每个会话的对话历史和AI结果，避免多请求数据冲突。
3. Request对象处理：所有请求数据在主线程中获取，子线程仅处理业务逻辑，不再访问Request对象。
4. 逻辑简化：去掉无效时间判断，按请求action分支处理：首次请求返回followup事件，后续请求检查AI结果，有结果则返回，无结果继续触发超时重置。
5. 对话历史修正：先获取OpenAI结果再更新对话历史，确保历史内容准确。

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