当前位置：首页 > news >正文

给 AI 的 Agent 实现指南，可控 Agent 的关键

news 2026/7/30 0:04:34

好久没聊 Agent 了，今天继续聊聊 Agent 的关键实现要点，事实上一个 Agent 在日常生产里最主要的问题就是可控和可靠，所以虽然理论上 Agent 就是 LLM + Tools + RAG 的简单组合，但是实际上不同场景有很多基础细节需要处理，比如：

能不能正确理解了用户目标？
为什么选择这个工具 A 而不是 B ？
参数缺失时会追问还是重试？
工具失败后会修复和降级吗？
删除、退款、发邮件、改配置之类的动作，可不可以被权限和确认机制拦住？
长任务中断后怎么从检查点恢复？
记忆会不会过期、冲突和被污染？
系统出了问题，怎么定位是检索、规划、参数、工具还是模型生成的问题？

这个些问题其实我们也聊过很多次，基本就是一个完整的 Agent 至少包含六类能力：

目标理解、任务规划、工具使用、状态与记忆、执行反馈、安全与评测。

AI 模型确实是 Agent 里关键的决策组件，但是决定系统是否可控和可靠的，基本都是外部的工程边界。

举个例子，比如现在要做一个「合同助手」的 AI Agent ，那么这时候用户可能提出：

找出上周老板发来的合同，查看付款条款，整理成一段话，准备发给财务。

这种场景需要的就是一整条任务链，比如：

理解“上周”“老板”“那个合同”这些上下文实体
搜索邮件和识别候选附件
读取和解析对应合同
检索付款条款
对条款进行完整和贴合原始内容的总结
生成发给财务的草稿
在真正发送前请求用户确认
然后进入邮件和发送

那你觉得这个任务里应该怎么做合适？一般来说整个「合同助手」的 Agent flow 应该类似这样，有动态规划的 Agent ，也有固定流程 workflow 进行组合：

实际上这样是生产环境最常见的混合实现：

Workflow 决定可执行的轨道，Agent 负责轨道内的语义判断和动态选择。

然后前面的图我们从外部看从内部看，它可以被拆成个多个层级：

入口层：身份、租户、会话、请求标准化
路由层：识别意图、风险和任务类型
编排层：维护状态、预算、检查点和下一步
规划层：拆任务、选择工具、判断缺失参数
上下文层：装配当前状态、知识、记忆和工具说；
执行层：校验、授权、确认、幂等、调用与熔断
验证层：判断工具结果是否可信、任务是否完成
输出层：生成带证据、可解释、符合格式的最终结果
横切能力：安全、审计、评测、可观测、成本治理

基本上一个最普通的 Agent 也都会具备这样的层级能力，因为要生产可靠和可控，Agent Loop 就不能只是「思考—行动—观察」这样简单的 Loop ，业务上肯定是需要：

理解目标 → 读取状态和上下文 → 生成候选计划 → 选择受限工具 → 参数与权限校验 → 必要时请求确认 → 执行工具 → 归一化结果 → 验证是否满足完成条件 → 未完成则重新规划，完成则输出

然后在这条链路里，每一步都可能出现失败，所以你需要把错误当成正常分支处理，保证每一次至少有结果反馈。

一、意图识别

这个其实一直是老生常谈的东西，用户问的问题，比如前面的「找出上周老板发来的合同，查看付款条款，整理成一段话，准备发给财务和。」，你肯定是不能直接就发给模型处理。

Agent 要做的是先进行意图识别，意图识别的目标是在准确率、延迟、成本和风险之间做取舍和分类，比如下面就是一个常见的意图识别分类场景：

也就是你需要根据你的业务看是需要怎么分成，而「合同助手」这种 Agent 就很明显存在确定性规则：

第一层可以做确定性规则，用来适合处理一些高频、明确、低歧义的指令，比如例如：
- 退出登录
- 转人工
- 重新生成
- 取消当前步骤
- 打开设置

规则层必须明确，并且精简，主要目的是可以快速拦截最确定的一批请求，如果规太多，反而会导致变成屎山，后面可能演化成更难维护的关键词和 if-else 集合。

第二层可以做上下文路由，因为很多真实表达脱离上下文会很难理解，比如
- “就这个吧”
- “还是不行”
- “改成 300”
- “再试一次”

