过去一年里我见过太多 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(instanceraw_call, schemaCALL_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(instancearguments, schematool[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先保证模型输出的是一个“名字 参数对象”的结构而不是一坨自由文本。每个工具自带 Schemaquery_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_asciiFalse) 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_