Agent 核心循环：Event Generator 模式

Agent 核心循环：Event Generator 模式

学习目标

理解 Khoj 如何通过 event_generator() 函数实现一个工具增强的流式 Agent 循环，以及 WebSocket 中断、断连保存等高级特性。

前置知识

本章涉及 Agent 控制流的通用原理，建议先阅读：

• Agent 核心循环设计
• 流式传输架构

下文假设你已理解上述概念，直接聚焦 Khoj 的 Event Generator 具体实现。

项目实践

入口：HTTP 与 WebSocket 双通道

Khoj 的聊天接口通过两条路由暴露：

• POST /api/chat — HTTP 请求，返回 StreamingResponse
• GET /api/chat/ws — WebSocket 连接，通过 WebSocketConnectionManager 管理

两者的核心逻辑都指向同一个函数：event_generator()（位于 src/khoj/routers/helpers.py，约 800 行）。这是 Khoj 的 Agent 核心循环。

Event Generator 的执行流程

关键设计决策

1. 工具顺序执行而非并行

查询 → 命令解析 → [记忆拉取] → 工具1 → 工具2 → ... → LLM → 流式返回

工具按命令类型顺序调用，而非并行。例如 /research 命令先执行在线搜索，再读取网页，最后运行代码分析。这样做的好处是后一个工具可以依赖前一个工具的结果（如搜索结果决定要读取哪些网页），牺牲了延迟但保证了确定性。

2. ResponseWithThought 异步生成器

响应通过一个自定义的异步生成器逐事件返回。每个事件可以是：

• 思考过程（thought）
• 流式文本片段
• 状态更新（如”正在搜索…”）
• 工具结果
• 结束标记

3. 断连监控后台任务

monitor_disconnection() 是一个独立运行的后台协程，持续检测 WebSocket 连接状态。当客户端断开时：

• 设置 cancellation_event
• 将当前对话状态保存到数据库
• 允许正在运行的子任务（如 research、operator）优雅退出

4. WebSocket 中断传递到子任务

用户可以通过 WebSocket 发送中断信号。这个信号不仅停止当前 Agent 循环，还可以通过 child_interrupt_queue 传递到正在运行的子任务（如一个正在执行的 research 任务）。子任务检查队列中的消息，决定是中止还是用新指令继续。

Deepthought 模式

对于推理密集型查询，Khoj 支持”deepthought”模式：自动将请求路由到 reasoning 模型（如 o1/o3 系列），使用不同的系统提示和参数。这通过 ChatModel 的模型槽位系统实现。

问题与规避

问题	影响	规避策略
工具顺序执行导致延迟	用户等待时间长	对单工具命令（如纯 /notes）跳过空工具；使用流式状态更新保持用户感知
WebSocket 连接池管理	连接泄漏导致 OOM	WebSocketConnectionManager 设置每用户最大 5-10 连接，断连时自动清理
断连时部分状态丢失	对话历史不完整	monitor_disconnection() 在检测到断连时立即保存当前状态
中断传递的时序竞争	中断信号可能被忽略	使用 asyncio.Queue 而非 asyncio.Event，保证中断消息不丢失

设计取舍

为什么选择 StreamingResponse 而非 Server-Sent Events？

Khoj 使用 FastAPI 的 StreamingResponse（配合自定义事件格式）而非标准 SSE。原因是 WebSocket 通道已经需要自定义帧格式，HTTP 通道保持一致性更简单。缺点是缺少 SSE 标准库的断线重连支持。

为什么不使用 LangGraph 或类似的编排框架？

Khoj 的核心循环是手写的 event_generator()，而非基于 LangGraph 的状态图。原因是 Khoj 的工具管道相对固定（搜索 → 读取 → 回答），不需要复杂的条件分支和循环。手写代码更直接，且可以精确控制流式输出的每个事件。

参考来源

• FastAPI StreamingResponse 文档
• WebSocket 协议 RFC 6455