所以这一层需要结合比如上一轮对话、当前页面和选中对象、当前任务节点之类的场景做路由，也就是可以用一些轻量化本地文本小模型配合历史信息，快速做语义的意图判断。

第三层就是复杂任务理解和工具规划，也就是复杂度太高，或者匹配不上的，最后才会进入大模型处理，而就算到了这一层，模型输出的也不能只是一个意图标签，我们需要的应该是一个结构化路由结果，例如：

{ "intent": "review_contract_and_prepare_email", "entities": { "sender_role": "manager", "time_range": "last_week", "document_type": "contract", "target_audience": "finance" }, "missing_fields": [], "risk_level": "medium", "confidence": 0.87, "next_action": "search_mail" }

而实际上在意图这里，一直有个置信度的存在，它的作用就是用来决定下一步怎么走，一般来说置信度至少能触发三种不同策略：

高置信度/低风险：直接路由
中等置信度：进入更强模型或补充上下文
低置信度/高风险歧义：向用户澄清

比如“取消订单”和“取消预约”这种意图，我们肯定不能靠模型去赌一个答案成功率是不是 90%，高风险操作肯定需要 human on loop 操作，就算匹配上了，也需要进入澄清阶段。

二、编排和状态机

Agent 知道用户意图之后，还必须知道自己做到哪一步，这个状态你不可能说直接就用历史会话来判断，因为聊天记录只有输入历史，不存在完整的任务状态，所以一个可靠的 Agent State 至少需要包含类似：

from typing import Any, Literal, TypedDict  class AgentState(TypedDict): task_id: str user_goal: str current_step: str status: Literal["running", "waiting_user", "completed", "failed"] collected_slots: dict[str, Any] selected_tools: list[str] observations: list[dict[str, Any]] evidence: list[dict[str, Any]] retry_count: dict[str, int] token_budget: int time_budget_ms: int pending_approval: dict[str, Any] | None

在这里，我们需要在状态里尽可能保存原始数据和结构化结果，这样才能可控，如果是一份 markdown 输出肯定是不行的，需要一份结构化的数据，才可以在不同节点复用同一份事实和状态。

比如状态最大的作用就是明确和定义完成条件，每个任务都必须有显式完成条件：

必需字段是否齐全
关键工具是否成功
证据是否覆盖问题
输出是否通过 Schema
是否存在未处理的高风险动作
是否超过步骤、时间、Token 或费用预算

比如「合同助手」的完成条件肯定需要包含：合同来源已确认、付款条款有没有原文证据、总结和证据必须一致、邮件生成草稿、发送是否已经执行。

然后还需要支持「检查点和持久化」能力，因为长任务过程肯定会存在暂停和恢复，比如：

等待用户补充订单号
等待审批
第三方接口暂时不可用
进程重启或节点故障
人工修改中间状态后继续运行

支持在 Agent 保存这些检查点也是很关键的需求，一般来说至少需要保存：状态版本、已完成节点、待执行节点、工具结果摘要、幂等键和审批信息。

然后最后就是「恢复和执行」的支持，这里最重要的就是「防止重复执行」，因为和 Coding Agent 不同，Coding 你重复编译和重复扫描问题不大，但是类似「合同助手」这种肯定不行，系统从检查点恢复时，可能某个节点已经运行过了，但是当时我们没等待结果，所以所有具备有副作用的操作都要考虑幂等：

发送邮件使用幂等键
数据更新使用条件更新或 upsert
支付和退款保存业务请求号
创建记录前先检查是否已存在
审批前不执行不可逆副作用

这部分其实就和 Agent 没直接关系，更多是业务后端的能力和状态同步设计。

三、工具系统

工具其实就是 Agent 里的业务抽象，也是一个 Agent 的价值和智力表现，但是定义工具也不是说就写一个 funciton call 就完事了，它本身也需要一套可靠流程支持：

一般来说，Agent 里的工具类似一份契约，你需要给工具定义和准备好一整套的定义，比如：

名称和版本
适用场景与禁用场景
输入 JSON Schema
输出 Schema
权限要求
风险等级
是否有副作用
需不需要支持幂等
超时、限流和重试策略
可能返回的标准错误码

比如一个工具定义可以类似：

