Agent 核心循环（流式 Stream 架构）

Agent 核心循环（流式 Stream 架构）

学习目标

读完本章后，你将能够：

• 理解 Agent::reply() 返回 BoxStream<'_, Result<AgentEvent>> 的流式架构设计
• 掌握 Turn 循环内部的完整执行流程：准备上下文 → 流式 LLM 调用 → Inspector 审查 → 并发工具执行
• 理解上下文修复管道（fix_conversation）如何解决消息格式不一致问题
• 分析 Goose 选择流式而非同步循环的权衡与收益
• 在自己的项目中设计类似的流式 Agent 循环

前置知识

• [[agent-design/01-overview]] — Agent 的 Session/Turn 两层循环与事件驱动架构
• [[streaming/01-streaming-architecture]] — Token/块/预览三层流式模型与分块算法

---

项目实践

reply() 入口：一切都是流

Goose 的 Agent::reply() 方法签名定义了它的核心设计理念：

reply(user_message, session_config, cancel_token)
    -> BoxStream<'_, Result<AgentEvent>>

与同步循环返回 Future<Response> 不同，Goose 返回一个异步事件流。每次 Turn 产生的事件通过流逐个发射，前端（Desktop / CLI / Server）可以实时消费，无需等待整个 Turn 完成。

AgentEvent 定义了四种核心事件类型：

事件类型	说明	消费方
Message(text_delta)	流式文本输出增量	UI 实时渲染
McpNotification(...)	MCP 工具调用状态（开始、完成、错误）	UI 工具状态指示器
HistoryReplaced	上下文压缩后历史对话被替换	Session 管理器更新持久化
ToolConfirmation(...)	需要用户确认才能执行的工具	前端弹出审批对话框

这种设计将 Agent 内部执行的每一步都暴露为可观察的事件，前端可以根据事件类型做出不同响应，实现真正的实时交互。

Turn 循环内部流程

reply() 内部调用 reply_internal()，这是一个带有循环的异步流构建器。整个流程如下：

每个 Turn 循环的关键步骤：

第一步：取消检查与 Turn 限制。 默认最大 1000 turn，防止 Agent 陷入无限循环。每次循环前检查 CancellationToken，用户可随时中断。

第二步：自动压缩检查。 当对话消息数量超过阈值时，触发 compact_messages()，自动压缩历史对话。压缩完成后发射 HistoryReplaced 事件，通知前端历史已被替换。

第三步：流式 LLM 调用。 调用 Provider 的流式接口，将模型输出作为流逐 chunk 消费。文本 delta 立即封装为 AgentEvent::Message 发射。

第四步：工具分类。 模型返回的工具调用被分为两类：Frontend 工具（由 UI 侧执行）和 Server-side 工具（由后端执行）。

第五步：Inspector Pipeline（仅非 Chat 模式）。 Server-side 工具在执行前经过安全审查链：安全审查（SecurityInspector）、出口审查（EgressInspector）、对抗审查（AdversaryInspector）、重复检查（RepetitionInspector）。

第六步：并发工具执行。 通过 stream::select_all 将多个工具调用合并为单一流，实现真正的并发执行。

并发工具执行：stream::select_all

多个独立工具调用可以并行执行，Goose 使用 stream::select_all 合并多个工具流：

// 伪代码：并发工具执行
for each tool_call in tool_requests:
    // 每个工具执行返回一个 Stream<AgentEvent>
    tool_stream = execute_tool(tool_call, category)

// 合并所有工具流为一个流，任一工具产生事件即发射
merged = stream::select_all([tool_stream_1, tool_stream_2, ...])

// 消费合并后的流，将事件发射给前端
while let Some(event) = merged.next():
    yield event

设计优势：

• 多个独立工具（如同时读取文件和执行 grep）可以并行执行，减少整体延迟
• 每个工具的进度事件（开始、完成、错误）仍然可以实时传递给 UI
• 无需等待所有工具完成，先完成的工具结果可以先行处理

上下文修复管道（fix_conversation）

在每次调用 LLM 前，Goose 对对话历史应用一系列修复操作，确保消息格式满足 API 要求：

fix_conversation(messages):
    1. 合并连续同角色消息
       如果连续两条 user 或 assistant 消息，合并为一条

    2. 移除空消息
       删除无内容或仅含空白字符的消息

    3. 修复孤立的工具调用
       没有对应 tool_result 的 tool_call 被清理

    4. 确保首尾消息是 user
       第一条消息必须是用户输入
       最后一条消息也必须是用户输入

    5. 空对话占位
       如果对话为空，添加占位消息防止 API 报错

