AI Agent 工程化实战:工具设计、循环控制、上下文管理与可观测性

用现在的框架搭一个 AI Agent 的 demo,有多容易?几十行代码:挂个大模型、注册几个工具、写个 ReAct 循环,它就能像模像样地"自己思考、自己调工具、自己给答案"。我们当时就是这么搭出第一个版本的,在会议室里演示,输入一句"帮我查下这个订单为什么还没发货并通知客户",它真的去查了订单、读了物流、起草了通知,全场惊艳。

然后我们把它放给真实用户用,一周内见识了它各种匪夷所思的翻车姿势:有一次它陷入死循环,反复调同一个查询工具几百次,一夜之间烧掉了平时一个月的 API 预算;有一次它"自信地"把退款工具的参数填错,差点给错了客户退款;还有用户反馈同样的问题问两次给出南辕北辙的两个答案,根本没法复现。那段时间我才真正意识到:让 Agent "能跑"和让它"能上线",中间隔着一整套工程。demo 比的是模型聪不聪明,生产比的是这套系统在模型犯傻、工具报错、输入刁钻时还能不能可控、可靠、可观测、不失控烧钱。这篇就把我们把一个惊艳的 demo 打磨成线上能用的 Agent 过程中,补齐的那几块工程拼图讲清楚。

先看清:Agent 在生产里到底怎么翻车

把这一年踩过的坑归归类,会发现 Agent 的故障模式和传统服务很不一样——它们大多源于"大模型的输出本质上是不确定的",而我们却把一个不确定的大脑接到了能产生真实副作用的工具上。先把这些翻车模式列出来,后面逐个给工程解法:

这五类问题有个共同点:它们在 demo 那种"一次性、理想输入、人盯着看"的场景下几乎都不暴露,只在真实流量的长尾输入、长任务、无人值守里集中爆发。而治理它们的思路,不是去祈求模型变得更聪明,而是在模型外面套上一圈"约束 + 兜底 + 观测"的工程脚手架。先从这套脚手架的骨架——Agent 的执行循环说起。

Agent 的执行循环:每一步都要能被掐住

主流 Agent 本质都是一个"思考→调工具→拿结果→再思考"的循环(ReAct 范式)。demo 版本往往把这个循环写得很乐观:让模型一直转,直到它自己说"我搞定了"。但生产版本必须在循环的每一个环节都预留可以掐断、可以兜底、可以记录的钩子。下面这张图是我们线上 Agent 的实际控制流,和裸 demo 的差别全在那些灰色的"闸门"上:

注意图里那几道闸门:最大步数挡住死循环,参数校验挡住工具调用幻觉,超时/重试/幂等挡住工具层的不可靠,每步 trace则是事后能复盘的前提。下面就从最常被忽视、却最影响 Agent 行为的一环——工具设计——开始拆。

工具设计:Agent 调得对不对,一半取决于你怎么描述工具

很多人以为 Agent 调错工具是模型不够聪明,其实更多时候是工具的描述写得太烂。模型决定调哪个工具、怎么填参数,几乎全靠你给工具写的那段自然语言描述和参数 schema。描述含糊、边界不清,模型就只能靠猜:

# ❌ 烂工具描述:名字含糊、说明一句话、参数没约束,模型只能瞎猜
@tool
def query(id: str) -> str:
    """查询信息"""          # 查什么信息?id 是什么的 id?啥时候该用它?全靠模型猜
    return db.find(id)

# ✅ 好工具描述:职责单一、说清何时用何时不用、参数有类型有约束有示例
@tool
def get_order_status(order_id: str) -> dict:
    """根据订单号查询订单的当前状态与物流信息。
    使用场景:用户询问订单进度、是否发货、何时到货时调用。
    不要用于:查询用户账户、商品库存(那是别的工具)。
    参数 order_id:订单号,格式形如 'ORD-' 开头加 12 位数字,例如 'ORD-202405010001'。
    返回:{status, shipped_at, carrier, tracking_no} 的字典;订单不存在时返回 {error}。
    """
    if not re.match(r'^ORD-\d{12}$', order_id):
        return {"error": f"订单号格式非法: {order_id}"}   # 边界处自己兜住,别让模型乱传也不报错
    return order_repo.get_status(order_id)