from typing import Literal from pydantic import BaseModel, Field  class SearchMailArgs(BaseModel): sender_role: Literal["manager", "finance", "customer", "unknown"] start_date: str = Field(description="ISO-8601 日期") end_date: str = Field(description="ISO-8601 日期") attachment_type: Literal["contract", "invoice", "any"] = "contract" max_results: int = Field(default=20, ge=1, le=50)  class ToolProposal(BaseModel): tool_name: Literal["search_mail", "read_attachment", "extract_clause", "create_draft"] arguments: dict expected_result: str risk_level: Literal["read", "write", "high_risk"]

当然，在这里 Schema 只保证执行结果的结构，但是结果是否正确还需要后续判断，比如：日期可能格式合法但范围错误，金额可能是数字但超过权限，邮箱可能合法但属于错误收件人，这些都需要 Schema 校验在进行有业务校验。

其次就是工具匹配和检索，前面我们说的意图拆分，大部分为的就是匹配上合适的工具，所以工具最好能做到尽可能精准，如果模型上限不是 Top ，规则也做不到工具精细化区分，那么工具一定不要太多。

比如一次性把几十或上百个工具暴露给模型，大部分时候只会增加误选和参数混淆，所以一般来收在 Agent 会提前先检索工具，然后再注册给这次任务的大模型，比如可以：

根据意图和领域筛选工具集合
根据用户权限删除不可用工具
根据当前状态排除不合适工具
只向模型暴露最相关的少量工具

然后在执行前，都需要做格式校验、业务校验、权限校验和风险确认，这也是工具执行的基本礼仪。

def execute_tool(proposal, context): args = schema_registry.validate(proposal.tool_name, proposal.arguments) policy.authorize(context.user, proposal.tool_name, args) policy.validate_business_rules(context, proposal.tool_name, args)  if policy.requires_confirmation(proposal.tool_name, args): return pause_for_human_approval(proposal, args)  idempotency_key = build_idempotency_key( context.task_id, proposal.tool_name, args )  return executor.run( tool_name=proposal.tool_name, args=args, idempotency_key=idempotency_key, timeout_ms=10_000, )

另外就像前面说的意图置信度和风险，工具也需要有风险评级，或者说前面的意图风险基本就是来自工具风险，比如：

等级	典型操作	默认策略
R0 纯推理	分类、摘要、改写	可自动执行
R1 只读	查询订单、搜索邮件、读取文档	授权后自动执行，记录审计
R2 可逆写入	创建草稿、添加标签、保存临时配置	可自动或轻确认，必须幂等
R3 高风险写入	发邮件、退款、删除、转账、执行脚本、修改生产配置	强确认、最小权限、完整审计
R4 禁止或隔离	超出业务授权、跨租户访问、敏感批量操作	直接拒绝或转人工

甚至在用户 UI 上，我们可能需要展示：接下来将调用什么工具、操作哪个对象、影响范围、是否可撤销、关键参数是什么之类的展示。

然后工具肯定会有执行失败的时候，这里失败的错误也必须结构化，比如：

{ "status": "error", "error_type": "MISSING_ARGUMENT", "retryable": false, "field": "contract_id", "message": "缺少合同标识，需要用户选择候选合同", "suggested_action": "ask_user" }

一般来说常见错误分类：

节点错误：超时、限流、网络抖动，可指数退避重试
模型可修复错误：参数格式错误、缺字段，可反馈后重规划
用户可修复错误：缺少订单号、候选对象不唯一，应追问
权限与策略错误：不可重试，应解释并终止
永久业务错误：订单状态不允许取消，不应反复调用
系统性错误：工具连续失败，触发熔断和告警

有错误分类和评级，然后才能够对错误进行重试，重试的次数也可以按工具的幂等性、错误类型、时延预算和业务后果配置。

高风险操作肯定是允许重试次数为 0。

四、结构化输出

结构化我们前面一直强调，需要在 Agent 内部流转和持有，但是模型输出结构化也是有讲究的，一般 LLM 输出 JSON 时常见问题会有：

夹带解释文字、Markdown 代码块、字段缺失、类型错误、截断、枚举越界和语义错误。

这些都是很常见的问题，所以兼容和自动处理错误就需要一个可靠的 JSON flow 支持，比如：

明确的输出契约 → 模型原生 Structured Output / Function Calling → 检查停止原因和截断 → JSON 解析 → Schema 校验 → 业务语义校验 → 有限修复 → 带错误信息重试 → 降级或拆分任务