为什么需要修复？ LLM API 对消息格式有严格要求：user/assistant 必须交替出现、每个 tool_call 必须有对应的 tool_result。Agent 循环中的压缩、中断、错误恢复等操作可能破坏这些不变量。修复管道在每次调用前确保格式正确，比依赖调用方保证更可靠。

自动压缩（Auto-Compaction）

当对话消息数量超过阈值（默认值通过 DEFAULT_COMPACTION_THRESHOLD 配置）时，Goose 自动触发压缩：

if messages.len() > compaction_threshold:
    // 发射状态通知，告知用户正在压缩
    yield AgentEvent::Message("goose is compacting the conversation...")

    // 调用压缩函数，使用小模型生成摘要
    compact_messages(messages, small_model_provider)

    // 压缩后修复对话结构
    fix_conversation(messages)

    // 通知前端历史已被替换
    yield AgentEvent::HistoryReplaced

压缩完成后，Agent 使用压缩后的对话继续循环。压缩最多允许 2 次恢复重试，防止压缩后的对话仍然超出上下文限制。

---

问题与规避

Turn 循环卡死

问题：当 Agent 连续调用工具但每次都返回相同错误时，Turn 循环可能无限重复，永远不会达到终止条件。

规避策略：

• Turn 数量上限：默认 1000 turn 硬限制，达到后强制终止
• RepetitionInspector：实时检测工具调用重复模式，当同一工具连续调用 N 次且参数相似时，审查链可以拒绝执行
• CancellationToken：用户可随时通过前端发送中断信号，立即终止当前 Turn

上下文压缩后的 API 错误

问题：压缩操作可能产生格式不正确的消息（如孤立的 tool_call、连续的同角色消息），导致后续 LLM API 调用返回格式错误。

规避策略：

• 压缩后必修复：compact_messages() 完成后立即调用 fix_conversation()，确保消息格式正确
• 压缩前检查：check_if_compaction_needed() 在压缩前评估当前消息列表，避免过度压缩导致信息丢失
• 小模型选择：压缩任务使用专门的小模型（通过 get_small_model() 路由），而非主模型，节省成本

Inspector 审查升级不可逆

问题：Inspector 链的审查结果只能升级（从允许到拒绝），不能降级（从拒绝到允许）。这意味着如果某个 Inspector 拒绝了工具调用，后续 Inspector 无法覆盖这个决定。

规避策略：

• 这是有意为之的设计。安全策略遵循”最严格者胜出”原则，避免低优先级 Inspector 覆盖高优先级的安全判断
• Inspector 按优先级排列：安全审查 → 权限审查 → 对抗审查 → 重复检查

---

设计取舍

流式循环 vs 同步循环

维度	同步循环（如 Hermes）	流式循环（Goose）
返回类型	Future<Response>	Stream<Event>
UI 更新	完成后一次性展示	实时逐 chunk 渲染
工具执行	全部完成后返回	边流式输出边并行启动
中断	需要显式检查 CancellationToken	流自然终止
内存	累积全部响应	逐 chunk 消费
实现复杂度	低	高（需要处理流合并、背压）

为什么 Goose 选择流式？

1. 三端统一需求：Desktop（Electron）、CLI、Server 三种前端对响应速度的要求不同。流式架构让每个前端可以自行决定消费节奏
2. 并发工具执行：流式模型天然支持”边输出边执行工具”，多个工具可以并行启动
3. 实时可观测性：每个中间状态（工具开始、工具完成、压缩进行）都可以实时通知前端
4. 内存效率：长对话场景下，流式消费避免将整个响应累积在内存中

代价：

• 流的合并、排序、错误处理比同步调用复杂得多
• stream::select_all 需要在多个工具流之间公平调度
• 前端需要处理事件流的乱序到达和可能的重复事件

修复管道 vs 调用方保证

问题：谁负责保证消息格式正确？

方案	优势	代价
调用方保证	修复逻辑集中在一处，每次操作后手动修复	容易遗漏、新增操作时需要记住修复
修复管道	每次调用 LLM 前自动修复，调用方无需关心	每次调用都有额外开销、可能掩盖上游 bug

Goose 的选择：修复管道。在 prepare_reply_context() 阶段统一修复，无论上游操作如何破坏消息结构，最终传递给 LLM 的消息总是格式正确的。这遵循”防御性编程”原则——在边界处保证不变量。

---

参考来源

• [[agent-design/01-overview]] — Agent 核心循环设计
• [[streaming/01-streaming-architecture]] — 流式传输架构
• Goose 源码：crates/goose/src/agents/agent.rs（reply() / reply_internal()）
• Goose 源码：crates/goose/src/conversation/（fix_conversation / debug_conversation_fix）
• Goose 源码：crates/goose/src/context_mgmt/（compact_messages / check_if_compaction_needed）