好的工具描述有几条硬标准:名字直白表意(get_order_status 而非 query);说清"何时该用、何时不该用",把它和其它工具的边界划开,避免模型在几个相似工具间选错;参数给出类型、格式、示例,让模型有样学样;在工具内部对参数做校验,非法输入就地返回清晰的错误而不是抛异常或返回脏数据。一个原则贯穿始终:把工具当成"给一个聪明但没看过你代码的实习生用的 API"来设计文档——你描述得越清楚,Agent 调得越准。

循环控制:给 Agent 装上"刹车"和"收敛判定"

那次烧掉一个月预算的死循环,根因就是 demo 版本太信任模型——让它一直转到自己说"完成"为止。可模型一旦卡在某个判断上,完全可能反复调同一个工具、永远不收敛。生产版本必须强制装上几道刹车:最大步数上限、对重复调用的检测、以及一个独立于模型自我判断的收敛/超限处理:

def run_agent(task: str, max_steps: int = 8, budget_usd: float = 0.5):
    ctx = build_context(task)
    seen_calls = {}                       # 记录"工具名+参数"出现次数,抓重复调用
    cost = 0.0

    for step in range(max_steps):         # ✅ 硬上限:绝不允许无限循环
        resp = llm.chat(ctx, tools=TOOLS)
        cost += resp.cost_usd
        if cost > budget_usd:             # ✅ 预算闸:单任务烧超就强制收尾
            return finalize(ctx, reason="budget_exceeded")

        if not resp.tool_call:            # 模型给出最终答案
            return validate_output(resp.content)

        call = resp.tool_call
        key = (call.name, json.dumps(call.args, sort_keys=True))
        seen_calls[key] = seen_calls.get(key, 0) + 1
        if seen_calls[key] >= 3:          # ✅ 同样的调用重复 3 次 = 八成卡死了
            return finalize(ctx, reason="loop_detected")

        result = exec_tool(call)          # 见下一节:带超时/重试/幂等
        ctx.append_tool_result(call, result)

    return finalize(ctx, reason="max_steps_reached")  # ✅ 步数耗尽也要优雅收尾

这里的关键观念是:绝不能把"任务是否完成"这个判断完全交给模型自己。模型说"我还需要再查一次"可能是对的,也可能是它陷进了某个循环。所以我们在模型的自我判断之外,叠加三道客观的硬约束——步数、预算、重复检测——任何一道触发,就由系统(而非模型)接管,强制走"优雅收尾"逻辑:把已经拿到的信息整理成一个尽量有用的回复,并明确告诉用户"因为某某限制提前结束了"。宁可给一个诚实的半成品,也不能放任它无限空转。

上下文管理:别让历史无限膨胀,撑爆窗口又烧钱

第二个高发问题是上下文爆炸。多轮对话或长任务里,如果每一步都把全部历史一股脑塞进上下文,它会越滚越长,直到撑爆模型的上下文窗口直接报错;就算没爆,每一步都携带全量历史也意味着 token 越烧越多、响应越来越慢。生产 Agent 必须主动管理上下文,而不是无脑堆积:

def build_context(task, history, max_tokens=6000):
    ctx = [system_prompt(), {"role": "user", "content": task}]

    # 策略一:保留最近 N 轮原文(近期信息最相关,原样保留)
    recent = history[-4:]

    # 策略二:更早的历史压缩成摘要,只留关键事实,不留逐字对话
    older = history[:-4]
    if older:
        summary = llm.summarize(older, focus="已确认的事实、已调用的工具及其结果")
        ctx.append({"role": "system", "content": f"【早期对话摘要】{summary}"})

    ctx.extend(recent)

    # 策略三:工具返回的大块结果(如长 JSON、长文档)按需截断或转引用
    ctx = truncate_tool_outputs(ctx, per_output_limit=800)

    # 最后兜底:仍超限就从最旧的非系统消息开始丢
    while count_tokens(ctx) > max_tokens:
        drop_oldest_non_system(ctx)
    return ctx

