前面几篇文章讲了 Function Calling、Agent 设计、MCP 协议、RAG——这文章把所有这些技术组合起来，构建一个完整的 Agent Workflow 系统。
传统应用调用是线性的：请求 → 处理 → 响应。Agent Workflow 不一样——它是一个有分支、有循环、有状态管理的编排过程。
这个智能助手能做什么：
能联网搜索（搜索 API）
能查本地文档（RAG）
能记住上下文（短期记忆 + 长期记忆）
能执行简单操作（发邮件、创建工单）
能多轮对话（在对话中修正和追问）
Workflow 总体架构
全栈智能助手的 Workflow 由四个阶段组成，每个阶段内部又有多个步骤：
用户消息
│
▼
┌────────────────────────────────────────────────────────────┐
│ 阶段 1：路由（Route） │
│ │
│ 用户输入 → IntentRouter → 分支决策 │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ search query action chat ambiguous │
│ (搜索) (知识库) (操作) (闲聊) (需澄清) │
│ │
│ → 输出：路由结果 + 工具列表 │
└────────────────────────────────────────────────────────────┘
│
▼
┌────────────────────────────────────────────────────────────┐
│ 阶段 2：编排（Orchestrate） │
│ │
│ Agent Loop（最多 10 轮）： │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ 第 N 轮： │ │
│ │ ① LLM 推理：分析上下文，决定调用工具还是直接回答 │ │
│ │ ② 如果是工具调用 → 批量执行 → 结果反馈 → 继续循环 │ │
│ │ ③ 如果是直接回答 → 结束循环 │ │
│ └──────────────────────────────────────────────────────┘ │
│ │
│ → 输出：执行结果链 │
└────────────────────────────────────────────────────────────┘
│
▼
┌────────────────────────────────────────────────────────────┐
│ 阶段 3：记忆（Memory） │
│ │
│ 短期记忆： 长期记忆： │
│ ├─ 追加本轮交互 ├─ 检测是否需要检索│
│ ├─ 检查是否溢出 ├─ 语义检索相关历史 │
│ ├─ 溢出→压缩早期轮次为摘要 └─ 若需要→注入上下文 │
│ └─ 保留系统提示 + 最近 N 轮 │
│ │
│ → 输出：更新后的上下文 │
└────────────────────────────────────────────────────────────┘
│
▼
┌────────────────────────────────────────────────────────────┐
│ 阶段 4：安全（Safety） │
│ │
│ ├─ 高风险操作检测（删除/修改/发信→需二次确认） │
│ ├─ 输出内容过滤（PII 检测、有害内容拦截） │
│ └─ 审计日志持久化 │
│ │
│ → 输出：最终回复 │
└────────────────────────────────────────────────────────────┘
│
▼
最终回复
核心设计原则：路由层做分支，编排层做循环，记忆层做状态，安全层做约束。 每层关注一件事，互不干扰。
阶段一：路由（Route）——Workflow 的第一个分支
路由是整个 Workflow 的入口决策点。不是所有的用户消息都需要进入 Agent Loop——有些可以直接回答，有些需要检索，有些需要反问。
路由 Workflow：
┌─────────────┐
│ 用户输入 │
└──────┬──────┘
│
┌──────▼──────┐
│ IntentRouter│
│ (LLM 分类) │
└──────┬──────┘
│
┌────────────────┼────────────────┐────────────────┐
▼ ▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐
│ search │ │ query │ │ action │ │ ambiguous │
│ 暴露搜索 │ │ 暴露 RAG │ │ 暴露操作 │ │ 反问用户澄清 │
│ 工具 │ │ 工具 │ │ 工具 │ │ │
└────┬─────┘ └────┬─────┘ └────┬─────┘ └──────────────┘
│ │ │ │
└───────────────┼───────────────┘ │
▼ │
┌──────────────┐ │
│ 进入编排阶段 │ ←──────── 直到明确意图 ────┘
└──────────────┘
路由的好处不仅是"减少不必要的工具注册"，更重要的是控制了 Workflow 的分支路径——搜索路径只暴露搜索工具，工具列表短、LLM 选择准确率高；操作路径暴露写操作工具，并预先加载安全门控。
歧义处理是路由中关键的一个子 Workflow：
ambiguous 子 Workflow：
① LLM 识别出用户意图不明确
② 生成候选意图列表（2-3 个）
③ 反问用户选择
④ 用户确认 → 重新路由
⑤ 如果超过 2 次反问都不明确 → 转人工
这个子 Workflow 防止了 Agent 在意图不明确时"瞎猜"。
阶段二：编排（Orchestrate）——Workflow 的核心循环
编排阶段是整个 Agent Workflow 的心脏。它本质上是一个带反馈的 while 循环：
Agent Loop Workflow：
┌─────────────┐
│ 进入编排阶段 │
│ messages: │
│ [系统提示+ │
│ 用户问题+ │
│ 历史] │
└──────┬──────┘
▼
┌─────────────────────┐
│ round = 0 │
└─────────┬───────────┘
▼
┌─────────────────────┐
│ round < max_rounds? │──→ 否 → 超时退出
└─────────┬───────────┘
│ 是
▼
┌─────────────────────┐
│ LLM.chat(messages, │
│ tools) │
└─────────┬───────────┘
▼
┌─────────────────────┐
│ 有 tool_calls? │
│ ↓ ↓ │
│ 是 否 │
└────┬────────┬───────┘
│ │
▼ └──→ 返回 text，结束
┌────────────────────┐
│ 批量执行工具调用 │
│ 并行：互相独立 │
│ 串行：有依赖关系 │
└─────────┬──────────┘
▼
┌────────────────────┐
│ 结果追加到 messages │
│ (tool role) │
└─────────┬──────────┘
▼
┌────────────┐
│ round++ │──→ 回到循环检查
└────────────┘
核心代码骨架：
class AgentWorkflow:
def __init__(self):
self.llm = LLMProvider("claude-sonnet-4")
self.tools = ToolRegistry()
self.max_rounds = 10
async def orchestrate(self, messages: list) -> str:
"""编排阶段：Agent 主循环"""
for round in range(self.max_rounds):
response = await self.llm.chat(
messages=messages,
tools=self.tools.list()
)
if not response.tool_calls:
# 没有工具调用 → 返回最终回复，结束循环
return response.text
# 批量执行工具调用
# 步骤 A：检查工具之间是否有数据依赖
groups = self._group_independent_calls(response.tool_calls)
for group in groups:
# 同组内的工具互相独立，可以并行
results = await asyncio.gather(*[
self._execute_tool(call) for call in group
])
for call, result in zip(group, results):
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": result
})
# 有依赖的组需要串行，等上一组结果
# 下一轮 LLM 会看到上一组结果再决定后续调用
return "我已经思考了足够多轮，暂时无法完成这个请求。"
def _group_independent_calls(self, calls: list) -> list:
"""检查工具调用之间的依赖关系，分组执行"""
# 简单实现：名字相同的工具之间可能有依赖
# 实际中可以用 DAG 表示依赖关系
return [[c] for c in calls] # 默认串行
编排阶段最重要的设计决策是什么时候并行、什么时候串行：
并行场景（工具之间无数据依赖）：
工具 A：search_weather("北京")
工具 B：search_news("AI")
→ 同时执行，各等各的结果
串行场景（工具之间有数据依赖）：
工具 A：get_contact("张三") → 返回 email
工具 B：send_email(email, ...)
→ 必须等 A 返回结果后才能调用 B
Workflow 引擎应该自动检测依赖关系，而不是让 LLM 去决策"要不要并行"。LLM 只需要输出该调什么工具，执行顺序是编排层的职责。
阶段三：记忆（Memory）——Workflow 的状态管理
Workflow 的每一轮交互都在改变状态。状态管理不好，上一轮做过的事下一轮就忘了。
短期记忆的管理是一个**"溢出 → 压缩"的 Workflow**：
短期记忆 Workflow：
┌─────────────┐
│ 新的消息加入 │
│ messages[] │
└──────┬──────┘
▼
┌──────────────┐
│ count_tokens()│
│ ≤ max_tokens? │
│ ↓ ↓ │
│ 是 否 │
└────┬─────────┘
│
▼ 否
┌──────────────────────┐
│ 压缩策略（逐级降级）： │
│ │
│ Level 1：移除最早工具 │
│ 调用的完整结果 │
│ → 压缩为摘要 │
│ │
│ Level 2：如果还不够 │
│ 压缩最早的一轮对话 │
│ 为 1 句话摘要 │
│ │
│ Level 3：如果还不够 │
│ 丢弃最早的非系统消息 │
└──────────────────────┘
│
▼
┌──────────────┐
│ 继续 Workflow │
└──────────────┘
长期记忆的检索也是一个子 Workflow：
长期记忆检索 Workflow：
┌─────────────┐
│ 用户新消息 │
└──────┬──────┘
▼
┌─────────────────┐
│ 需要检索历史？ │
│ 关键词匹配或 │
│ LLM 快速判断 │
│ ↓ ↓ │
│ 需要 不需要 │
└────┬────────────┘
│
▼ 需要
┌────────────────────┐
│ Embedding 检索 │
│ 向量数据库 Top-5 │
└─────────┬──────────┘
▼
┌────────────────────┐
│ 相关性重排（Rerank） │
│ 取 Top-2 最相关的 │
└─────────┬──────────┘
▼
┌────────────────────┐
│ 注入 System Prompt │
│ "以下为对话历史摘要"│
└────────────────────┘
这个子 Workflow 的关键是第一步的判断——不是每次都需要检索。加上关键词触发和 LLM 快速判断的双保险，避免不必要的检索。
阶段四：安全（Safety）——Workflow 的守卫层
安全层拦截那些"不应该被执行的操作"和"不应该被输出的内容"。
安全 Workflow：
┌─────────────┐
│ Agent 输出 │
└──────┬──────┘
▼
┌────────────────────────┐
│ 高风险操作检测 │
│ (发送、删除、修改、审批) │
│ ↓ ↓ │
│ 高风险 低风险 │
└────┬──────────────────┘
│
▼ 高风险
┌────────────────────────┐
│ 请求用户二次确认 │
│ "您确认要发送这封邮件？" │
│ ↓ ↓ │
│ 确认 取消 │
└────┬──────────────────┘
│ │
│ 确认 ▼ 取消
▼ ┌────────────┐
┌────────────┐ │ 取消操作 │
│ 执行操作 │ │ 告知用户 │
└─────┬──────┘ └────────────┘
▼
┌────────────────────────┐
│ 输出内容过滤 │
│ ├─ PII 检测（身份证等） │
│ ├─ 敏感内容检测 │
│ └─ 格式校验 │
└─────┬──────────────────┘
▼
┌────────────────────────┐
│ 审计日志持久化 │
│ (完整 trace 写入数据库) │
└────────────────────────┘
安全层是 Workflow 中最"传统"的部分——它不需要 LLM，全部是确定性的校验逻辑。但恰恰是这部分，决定了 Workflow 能不能上生产。
完整的 Workflow 追踪（Trace）
每一个请求经过 Workflow 的完整路径都应该被记录下来，用于调试和复盘：
Request ID: req_fullstack_001
Phase 1 (Route):
- Intent: "query"
- Confidence: 0.92
- Tools loaded: [search_kb, web_search]
- Duration: 0.3s
Phase 2 (Orchestrate):
- Round 1:
+ LLM call: input 1520t → output 45t (tool_call: search_kb)
+ Tool: search_kb("Redisson 分布式锁", 3) → 3 docs
+ Duration: 1.5s
- Round 2:
+ LLM call: input 2100t → output 256t (final answer)
+ Duration: 1.8s
- Total rounds: 2
Phase 3 (Memory):
- Short-term: saved (5120 tokens)
- Long-term: saved new memory entry
Phase 4 (Safety):
- Risk check: low risk, no confirmation needed
- Content filter: passed
- Audit: logged
Total Duration: 4.1s
Total Cost: $0.0042
有了这样的 trace，调试和优化 Workflow 就不是靠猜了——可以精确定位哪个阶段、哪一步花的时间长或出错。
Workflow 设计原则
从这套系统里总结几条 Workflow 设计原则：
1. 每个阶段只做一件事
路由只做路由，编排只做循环，记忆只做状态管理，安全只做安全。如果一个阶段既要做路由又要做安全，逻辑耦合后改一个功能会影响另一个。
2. 分支越早越好
路由是 Workflow 中最靠前的分支点。越早做分支决策，后面的路径就越简单、工具列表越短、LLM 调用越便宜。
3. 循环要有上限
Agent Loop 必须设上限（10 轮），设超时（30s），设熔断机制。没有上限的循环就像没有 break 的 while 循环——出 bug 时无限调用 API，成本几分钟内就能追上来。
4. 状态必须可恢复
如果 Agent 在第五轮崩溃了，重启后应该能从上下文中恢复状态。这意味着每轮交互后的消息列表需要持久化（短期记忆存到 KV 或 DB），不能只靠内存。
5. 安全是 Workflow 的一等公民
安全层不是"最后加上的"——它从一开始就是 Workflow 的一个独立阶段。没有安全层的 Workflow 不能进入生产。
总结
全栈智能助手 = 路由 Workflow → 编排 Workflow → 记忆 Workflow → 安全 Workflow。
每个 Workflow 内部有多个步骤，Workflow 之间有清晰的输入输出边界。这种设计让系统的每一层都可以独立测试、独立升级、独立降级。
去掉不必要的复杂性，从最简 Workflow 开始——路由 → 编排 → 输出。能力不够时再加记忆，再不够时加安全。Workflow 是应需而生，不是预先设计的天花板。
评论