5 个即拿即用的 Agent 工作流

副标题：Agent 与上下文工程 · AI 提效

本文为 vergex.cn 原创首发，转载请注明出处。

“周五凌晨 2 点，值班机器人收到告警后自动拉起 Agent：分析日志、改配置、重启服务。十分钟后你被人叫醒——Agent 在 17 轮工具调用后把生产配置改回了三年前的版本，而你完全不知道它中间经历了什么。”

工作流 1：把 ReAct 从“自由作文”改成“填空题”

传统 ReAct 的 prompt 让模型自由写 Thought / Action / Observation，看似优雅，实则是开放命题：Action 可能带错参数，Thought 可能偷偷改目标，Observation 还可能被模型自己编造。在生产环境里，LLM 只应负责“想什么”和“调哪个工具”，不应负责状态结构和日志格式。

我的做法是把 ReAct 拆成三步固定的状态机：模型输出严格 JSON（字段只有 thought、tool_name、tool_args）；执行器负责调用工具并生成 observation；状态对象维护 turn_count、tool_results、final_answer。这样即便模型幻觉，也最多影响字段值，不会破坏协议。

LangGraph、PydanticAI 等框架现在支持“结构化输出 + 状态图”，但核心不是框架，而是把不确定的生成关在确定性的 schema 里。我要求团队所有 Agent 的“思考”必须能用 Pydantic BaseModel 校验，否则不准上线。

实现时推荐用 OpenAI function calling，让模型每轮必须调用同一个函数输出决策。注意 tool_choice 要强制指定该函数，否则模型可能直接回复用户而绕过状态机。下面的代码可以直接运行，真实工具替换占位函数即可：

import json
import os
from openai import OpenAI
from pydantic import BaseModel, Field

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

class AgentStep(BaseModel):
    thought: str
    tool_name: str | None = Field(
        description="可选 search / calc，无需工具时填 null"
    )
    tool_args: dict | None = Field(
        description="工具参数对象，无需工具时填 null"
    )

tools = [
    {
        "type": "function",
        "function": {
            "name": "think_and_act",
            "description": "输出本轮思考与计划调用的工具",
            "parameters": AgentStep.model_json_schema(),
        },
    }
]

SYSTEM = (
    "你每次必须调用 think_and_act 函数输出思考与工具选择，"
    "不要直接回复用户。"
)

def run_react(query: str, max_turns: int = 5):
    messages = [
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": query},
    ]
    for turn in range(1, max_turns + 1):
        resp = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice={
                "type": "function",
                "function": {"name": "think_and_act"},
            },
        )
        msg = resp.choices[0].message
        messages.append({
            "role": "assistant",
            "content": msg.content or "",
            "tool_calls": [tc.model_dump() for tc in msg.tool_calls],
        })

        if not msg.tool_calls:
            return msg.content

        args = json.loads(msg.tool_calls[0].function.arguments)
        step = AgentStep.model_validate(args)

        if step.tool_name is None:
            return step.thought

        observation = f"调用 {step.tool_name}({step.tool_args}) 的结果占位"
        messages.append({
            "role": "tool",
            "tool_call_id": msg.tool_calls[0].id,
            "content": observation,
        })

    return "达到最大轮数，未得出最终结论"

if __name__ == "__main__":
    print(run_react("2024 年北京 GDP 是多少？"))

这套结构还有一个额外好处：可观测性。因为每轮决策都是结构化对象，你可以把它直接写入日志或 tracing 系统，后续做审计、回滚、评估都很方便。完整版工作流模板集已整理在 VergeX 站点，包含 schema 与状态转换图。

工作流 2：双层路由 —— 先分意图，再批权限

把用户请求直接丢给 Agent 是产品事故的开端。Agent 拥有写库、发邮件、调第三方 API 的权限，一旦遇到恶意 prompt、越狱或敏感操作，后果不堪设想。路由层的职责不是“理解”，而是“把关”。

我采用双层路由。第一层是轻量分类器，用 tool_choice="none" 强制模型只输出意图标签，如 read、write、admin、unsafe。标签为 unsafe 直接拒绝并记录；write/admin 进入第二层安全审批。第二层不依赖大模型，而是对照规则表：操作对象、影响范围、必需字段、人工确认阈值。两条规则都满足才进入执行态。

关键细节是第一层模型必须小、快、只输出标签。不要用同一个大模型既做推理又做安全判断，否则它在角色冲突时会“和稀泥”。我一般用 gpt-4o-mini，温度设为 0，最多返回 3 个 token。合规要求高的场景，用确定性分类器（关键词 + 正则）兜底，模型输出仅作辅助。