上下文管理的核心是认清一个事实:上下文窗口是一种有限且昂贵的资源,要像管内存一样管它。不是所有历史都同等重要——最近几轮的原文要保真,久远的对话压成摘要就够,工具吐出来的超长结果该截断截断、该转成"可按需再查的引用"就别全塞进去。这套"近期保真、远期摘要、大块截断、超限淘汰"的分层策略,能在不明显损失任务相关信息的前提下,把上下文长度稳稳摁在一个可控的区间里,既防爆窗口,又省 token。

结构化输出:让 Agent 的回答能被程序可靠地接住

还有一类隐蔽的翻车:Agent 最终给的是自然语言,而下游程序需要的是结构化数据(状态、字段、下一步动作)。如果靠正则去抠模型的自由文本,模型措辞一变解析就崩。正确做法是用 schema 强约束输出,并在解析失败时把错误回灌给模型让它自己修:

from pydantic import BaseModel, ValidationError

class AgentDecision(BaseModel):
    action: Literal["reply", "escalate", "refund"]   # ✅ 动作被约束成有限枚举
    message: str
    refund_amount: float | None = None

def get_decision(ctx, retries=2):
    for _ in range(retries + 1):
        raw = llm.chat(ctx, response_format=AgentDecision)  # 让模型按 schema 产出 JSON
        try:
            return AgentDecision.model_validate_json(raw)   # ✅ 校验通过才放行
        except ValidationError as e:
            ctx.append({"role": "system",
                        "content": f"上次输出不符合要求: {e}。请严格按 schema 重新输出。"})
    raise RuntimeError("模型多次未能产出合法结构化输出")

把输出约束成 schema 有双重收益:一是下游能可靠地接住结果,不用赌模型的措辞;二是把 action 这种关键决策限定成有限枚举,从源头杜绝模型"发明"一个你没实现的动作。配合"校验失败就把具体错误回灌、让模型自我修正"的重试,绝大多数格式问题在一两轮内就能自愈,而不至于把脏数据丢给下游。

工具执行容错:工具会超时、会报错、会被重复调

Agent 调的工具往往是真实的外部依赖——数据库、第三方 API、内部微服务,它们会超时、会限流、会偶发失败。demo 里工具总是秒回成功,生产里你必须假设每个工具调用都可能出问题,并给它套上超时、重试、幂等三层保护;尤其那些有副作用的工具(下单、退款、发消息),重复执行的后果可能很严重,幂等不是可选项:

def exec_tool(call, timeout=10, retries=2):
    fn = TOOLS[call.name]
    last_err = None
    for attempt in range(retries + 1):
        try:
            # ✅ 超时保护:别让一个卡住的工具拖垮整个 Agent
            return run_with_timeout(fn, call.args, timeout=timeout)
        except (TimeoutError, TransientError) as e:   # 只对"可重试"的瞬时错误重试
            last_err = e
            time.sleep(2 ** attempt)                  # 指数退避
        except PermanentError as e:
            # ✅ 永久性错误(参数非法等)别重试,直接把错误回灌给模型让它换法子
            return {"error": str(e), "retryable": False}
    return {"error": f"工具多次失败: {last_err}", "retryable": False}

# ✅ 有副作用的工具靠幂等键防重复执行:同一个 idempotency_key 只真正执行一次
@tool
def issue_refund(order_id: str, amount: float, idempotency_key: str) -> dict:
    if refund_store.exists(idempotency_key):
        return refund_store.get(idempotency_key)      # 重复调用直接返回首次结果
    result = payment.refund(order_id, amount)
    refund_store.save(idempotency_key, result)
    return result

这里有个容易忽略的细节:工具失败后,要把"失败"本身作为一条可读的信息回灌给模型,而不是直接抛异常崩掉整个 Agent。模型拿到"订单查询超时了"这条反馈,完全可能自己决定换个工具、或者如实告诉用户"暂时查不到、请稍后再试"——这种优雅降级,恰恰是 Agent 比传统硬编码流程更灵活的地方。但前提是你得把错误规整成模型能理解的文字喂回去,而不是让它在异常里裸奔。

可观测性:每一步都要留痕,否则你永远在盲调

