Agent 工作流模式与工程实践

LangGraph 给了你把 Agent 建模成图的能力，但「图长什么样」仍然需要你设计。

LangChain 文档总结了五种常见的工作流模式，覆盖了从简单流水线到复杂多 Agent 协作的大部分场景。选对模式，比堆更多 prompt 有效得多。

---

Agent 的核心等式

在聊模式之前，先对齐 Agent 是什么：

Agent = LLM + Tools + Memory + Planning

• LLM 是大脑
• Tools 是手脚——描述写得好不好，直接决定 Agent 能不能正确行动
• Memory 提供记忆力
• Planning 决定逻辑能力

Agent loop 本质上就是一个 while 循环：推理 → 调工具 → 写回上下文 → 再推理，直到达到目标或停止。

Agent 面临的现实问题：

1. 记忆 — 上下文窗口有限，长任务容易忘
2. 烧 token — 每轮循环都在消耗
3. 推理路径易失效 — 走偏了就是资源浪费

---

五种工作流模式

1. Prompt Chaining（提示链）

每步在前一步输出上增强，类似建造者模式。每步职责不同。

例子：生成小说 → 检查剧情连贯 → 检查人物关系 → 检查格式。

适合步骤固定、每步可独立验证的任务。路径 predetermined，不需要 LLM 动态决策。

LangGraph 里用 add_edge 串起来就行——上一步的输出作为下一步的输入，State 里的字段逐步丰富。

2. Parallelization（并行化）

两个用途：

• 任务拆解并行 — 把大任务拆成子任务同时跑，加快速度
• 多次运行增强可靠性 — 同一任务跑多次取最优

例子：

• 系统 bug 检查：权限控制、并发安全、数据一致性并行排查
• 性能测试：不同测试集同时跑，评估框架表现

和 Send API 配合就是 LangGraph 里的 Map-Reduce。同一个 Super-step 里，没有依赖关系的节点会自动并行执行。

3. Routing（路由）

即意图识别——根据输入决定走哪条路径。

例子：用户问「帮我查订单」→ 订单查询 Agent；问「系统挂了怎么办」→ 故障诊断 Agent。

Routing 关注的是识别，不是任务分解。和 Orchestrator-worker 容易混淆，但目的不同——Routing 是「你是谁的问题」，Orchestrator 是「这个问题怎么拆」。

条件入口的典型写法：

def route_by_intent(state) -> Literal["order_agent", "sre_agent"]:
    intent = classify(state["messages"][-1].content)
    return "order_agent" if intent == "order" else "sre_agent"

graph.add_conditional_edges(START, route_by_intent)

4. Orchestrator-Worker（编排者-工作者）

一个主脑分派任务给子系统：

• 将任务分解为子任务
• 委托给 Worker 执行
• 汇总 Worker 输出为最终结果

和 Parallelization 的区别：并行化的路径是固定的，Orchestrator-worker 的路径由主脑动态生成——多了一个前置节点来规划和分发。

例子：SRE Agent 里 Supervisor 拆分故障主题 → 多个 RCA Worker 并行诊断 → 汇总报告。

5. Evaluator-Optimizer（评估-优化）

迭代系统：一个 LLM 生成，另一个 LLM 评估，不通过就反复迭代。

适合质量要求高、可以客观评估的场景，比如代码生成、文档撰写。

---

模式选型速查

你的需求	推荐模式
固定步骤流水线	Prompt Chaining
同一任务要多角度/多次跑	Parallelization
根据输入类型分流	Routing
复杂任务要拆分派活	Orchestrator-Worker
输出质量要迭代打磨	Evaluator-Optimizer
LLM 自己决定下一步	动态 Agent（ReAct 循环）

实际项目经常是混合使用——比如 Orchestrator-Worker 做顶层调度，每个 Worker 内部是 ReAct 循环，最后 Evaluator 做质量把关。

---

Agent 的思考范式

除了工作流模式，Agent 的「思考逻辑」也有几种经典范式：

ReAct — 根据反馈执行下一步。灵活，适合动态环境。

Plan-and-Execute — LLM 先生成全局分步计划，再逐步执行。适合长期任务，但动态调整和容错不如 ReAct。