一般来说，肯定优先使用模型原生结构约束，比如支持严格 JSON Schema、Function Calling 或 Structured Output 的模型，肯定是不能依靠 Prompt 要求“只输出 JSON”这种毫无保障的渣男语录。

但是一般来说，我们还是必须区分三种正确状态：

语法正确：JSON 能解析
结构正确：字段和类型符合 Schema
语义正确：值符合真实业务和用户意图

严格的 Schema 判定主要解决前两项，语义正确就需要业务后续审查。

然后针对语法正确和结构正确问题，Agent 运行过程中肯定会出现错误，所以我们可以增加多种方式进行修复和重试，但是不要保证一定能修成功，不然容易死循环。

实际上 JSON 结构问题，可以安全修复的通常只有明确的格式问题，比如：

去掉代码块
提取唯一 JSON 片段
去掉尾逗号
补齐可确定的末尾括号
清理前后无关文本

但是如果遇到这些问题，一般不建议修复，比如：

不知道模型本来想选择哪个枚举
字段值在业务上矛盾
输出被截断且缺失大量内容
多个候选 JSON 无法判断哪个正确

这时候应该直接判定失败，然后重新请求或拆分任务重试，直接忽略这一份 JSON 结果。

另外，不建议用 Agent 去处理又长又深的结构化数据，如果数据层级又长又深，那就需要进行拆分，比如可以拆成：

先抽取事实
再生成标签
再做风险判断
最后再通过程序合并

不然截断概率和问题会很多，拆分后也可以对每个子步骤独立校验和评测。

五、RAG

RAG 这个也是老生常谈的话题了，按照 Agent 上的常见规则，一般可以分成：

先保证找得到
再保证排得对
最后保证答得忠实

也就是说，我们虽然简单说，RAG 不就是“向量库 + Top K + LLM” 么，但是实际生产链路上，一般至少包含：

解析、切分、索引、查询理解、多路召回、去重、重排、上下文装配、生成和引用。

比如文档解析，合同就是一个很典型的场景，它需要保留原始结构，因为普通文本可以按标题、段落和长度快速切分，但是合同、技术规范、政策文件、表格和扫描件则需要更精细的结构化解析。

所以对于文档，Chunk 除了正文，还需要保存：

文档 ID、版本和来源
标题层级
页码和段落位置
生效时间
权限标签
内容类型
与前后 Chunk 的关系

然后再根据业务做资源分层，比如“70% 普通文档、30% 高价值文档”，这个就需要根据你的实际业务情况去尝试了，。

然后接下来要做的就是两件事：召回和重排。

在召回阶段一般可以做加法，主要是为了避免关键证据出现遗漏，比如可以组合：

向量检索：处理语义相似
BM25：处理关键词、编号、专有名词
Metadata Filter：处理时间、部门、文档类型和权限
Query Rewrite：补充实体、同义表达和时间范围
Multi-Query：从多个角度扩展召回
Parent-Child Retrieval：小块匹配，大块提供上下文

然后重排阶段就需要做减法了，召回较多候选后，我们就需要用 Reranker 和规则特征排序，比如：

与问题的直接相关性
标题和章节匹配
来源权威性
时效性
权限与租户
是否包含完整答案所需的多个事实

最后就是上下文装配，把东西都拿到手之后，就可以进行上下文状态，这里一般就可以做一些二次优化，比如：

针对每个来源控制可能的最大长度
重复内容去除
多跳问题的证据覆盖
新旧版本冲突处理
引用编号校对
不可信内容与系统指令的隔离

比如对于一个合同助手，检索结果必须能追溯到具体合同、页码和条款，不能只给模型一段失去来源的文本。

六、记忆

Agent 里的记忆，一般来说可以分成短期状态、长期事实和外部知识，结构上类似：

也就是一般来说可以做四层记忆，例如：

层次	保存内容	典型存储	生命周期
工作记忆	当前任务步骤、临时变量、最近消息	Graph State / Checkpoint	当前线程
摘要记忆	项目目标、关键决策、未完成事项	摘要表或线程摘要	中期
语义记忆	历史对话片段、代码讨论、事件	向量库	长期、可检索
结构化记忆	稳定偏好、项目配置、联系人、版本	关系库 / KV / 文档库	长期、可更新

简单来说就是：

短期状态解决“当前做到哪一步”
长期记忆解决“过去有哪些与当前任务相关的信息”
外部知识库解决“组织拥有的事实资料”

这几个东西需要分开存储，不能混成一个向量库。

