Flask聊天机器人API超时处理：返回等待提示并持续执行请求

你这个需求的核心是把耗时的API调用放到后台异步执行，同时立即给聊天机器人返回等待提示，等后台任务完成后再把最终结果推送给机器人。下面给你两种可行的实现方案，都是基于Python Flask生态的：

方案一：使用线程（简单快速，适合小型场景） #

利用Python自带的threading模块，把runapi的执行放到单独线程里，主线程可以直接返回等待提示，后台线程继续执行任务，完成后自动调用apiasyncresp发送最终结果。

修改后的代码示例：

import threading
import requests
import json
from collections import OrderedDict
from flask import request, jsonify
from requests.auth import HTTPBasicAuth

# 假设这些是你的全局配置项
user = "your_mulesoft_username"
password = "your_mulesoft_password"

def runapi_and_notify(uri, spaceName):
    """后台执行API调用并推送结果给聊天机器人"""
    try:
        # 给API调用设置更长的超时，避免提前中断
        r = requests.get(uri, auth=HTTPBasicAuth(user, password), timeout=300)
        parsed = json.loads(r.text, object_pairs_hook=OrderedDict)
        # 任务完成后推送最终结果
        apiasyncresp(parsed, spaceName)
    except Exception as e:
        # 处理API调用失败的情况，可自定义错误提示
        apiasyncresp(f"查询失败：{str(e)}", spaceName)

def generateresponse(event, spaceName):
    # 先返回等待提示
    prompt_text = "waiting for response"
    # 根据event生成对应的Mulesoft API地址（需根据你的业务逻辑调整）
    uri = f"your_mulesoft_api_endpoint?query={event['query']}"
    # 启动后台线程执行任务，daemon=True确保Flask退出时线程自动终止
    threading.Thread(target=runapi_and_notify, args=(uri, spaceName), daemon=True).start()
    # 立即返回提示文本
    return prompt_text

def on_event():
    event = request.get_json()
    # 从event中提取spaceName，或使用全局配置
    spaceName = event.get("spaceName", "default_space")
    resp = generateresponse(event, spaceName)
    # 返回等待提示给聊天机器人
    return jsonify({"response": resp})

注意点： #

• 线程设为daemon=True，避免Flask停止后残留后台进程。
• 必须处理API调用的异常，防止后台线程报错导致程序崩溃。
• 若Flask采用多进程部署（如Gunicorn多worker），线程方案依然可用，但要注意全局变量的共享问题。

方案二：使用Celery（适合生产环境，可扩展） #

如果你的应用是生产环境，需要更可靠的异步任务管理（比如任务重试、状态监控），推荐用Celery配合消息队列（如Redis或RabbitMQ）实现。

步骤1：安装依赖 #

pip install celery redis

步骤2：配置Celery实例 #

在Flask项目中初始化Celery：

from celery import Celery
from flask import Flask

app = Flask(__name__)
# 配置Celery的消息队列和结果存储（这里用Redis示例）
app.config.update(
    CELERY_BROKER_URL='redis://localhost:6379/0',
    CELERY_RESULT_BACKEND='redis://localhost:6379/0'
)

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.config)
    return celery

celery = make_celery(app)

步骤3：定义异步任务并修改业务逻辑 #

import requests
import json
from collections import OrderedDict
from flask import request, jsonify
from requests.auth import HTTPBasicAuth

user = "your_mulesoft_username"
password = "your_mulesoft_password"

@celery.task
def runapi_task(uri, spaceName):
    """Celery异步任务：执行API调用并推送结果"""
    try:
        r = requests.get(uri, auth=HTTPBasicAuth(user, password), timeout=300)
        parsed = json.loads(r.text, object_pairs_hook=OrderedDict)
        apiasyncresp(parsed, spaceName)
    except Exception as e:
        apiasyncresp(f"查询失败：{str(e)}", spaceName)

def generateresponse(event, spaceName):
    uri = f"your_mulesoft_api_endpoint?query={event['query']}"
    # 提交异步任务到Celery队列
    runapi_task.delay(uri, spaceName)
    # 立即返回等待提示
    return "waiting for response"

def on_event():
    event = request.get_json()
    spaceName = event.get("spaceName", "default_space")
    resp = generateresponse(event, spaceName)
    return jsonify({"response": resp})

为什么选Celery？ #

• 支持任务重试、超时控制、状态监控，比线程方案更稳定。
• 适配多worker、分布式部署场景，不会因进程退出丢失任务。
• 可通过Flower工具监控任务状态，方便排查问题。

核心注意事项 #

1. 结果推送时机：最终的查询结果是在后台任务完成后通过apiasyncresp推送给聊天机器人，而非通过generateresponse返回——因为generateresponse已经提前返回了等待提示。
2. URI生成逻辑：需根据event中的具体内容生成对应的Mulesoft API地址，这部分要结合你的业务需求调整代码。
3. 超时设置：给requests.get设置合理的超时时间（如300秒），避免后台任务无限挂起。

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