Deep-Searcher项目中的实时消息回调机制设计与实现

背景与需求分析

在AI代理开发过程中，实时监控代理的工作状态对于提升用户体验至关重要。Deep-Searcher作为一个开源项目，其核心功能是通过多轮检索和推理来回答用户查询。传统实现中，前端用户往往无法实时了解代理的内部工作流程，只能被动等待最终结果。

技术方案设计

基于Celery和SQLite的异步消息回调系统能够有效解决这个问题。该方案包含三个核心组件：

1. 任务分发层：使用Celery作为分布式任务队列，接收前端查询请求并异步执行
2. 状态存储层：采用SQLite数据库持久化存储任务执行过程中的各种状态信息
3. 消息回调层：实现前端轮询机制，实时获取任务执行进度

关键实现细节

数据库设计

SQLite数据库中设计了专门的任务状态表，包含以下字段：

• task_id：唯一任务标识符
• status：当前任务状态（运行中/完成/错误）
• answer：生成的最终答案
• chunks：检索到的文本片段
• debug_message：详细的调试信息
• timestamp：状态更新时间戳

状态更新机制

在代理执行的每个关键节点（如开始检索、生成摘要等）都会更新数据库状态。采用增量更新策略，确保历史信息不会丢失：

# 示例状态更新代码
data = {
    'task_id': input_data["id"],
    'status': "Running",
    'answer': "",
    'chunks': "",
    'debug_message': "正在处理检索结果..."
}
update_agent_msg(data)

前端轮询实现

前端通过定期查询数据库获取最新状态，实现了准实时监控：

def print_task_msg(id):
    while True:
        task = get_task_by_id(task_id=id)
        if task and task['status'] == 'Done':
            print(f"任务完成: {task['answer']}")
            break
        time.sleep(0.1)  # 100ms轮询间隔

技术优势

1. 低延迟：通过短周期轮询（100ms）实现准实时反馈
2. 可靠性：SQLite确保状态持久化，即使服务重启也不丢失
3. 可扩展性：Celery架构支持分布式部署，适合高并发场景
4. 调试友好：完整记录代理执行过程，便于问题排查

实际应用效果

该机制实施后能够：

• 实时显示代理的思考过程（如"正在检索相关文档"）
• 及时反馈执行错误信息
• 保留完整的执行历史记录
• 支持多用户并发查询

总结与展望

这种消息回调机制为Deep-Searcher项目带来了显著的体验提升。未来可以考虑：

1. 改用WebSocket实现真正的实时推送
2. 增加状态变更的事件通知机制
3. 支持更细粒度的执行步骤追踪
4. 添加执行耗时统计等监控指标

该方案不仅适用于Deep-Searcher项目，也可为其他需要长时任务处理的AI系统提供参考。通过简单的技术组合，实现了专业级的任务监控体验。