过去一年里,我见过太多 Agent 项目死在同一个地方:演示时一切正常,一上生产就崩。模型偶尔调用错参数、工具返回结果没人校验、整条链路没有日志可以追踪、改了一版 prompt 之后不知道哪些场景退化了、某个高危操作差点在无人确认的情况下直接执行。这些问题单独看都不难解决,但合在一起,就把“能写 Agent”和“能上线 Agent”画成了一道分水岭。
在梳理业界关于 Agent 工程化的资料时,Linear 团队的分享让我印象很深。相比把模型效果调到多惊艳,他们更强调把不确定性管住:流程、契约、观测、评估、兜底,每一层都用工程手段给 Agent 上保险。结合我在实际项目里的落地经验,这篇文章把这套思路整理成构建生产级 Agent 的 5 条规则。每条规则都会讲清楚“为什么重要”和“怎么落地”,并配可运行的代码示例。最后一节我会把 5 条规则整合到一个带安全闸门的工单处理 Agent 项目里,方便你对照着搭自己的工程骨架。
无论你是在做客服机器人、代码助手、数据分析 Agent,还是企业内部知识库问答,这套方法都适用。
1. 背景:生产级 Agent 到底难在哪里
1.1 原型 Agent 与生产级 Agent 的差距
很多人对 Agent 的第一印象来自 Demo:输入一句话,模型自动调用工具、给出答案,看起来很聪明。但 Demo 能跑通,并不代表它能稳定运营。
原型 Agent 和生产级 Agent 的差别,可以看下面这张表:
| 维度 | 原型 Agent | 生产级 Agent |
|---|---|---|
| 任务路径 | 单条链路,怎么试都能通 | 大量分支、异常、边界情况都要处理 |
| 模型输出 | 大部分时候正确 | 必须假定模型会犯错,并做出防御 |
| 工具调用 | 参数手动构造,偶发错误可接受 | 参数必须严格校验,错误不能下钻到业务 |
| 可观测性 | 打印几行日志 | 全链路 trace、耗时、成本、token 都可追踪 |
| 变更控制 | 改 prompt 随时生效 | 有评估、灰度、回滚机制 |
| 安全边界 | 无风险操作 | 高危操作必须有人确认,权限最小化 |
| 稳定性 | 挂了就挂了 | 有降级、重试、熔断、兜底 |
差距的本质是:原型阶段你面对的是“模型的智能问题”,生产阶段你面对的是“系统的工程问题”。后者需要一套规则来约束。
1.2 理解 Agent 的三层结构
为了后面讨论不跑偏,先统一一下 Agent 的架构模型。一个生产级 Agent 通常分为三层:
- 模型层:LLM 负责意图理解、信息抽取、决策生成。这是最灵活、也最不可控的一层。
- 编排层:负责状态流转、工具调度、上下文管理。这一层应该是确定性的。
- 工具与数据层:Agent 实际触达业务系统的通道,比如查订单、发工单、写库存。这一层必须有契约和安全边界。
很多团队的问题在于,把“智能”放得太满。模型层做了太多事,编排层和工具层又没有约束,最后整个系统呈混沌状态。正确的做法是:确定性交给代码,不确定性交给模型,且模型只能在我们划定的范围内做决策。这正是下面 5 条规则的核心思想。
1.3 5 条规则总览
先给出全貌,方便你建立整体印象:
- 规则一:先定义流程,再让模型做决策。用状态机固定业务骨架,模型只做分支选择。
- 规则二:工具调用必须契约化。参数结构、类型、取值范围都在调用前校验。
- 规则三:可观测性是 Agent 的生命线。没有 trace 就没有排查能力,没有日志就没有迭代依据。
- 规则四:把评估当成测试用例来维护。每次改动都要跑回归,用数据说话。
- 规则五:为失败设计,而不是追求完美。高危操作必须人工确认,系统必须能优雅降级。
接下来,我们逐条展开。
2. 规则一:先定义流程,再让模型做决策
2.1 为什么不能把整个链路都交给模型
先看一个反面案例。很多人在实现客服 Agent 时,直接把用户输入丢给模型,让模型“自由发挥”:判断意图、决定调什么工具、构造参数、生成回复。看起来灵活,实际上一旦业务复杂,模型会在三个地方失控:
- 意图判断不稳定。同一句话换个说法,可能走了完全不同的分支。
- 工具选择不可预测。模型可能把“查询订单”理解成“创建退款”。
- 流程边界模糊。什么时候该结束、什么时候该人工介入,模型没有全局视角。
流程的本质是“限制”,而生产系统恰恰需要限制。先把业务路径画成状态机,让模型在状态机允许的动作里做选择,系统的行为才是可预期的。
2.2 用状态机固定流程骨架
以一个工单处理 Agent 为例。用户进来之后,Agent 需要经历三个阶段:
- 识别意图(查询 / 退款 / FAQ)。
- 收集必要信息(订单号、金额等)。
- 调用工具执行,并决定是否需要人工审批。
这个过程用状态机表达非常清晰:
INIT -> INFO_GATHERING -> TOOL_EXECUTING -> CLOSED | +----> HUMAN_APPROVAL(高危操作等待人工)状态机的好处有几点:
- 流程可见。产品、开发、测试看到的是同一张状态图。
- 行为可控。非法跳转根本不会发生,比如没收集完信息就去调用工具。
- 便于测试。每个状态和转移都是独立单元,可以单测。
2.3 状态机代码示例
# 文件路径:agent/workflow.py from enum import Enum class TicketState(str, Enum): INIT = "init" INFO_GATHERING = "info_gathering" TOOL_EXECUTING = "tool_executing" HUMAN_APPROVAL = "human_approval" CLOSED = "closed" class Transition(str, Enum): INTENT_KNOWN = "intent_known" INFO_COMPLETE = "info_complete" NEED_APPROVAL = "need_approval" CONFIRMED = "confirmed" REJECTED = "rejected" FAILED = "failed" WORKFLOW = { TicketState.INIT: { Transition.INTENT_KNOWN: TicketState.INFO_GATHERING, Transition.FAILED: TicketState.CLOSED, }, TicketState.INFO_GATHERING: { Transition.INFO_COMPLETE: TicketState.TOOL_EXECUTING, Transition.FAILED: TicketState.CLOSED, }, TicketState.TOOL_EXECUTING: { Transition.CONFIRMED: TicketState.CLOSED, Transition.NEED_APPROVAL: TicketState.HUMAN_APPROVAL, Transition.FAILED: TicketState.CLOSED, }, } TERMINAL_STATES = {TicketState.CLOSED, TicketState.HUMAN_APPROVAL}这段代码里,WORKFLOW是一个“当前状态 -> 动作 -> 下一个状态”的映射表。执行器每轮只做一件事:根据当前状态,让模型从允许的动作里选一个,然后查表转移。模型永远不可能让状态跳到不该去的地方。
2.4 什么时候适合让模型自由发挥
强调流程,不代表完全禁止模型自由发挥。以下场景可以放开一些:
- 文本生成类任务,比如写周报、润色文案,本身没有强流程。
- 探索式任务,比如 open-ended 的资料分析,用户也不知道终点在哪。
- 工具链非常简单,只有一层调用,且调用失败无副作用。
判断标准只有一个:如果某一步出错会造成业务损失,它就必须被流程约束;如果只是信息处理,可以交给模型。在实际项目中,我倾向把 80% 的路径做成确定性的,只有 20% 的灵活分支留给 LLM。这样既保留了智能感,又守住了稳定性。
3. 规则二:工具调用必须契约化
3.1 工具就是 Agent 的边界
Agent 的能力边界,本质上由它能调用的工具决定。工具一旦可以被随意调用,风险也随之而来:
- 参数缺失或类型错误,下游服务直接 500。
- 参数越界,比如退款金额传入负数。
- 调用了不该调用的高危接口,比如删除数据。
- 模型幻觉出工具名,调用链直接断掉。
所以,工具层必须像对外 API 一样做契约管理。模型输出的是“调用意图”,不是“最终命令”。调用是否合法,必须由代码校验。
3.2 用 JSON Schema 做参数校验
一个比较通用的做法是:每个工具注册时,带上自己的参数 Schema。模型输出的工具调用,先经过 Schema 校验,通过后才真正执行。
# 文件路径:agent/tools.py from jsonschema import validate, ValidationError TOOL_REGISTRY = {} CALL_SCHEMA = { "type": "object", "properties": { "name": {"type": "string"}, "arguments": {"type": "object"}, }, "required": ["name", "arguments"], } def register_tool(name: str, schema: dict): def decorator(func): TOOL_REGISTRY[name] = {"func": func, "schema": schema} return func return decorator @register_tool( "query_order", { "type": "object", "properties": { "order_id": {"type": "string", "pattern": "^ORD-\\d+$"}, }, "required": ["order_id"], }, ) def query_order(order_id: str) -> dict: # 示例实现,实际项目中在这里接入订单中心 return {"order_id": order_id, "status": "已发货", "position": "上海分拨中心"} @register_tool( "create_refund", { "type": "object", "properties": { "order_id": {"type": "string", "pattern": "^ORD-\\d+$"}, "amount": {"type": "number", "minimum": 0.01}, }, "required": ["order_id", "amount"], }, ) def create_refund(order_id: str, amount: float) -> dict: return {"order_id": order_id, "amount": amount, "status": "refund_created"} def call_tool(raw_call: dict) -> dict: try: validate(instance=raw_call, schema=CALL_SCHEMA) name = raw_call["name"] arguments = raw_call["arguments"] tool = TOOL_REGISTRY.get(name) if tool is None: return {"ok": False, "error": f"未知工具: {name}"} validate(instance=arguments, schema=tool["schema"]) result = tool["func"](**arguments) return {"ok": True, "result": result} except ValidationError as e: return {"ok": False, "error": f"参数校验失败: {e.message}"} except Exception as e: return {"ok": False, "error": f"工具执行失败: {str(e)}"}注意几个要点:
CALL_SCHEMA先保证模型输出的是一个“名字 + 参数对象”的结构,而不是一坨自由文本。- 每个工具自带 Schema,
query_order要求订单号必须匹配ORD-加数字的格式;create_refund要求金额必须大于 0。 - 校验失败时返回标准错误结构,编排层可以据此决定是重试、追问用户,还是终止流程。
3.3 返回值也要归一化
很多人只校验了入参,忽略了返回值。生产实践中,我建议对所有工具的返回值做一层“归一化包装”:
{ "ok": true, "result": { "...": "业务数据" } }失败时统一返回:
{ "ok": false, "error": "错误描述" }这样做的好处是,编排层只认ok字段,不需要为每个工具写不同的异常处理分支。工具内部的异常不应该直接抛到上层,而应该被捕获并转换成结构化错误,否则 trace 会非常难查。
3.4 常见误区
- 只做基础类型校验,不做业务校验。Schema 里写了
minimum: 0.01,负数退款就进不来,这一步必须做,不能只靠模型自觉。 - 把校验逻辑写死在每个工具函数里。每个函数都
if not isinstance(x, str)会导致大量重复代码,统一注册 + Schema 校验更适合规模化。 - 返回原始异常字符串。把数据库连接错误直接拼进回复里给用户看,既不安全也不友好,应该统一包装。
4. 规则三:可观测性是 Agent 的生命线
4.1 Agent 可观测性要记录什么
传统服务排查问题,看报错日志就够了。Agent 不一样,它多了一层“模型决策过程”,问题往往出在:模型为什么选择了这个工具?当时上下文里有什么?哪个参数被填错了?
所以 Agent 的可观测性至少需要这几类数据:
| 数据类型 | 关键字段 | 用途 |
|---|---|---|
| 运行轨迹 | trace_id、状态流转、动作 | 还原整条链路 |
| 模型调用 | prompt、completion、token、耗时 | 分析模型行为和成本 |
| 工具调用 | 工具名、入参、出参、耗时 | 定位工具侧问题 |
| 业务结果 | 是否成功、状态码、错误信息 | 判断用户侧影响 |
这里最核心的概念是trace_id。一次用户请求从头到尾,无论产生了多少次模型调用、多少次工具调用,都应该携带同一个 trace_id,这样日志才能串成一条完整的链路。
4.2 结构化日志与 trace_id
我习惯把所有 Agent 日志输出为 JSON 格式,便于采集到日志平台后做检索和分析。下面是一个最小实现:
# 文件路径:agent/logging_conf.py import json import logging import time import uuid from contextvars import ContextVar trace_id_var: ContextVar = ContextVar("trace_id", default="-") class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: payload = { "time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(record.created)), "level": record.levelname, "module": record.name, "trace_id": trace_id_var.get(), "message": record.getMessage(), } extra = getattr(record, "extra_fields", None) if extra: payload.update(extra) return json.dumps(payload, ensure_ascii=False) def setup_logging() -> None: handler = logging.StreamHandler() handler.setFormatter(JsonFormatter()) root = logging.getLogger() root.handlers = [handler] root.setLevel(logging.INFO) def start_trace() -> str: trace_id = uuid.uuid4().hex[:12] trace_id_var.set(trace_id) return trace_id在编排层,每次状态流转都打一条结构化日志。排查问题时,直接按 trace_id 检索,就能看到类似这样的完整过程:
{"time": "2025-06-01 10:00:01", "level": "INFO", "module": "agent.core", "trace_id": "a1b2c3d4e5f6", "message": "state_