Agent 最让人头疼的是"黑盒":用户说它答错了,你却不知道它中间想了什么、调了哪些工具、为什么走到那一步。没有可观测性的 Agent,调试基本靠玄学。所以我们给每个任务分配一个 trace_id,把循环里的每一步——模型的思考、选了哪个工具、传了什么参数、工具返回了什么、花了多少 token 和钱——全都结构化地记下来:

def trace_step(trace_id, step, resp, call=None, result=None):
    log.info("agent_step", extra={
        "trace_id": trace_id,
        "step": step,
        "thought": resp.reasoning,             # 模型这一步的思考
        "tool": call.name if call else None,   # 调了哪个工具
        "tool_args": call.args if call else None,
        "tool_result_digest": digest(result), # 工具结果(大的存摘要+引用)
        "prompt_tokens": resp.prompt_tokens,
        "completion_tokens": resp.completion_tokens,
        "cost_usd": resp.cost_usd,             # ✅ 成本归因到每一步
        "latency_ms": resp.latency_ms,
    })
    metrics.incr("agent.tokens", resp.total_tokens, tags={"tool": call.name if call else "final"})
    metrics.incr("agent.cost_usd", resp.cost_usd)

有了逐步 trace,前面那些翻车都从"无从下手"变成"一眼看穿":死循环?trace 里同一个工具调用连刷十几条一目了然;成本失控?按 trace_id 聚合 cost_usd 就知道是哪类任务在烧钱、烧在哪一步;答错了?顺着 thought 和 tool_result 回放,能精确定位是模型判断错了还是工具喂了脏数据。可观测性不是上线后锦上添花的监控,而是开发阶段就必须内建的调试能力——Agent 的不确定性决定了你几乎不可能靠"本地复现"来调试它,只能靠生产环境里留下的完整轨迹来回溯。

评估:一个不确定的系统,怎么知道它到底行不行

最后一块、也是最容易被跳过的拼图是评估。传统软件改一行有单元测试兜底,但 Agent 改了个提示词、换了个模型,行为可能整体漂移,而它的输出又不是唯一正确答案,没法简单 assert ==。我们的做法是建一个固定的评测集,用多个维度去打分,每次改动前后都跑一遍对比:

# 固定评测集:每条是一个任务 + 期望达成的检查点(而非唯一标准答案)
CASES = [
  {"task": "查订单 ORD-202405010001 为何没发货",
   "must_call": ["get_order_status"],        # 必须调到的工具
   "must_not_call": ["issue_refund"],         # 绝不该碰的危险工具
   "max_steps": 4,
   "rubric": "应说明未发货原因,不得编造物流单号"},
]

def evaluate(agent, cases=CASES):
    report = []
    for c in cases:
        trace = agent.run(c["task"])
        called = [s.tool for s in trace.steps if s.tool]
        ok_tools = set(c["must_call"]) <= set(called) and not (set(c["must_not_call"]) & set(called))
        ok_steps = len(trace.steps) <= c["max_steps"]
        ok_quality = llm_judge(trace.final, c["rubric"])   # 用模型按 rubric 打分,人工抽检
        report.append({"task": c["task"], "tools": ok_tools, "steps": ok_steps, "quality": ok_quality})
    return report

Agent 评估的关键转变是:从"检查输出是否等于标准答案",转向"检查行为是否满足一组约束"。它有没有调对工具、有没有碰不该碰的危险工具、步数是否在合理范围、最终回答在 rubric 下质量如何——这些可检查的维度,比纠结"逐字答案对不对"实际得多。再配合用更强的模型当"裁判"(LLM-as-judge)对开放式回答打分、辅以人工抽检,就能在每次改提示词或换模型时,量化地看出是变好了还是变差了,而不是凭感觉上线、出了事才知道。

该给 Agent 加哪些"护栏",照这棵树决策

把上面这些工程手段串起来,其实就是一套"我这个 Agent 要不要上某道护栏"的判断逻辑。不是每个 Agent 都需要全套——一个只读、无副作用的查询助手和一个能下单退款的操作型 Agent,护栏的密度天差地别。下次设计时,先过一遍这棵树:

沉淀成清单的几条 Agent 工程铁律