最佳组合：Plan-and-Execute 生成全局任务，执行阶段用 ReAct 增加动态调整能力。

Reflection — 在关键节点加反思：

• 「真的吗？」— Reflexion 框架，失败后反思避免重复错误
• 「有改进方法吗？」— Self-Refine，任务完成后自我优化
• 「结果正常吗？」— CRITIC，引入外部工具验证（比如 AI 说某人做了某事，先用 search 工具核实）

Multi-Agent — 两种组织形式：

• 主从模式（Orchestrator-Subagent） — 一个 Agent 全局规划分发，子 Agent 执行
• 并行模式（P2P） — 多个 Agent 对等协作，典型如 Grok 圆桌会议

实际项目里，Plan-and-Execute + ReAct 的组合很常见：先用 Plan 生成全局任务清单（适合长期任务），执行每个子任务时用 ReAct 增加动态调整（适合应对意外）。Reflection 则在关键节点加质量关卡——不是每一步都反思，只在「输出给用户之前」和「工具调用失败之后」加，性价比最高。

---

ToolNode：框架帮你包装的工具节点

LangGraph 提供了 ToolNode 类来包装工具调用节点，内置：

• 并发控制 — 多个 tool_calls 可以并行执行
• 失败重试 — 工具调用失败自动重试

比自己手写 tool_node 更健壮，生产环境推荐直接用。

---

工程兜底：JSON 输出清洗

DeepSeek 等模型的 JSON 约束容易出问题——没有 strict 模式约束字段，只有 bind(response_format={"type": "json_object"}) 告诉模型用 JSON 返回，但有型无神，字段经常乱。

一套三层兜底流程：

第一层：语法兜底

AI 输出多是 markdown 格式，先剥掉代码块外壳，提取最外层 {...}：

import json
import re

def extract_json_object(text: str) -> str:
    text = strip_code_fence(text)
    start = text.find("{")
    end = text.rfind("}")
    if start != -1 and end != -1 and end > start:
        return text[start : end + 1]
    return text

第二层：结构兜底

JSON 解析失败时，把错误上下文追加到 message 里，让 AI 重新生成：

def call_with_json_retry(system_prompt, human_content, schema, max_retries=2):
    messages = [SystemMessage(system_prompt), HumanMessage(human_content)]
    last_err = None
    for attempt in range(max_retries + 1):
        raw = json_llm.invoke(messages)
        try:
            cleaned = extract_json(raw.content)
            return schema.model_validate_json(cleaned)
        except (json.JSONDecodeError, ValidationError) as e:
            last_err = e
            messages.append(AIMessage(raw.content))
            messages.append(HumanMessage(
                f"上面的输出不是合法 JSON，错误：{e}\n"
                f"请严格按照示例格式重新输出，只输出 JSON。"
            ))
    raise last_err

第三层：业务兜底

所有重试仍失败时，不能让整个流程崩掉——给一个安全默认值，让 Agent 继续跑。

---

Tools、MCP 与 Skills

Tools 在 Agent 里注册给大模型时，统一用 JSON Schema 描述自己。分两类：

• Function Call — 本地函数，直接注册
• MCP（Model Context Protocol） — 对 LLM 来说类似一套 TCP 协议，Agent 启动时向 MCP 服务请求函数列表，后续携带进上下文

MCP 和本地 Function Call 的区别在于解耦——工具逻辑在独立服务里，Agent 只管调用。

Skills 是给 AI 用的经验包——不是传统编程的 if-else，而是自然语言编码的 workflow。灵活的地方不在 skill 本身，而在 Agent 选择使用哪个 skill 的判断。

Skill 的具体内容就是 workflow，只是用自然语言编码。实际并没有多自由——自由的是 Agent 选择使用哪个 skill。还有检查类、规范类、报告模板类的 skill，各自服务不同环节。

MCP 对 LLM 来说类似一套 TCP 协议：Agent 启动时向配置中的 MCP 服务发请求，收到函数 JSON 和说明，后续携带进上下文即可调用。和本地 Function Call 比，MCP 实现了解耦——工具逻辑在独立进程里，Agent 只管调用接口。

---