另一个少有人提的点：意图标签必须和后续工具集一一对应。read 只绑定读工具，admin 才开放写工具。权限是绑在工作流入口的，不是模型自己“决定”的。让模型在运行时选工具，看似灵活，实则是把访问控制外包给了概率模型。

第一层输出一定要记录到审计日志，包括原始输入、标签、模型版本、时间戳。事后复盘时，这比任何 prompt 优化都重要。

工作流 3：RAG 不是一次检索，而是“检索-验证-再检索”

很多团队把 RAG 做成“提问 → 向量检索 Top-3 → 拼接 prompt → 出答案”。简单问答够用，一旦涉及跨文档推理、时间线对比、隐含条件查询，就会丢信息。真正靠谱的 RAG Agent 会在检索后做一次“自我审查”。

我的做法是把 RAG 拆成三态循环：检索（Retrieve）、评估（Evaluate）、补充（Refine）。首轮检索拿到候选文档后，不直接回答，而是让模型判断：这些片段是否覆盖问题的全部维度？置信度多少？是否缺少时间、主体、数值等关键信息？如果置信度低或维度缺失，模型输出新的子查询，进入下一轮检索，最多循环 3 次。

评估模型返回结构化结果：coverage_score（0-1）、missing_aspects（列表）、sub_queries（列表）。注意 coverage_score 不要直接当阈值，因为模型打分常偏高；我把它当作相对排序信号，真正的停止条件是 missing_aspects 为空且子查询不重复。若子查询和前一轮重复，说明检索已饱和，强制退出。

补充查询生成时，我会把上一轮已检索到的关键句一并传给模型，避免它重复提出已经回答的问题。每次循环结果写入状态，但不要把原始文档全文一直保留在上下文中。只保留“摘要 + 来源 + 关键句子”，原始内容存 side state，生成阶段按需再取。

VergeX 站点上的完整版模板集把这个循环做成了可配置节点，包含最大循环数、重排序模型和置信度校准策略。

工作流 4：工具返回的“脱水-再水化”

Agent 调用数据库、API 或代码解释器后，返回结果动辄几千 token。直接塞进对话历史，几轮后上下文就爆，模型开始“遗忘”最初的问题。这个坑几乎所有人都会踩，却很少被系统性地写出来。

我引入“脱水（Dehydrate）”和“再水化（Rehydrate）”。脱水：工具返回后，用 summarizer 模型把它压缩成固定 schema，如 {source, key_findings, relevant_rows, caveats}；原始结果存在 raw_tool_outputs，不进入 LLM 上下文。再水化：后续某步需要原始细节时（如用户追问“第 3 条记录的具体时间”），从 side state 调出相关片段。

对于表格类数据，脱水的摘要可以包括行数、列名、极值、异常值提示；真正需要细节时再按主键取原行。你可以用 tiktoken 或各厂商 tokenizer 测量前后 token 数。不要拍脑袋说“压缩了 80%”，而是把测量脚本放进 CI。我的标准是：进入下一轮决策的上下文里，单个工具结果不超过 400 token；超过就必须拆分或多级摘要。

下面是对比两种处理方式的表格：

不要试图用一个 prompt 让模型“自己记得少写点”。上下文管理必须外化到状态系统里，否则模型会一边删你要求删的内容，一边复述无关细节。

工作流 5：多个专家 Agent 之间的受控交接

业务复杂到单个 Agent 兼顾不了时，大家本能地想拆专家 Agent：财务、法务、客服。但如果没有受控交接协议，它们会互相踢皮球、重复提问、甚至循环调用。多 Agent 不是“把 prompt 拆成几份”，而是设计一个状态转移层。

我的方案：所有专家 Agent 注册到同一个 dispatcher，每个暴露 entry_topics、required_context、output_schema。当某个 Agent 判断任务超出自身范围，输出 handoff_to 字段，并附带 summary_so_far 和 open_questions。dispatcher 收到后做两件事：检查 handoff_count 是否超限；根据 entry_topics 和摘要选择下一 Agent。

交接时遵守两条规则。第一，上下文必须压缩：不能把完整对话历史传给下一个 Agent，只传摘要 + 必要实体 + 待解决问题。第二，必须防止回弹：A 交给 B 后，B 的候选目标不能再是 A，除非用户明确产生新分支。我在状态中维护 visited_agents 集合，一旦回环就触发人工兜底。