这套实践收口成了我们做 Agent 的工程规范,新 Agent 上线前按它逐条过审:

1. 工具描述按"给实习生的 API 文档"写:名字表意、说清何时用/不用、参数给类型格式示例,工具内部自校验参数。
2. 永远不把"是否完成"全交给模型:最大步数、预算上限、重复调用检测三道客观硬约束,任一触发由系统强制优雅收尾。
3. 上下文像内存一样管:近期原文保真、远期压成摘要、大块结果截断或转引用、超限淘汰最旧,别无脑堆历史。
4. 有副作用的工具必须幂等:用幂等键防重复执行;退款、下单这类高危动作再加二次确认或人工兜底。
5. 工具失败要降级不要崩:超时/重试/指数退避;把错误规整成文字回灌给模型,让它换法子或如实告知用户。
6. 关键输出用 schema 约束:决策动作限定成有限枚举,解析失败把错误回灌重试,杜绝模型发明未实现的动作。
7. 每一步都留痕、每一分钱都归因:逐步 trace(思考/工具/参数/结果/token/成本),配固定评测集在每次改动前后量化对比。

几个反复见到的认知误区

这一年里,我发现团队内外对 Agent 有几个相当普遍的误区,值得专门点破。

第一个误区是"模型够强,这些工程就不需要了"。这是最根本的误解。模型越强,确实会少犯一些低级错误,但它消除不了不确定性本身——再强的模型也是概率性输出,也会偶发地调错工具、陷入某种循环、给出格式不对的结果。工程护栏不是用来弥补模型的"笨",而是用来约束这种固有的不确定性、并在出问题时兜住。换更强的模型能提高上限,但护栏决定的是下限——而生产系统的稳定性,恰恰是由下限决定的。

第二个误区是把 Agent 当成"更聪明的 if-else"来期待,要么就走向另一个极端,觉得"既然它会自己思考,那把所有事都交给它自由发挥就好"。两头都不对。Agent 真正的甜区是"需要一定灵活判断、但又必须在明确边界内行动"的任务:你用工具的设计、提示词、护栏给它划好可行动的空间,让它在这个空间里灵活地选择路径。给的自由太少,它还不如硬编码流程;给的自由太多、又不设边界,它就会用你想象不到的方式闯祸。设计 Agent 的功夫,很大程度就是设计这个"自由与约束"的边界。

第三个误区是"demo 跑通了就差不多能上线了"。这是我们交过最贵学费的一条。demo 和生产之间的差距,几乎全部落在这篇讲的这些"不性感"的工程上:容错、限流、幂等、上下文管理、可观测性、评估。demo 验证的是"这个想法在理想情况下能不能成立",而这恰恰是整个工作里最简单的一小部分。一个能演示的 Agent 可能只完成了 20% 的工作量,剩下 80% 全在让它在真实世界的混乱输入、依赖故障、无人值守下依然可控。低估这 80%,就会像我们当初那样,在某个深夜被一封 API 账单或一笔错误退款惊醒。

写在最后

回头看这段把 demo 打磨成生产系统的经历,我最深的体会是:做 Agent,真正的难点从来不在"让它动起来",而在"让它在该停的时候停、在出错的时候兜住、在烧钱的时候被拦下、在答错的时候能被查清"。大模型给了我们一个前所未有的、能理解意图并自主行动的"大脑",但一个能上生产的系统,需要的远不止一个聪明的大脑——它需要骨骼(执行循环)、需要神经(可观测性)、需要免疫系统(容错与护栏)、需要体检(评估)。这些不性感的工程,才是把"AI 玩具"和"AI 产品"区分开的东西。

所以如果你也正要把一个惊艳的 Agent demo 往生产推,别急着加更多花哨的能力,先回答这几个朴素的问题:它会不会停不下来?工具挂了它会不会崩?它会不会偷偷烧光预算?出了事我能不能查清是哪一步、为什么?我改了提示词之后怎么知道是变好还是变差?把这几个问题逐一用工程手段答好,你的 Agent 才算真正长出了能在真实世界里活下去的筋骨——这,远比再接入十个新工具重要得多。

—— 别看了 · 2026