而实际上，记忆写入也需要策略，用户角度肯定不是每句话都值得长期保存，所以写入前至少判断：

这是不是长期稳定的信息？
是否用户明确保存？
有没有包含敏感信息？
和已有的记忆会不会重复或者冲突？
内容是不是只是临时假设？
有没有过期时间？
用户需不需查看、修改和删除？

比如项目从 Flutter 改为 Swift 后，旧记忆就不能继续通过“当前技术栈”被召回，所以根据业务形态，我们需要保留版本关系：过去使用 Flutter，某天决策改为 Swift，当前有效值为 Swift。

最后还需要防止记忆污染，比如从网页、邮件或工具结果中读取的内容可能包含恶意指令。

所以外部内容只能作为数据，不能自动升级为系统指令或长期偏好，所以一般也会建议为记忆保存结构化信息，比如：

来源、创建时间、确认状态、可信度、有效期、覆盖关系和敏感等级，重要事实最好通过程序校验或用户确认后再写入。

这部分其实还可以看《Claude Opus 5 系统提示词泄漏》的介绍，Claude 就在它系统提示词设计了一套持久化 memory filesystem 规范。

七、评测

评测也是关键的实现，Agent 开发里，整个评测主要看的是执行轨迹：

一般来说可以比较完整的测评可以分成四级评测，比如：

第一层组件评测，单独测试：
- 意图识别准确率和澄清率
- 工具 Top-K 命中率
- 参数字段准确率
- JSON / Schema 成功率
- 检索 Hit Rate、Recall@K、MRR、NDCG
- Reranker 排序效果
- 权限和风险分类召回率

第二层轨迹评测，最终答案虽然可能对了，但是你也需要关注过程是不是安全，比如轨迹评测可以关注：
- 是否选择了必要步骤
- 有没有调用了多余工具
- 做没做先写后确认
- 有没出现在缺参数时脑补
- 是不是发生过循环
- 有没有超过预算
- 执行会不会有越权资源
- 失败后恢复路径有没有走对

第三层可以做结果评测，比如关注：
- 任务完成度
- 事实正确性
- 引用一致性
- 格式正确性
- 是否满足用户约束

第四层就是业务和运行指标，比如：
- 一次任务成功率
- 人工接管率
- 用户澄清轮数
- P50 / P95 / P99 时延
- 单任务 Token、模型和工具成本
- 高风险操作拦截率
- 重试率与熔断率
- 用户满意度和问题一次解决率

所以做 Agent 可能很快，但是跑评测，测试稳定性，可靠性，模型兼容性，反而是最费时费力的一个过程。

然后针对 RAG 的评测还要能反推故障层，因为在「合同助手」里，RAG 出现问题是最大也是最容易出现的，所以你需要知道 RAG 问题出现的轨迹，trace 信息要支持反推故障层，比如：

现象	更可能的问题	优先排查
Context Recall 低	关键资料没找回	Chunk、Query Rewrite、混合检索、Top-K
Context Precision 低	噪声太多	Reranker、Metadata、去重、查询理解
Recall 高但 Faithfulness 低	资料找到了，模型仍在编	生成模型、Prompt、引用约束、上下文冲突
Faithfulness 高但 Answer Relevancy 低	没乱编但答非所问	意图、问题改写、Prompt 和完成条件
工具调用成功但任务失败	工具结果没有解决用户目标	结果验证和重新规划
离线高分、线上低满意度	数据分布或业务目标不一致	线上样本、任务定义、产品交互

最重要的就是要建立完成的 Agent 测试集。

八、可观测性 Trace

Trace 就是 AI 调试 Agent 的最重要入口，我们前面说了那么多轨迹，说的就是 Trace，Agent 必须记录完整轨迹。

比如每次运行都记录：

trace_id / task_id / thread_id 用户与租户上下文 模型、版本、温度、Token 使用 路由结果、置信度和风险等级 状态快照和节点耗时 召回 Query、候选 Chunk、重排分数和引用 工具候选、最终选择、参数和 Schema 版本 授权、确认和策略判定 工具耗时、状态码、重试和幂等键 最终完成条件和失败原因 人工修改与用户反馈

当然，在日志上我们不能无边界保存 Prompt、邮件正文和敏感参数，Trace 本身也要遵守最小化、脱敏、访问控制和保留期限，比如关键看板可以包括：