dispatcher 最好还有一个全局超时：不论怎么交接，单次用户请求的总耗时不能超过业务 SLA，否则直接转人工。实现上不需要 fancy 框架，一个字典注册表加 Pydantic 校验就够了。LangGraph 等框架的节点路由机制、OpenAI Agents SDK 的 handoff 都是这个思路的封装。核心原则是：交接决策权在工作流引擎，不在模型本身。模型只建议，不执行。

避坑 A：不要把 Agent 的“记忆”直接塞进长上下文

Agent 需要记忆，但“把过去所有对话 + 工具结果拼成 prompt”不是记忆，是上下文垃圾。我见过有团队把 30 轮对话直接喂给模型，结果响应质量反而比无记忆还差，因为关键信号被淹没在旧信息里。

正确的记忆分两层：工作记忆（working memory） 和 长期记忆（long-term memory）。工作记忆只保留当前任务必须的上下文，比如本轮目标、已确认实体、已执行操作、待验证假设，控制在 2k token 以内。长期记忆用向量库或知识图谱存储，按需检索；每次进入新任务时，先查长期记忆，生成“角色/用户背景摘要”注入系统提示。

我会把长期记忆再分成两类：一类是事实型（用户偏好、项目背景），一类是过程型（某次具体工单）。事实型适合向量检索，过程型适合按时间线索引。不要在系统提示里塞用户全部历史记录。你应该只注入与当前意图最相关的 3-5 条摘要，并告诉模型“下面是可能有用的背景，不需要全部使用”。记忆检索的质量，比记忆本身的容量更重要。检索不准，再大的上下文也是噪声。

另一个细节：工具调用产生的中间结果不要默认入长期记忆。只有经过用户确认或标注为关键结论的片段才值得保存。否则你的“记忆库”会变成操作日志，检索命中率急剧下降。

避坑 B：不要迷信“自动重试”—— 失败必须有状态机

“出错了就让模型再试一次”是最危险的偷懒。没有状态机的重试，会让 Agent 在失败后继续用同一参数调用同一工具，或者把失败结果当成成功结果继续推理。更糟的是，重试过程中可能触发副作用：重复发邮件、重复扣款。

我给每个 Agent 工作流配显式状态机：状态包括 idle、awaiting_tool、tool_failed、needs_clarification、completed、escalated。每次工具调用返回后，执行器负责推进状态，而不是让模型决定。重试只在 tool_failed 状态发生，且必须满足：错误是可重试类型（如超时、限流）；已尝试次数小于上限；重试前更新参数（比如加入 jitter 或 fallback token）。

非重试类错误，比如 400 Bad Request、权限不足、数据校验失败，应该直接进入 needs_clarification 或 escalated，而不是让模型反复尝试。重试时必须带上幂等键。对外部 API 使用 idempotency_key，对数据库写操作使用业务唯一键。这样即使 Agent 重试 3 次，也不会产生重复数据。不要把希望寄托在模型“懂事”上——模型不保证幂等，状态机和幂等键才保证。

一个可落地的命令：在日志里给每次工具调用打上 run_id 和 attempt，并用 jq 快速定位失败链路：

cat agent.log | jq -c 'select(.level=="ERROR") | {run_id, tool, attempt, error}'

这条命令依赖日志是 JSON Lines 格式。如果你还在用纯文本日志，建议先切到结构化日志，否则排查 Agent 故障会是一场灾难。

反模式清单与一句总结

落地 Agent 工作流前，我会让团队先对照这份清单自查：

• [ ] 让模型自己写状态字段名，导致每轮输出格式不可解析。
• [ ] 把完整工具返回结果直接拼进对话历史，不做摘要或 side state。
• [ ] 权限控制放在 prompt 里，靠模型自觉选择工具。
• [ ] 用同一个大模型同时做分类、推理、生成、安全判断。
• [ ] RAG 只做一次检索，不做缺失维度检查和补充查询。
• [ ] 长期记忆 = 完整对话日志，检索时全量注入上下文。
• [ ] 出错后无状态地让模型“再试一次”，缺少幂等键与状态机。
• [ ] 多 Agent 交接时把全部历史带过去，导致循环踢皮球。

完整的工作流模板、状态机图和测量脚本可在 VergeX 站点下载。

最后一句核心总结：Agent 工作流的质量，不取决于模型有多强，而取决于你把多少不确定性关进了确定性的结构与状态机里。