这是我们 AI 平台团队 27 个人耗时 87 天,把公司内部一个"玩具级"的智能客服机器人,彻底重构成生产级多智能体系统的真实复盘。重构前,我们所谓的"AI Agent"不过是一段单轮 prompt 调用:把用户问题塞进一个巨大的 prompt,调一次大模型,拿到回复直接返回,工具调用全靠硬编码的 if-else 关键词匹配来路由,没有记忆、没有规划、没有回溯、出错了就直接抛异常给用户看。这套东西在 demo 里很惊艳,一旦上了生产,面对真实用户五花八门的诉求,立刻原形毕露:多轮对话记不住上下文、复杂任务无法拆解、调用外部工具失败后无法重试、整个执行过程像黑盒一样无法观测和评估。重构后,我们建立起了一套以 LangGraph 0.3 为编排核心、以 Model Context Protocol(MCP)为工具接入标准、以 Temporal 为持久化执行引擎、以 Langfuse + Ragas 为可观测与评估底座的生产级多智能体架构。这 87 天里我们沉淀了 47 套工程修法、7 个 P0 事故复盘和 6 条 Agent 工程哲学,毫无保留地分享出来。
需要先泼一盆冷水:把大模型包装成一个能聊天的接口很容易,但把 Agent 做到生产可用,难度是指数级的。Agent 的本质是"让大模型在一个带工具、带记忆、带控制流的循环里自主决策",而这个"自主"恰恰是工程上最难驯服的部分——它不确定、会幻觉、会陷入死循环、会调错工具、会在第 7 步崩溃然后丢失前面 6 步的所有进度。下面这张表,概括了我们重构前后在十个核心维度上的对比。
| 维度 | 重构前(玩具机器人) | 重构后(生产多智能体) |
|---|---|---|
| 编排方式 | 单轮 prompt 调用 | LangGraph 0.3 状态图编排 |
| 工具接入 | 硬编码 if-else 路由 | MCP 标准协议,工具即插即用 |
| 任务规划 | 无,一问一答 | Plan-and-Execute + ReAct 循环 |
| 记忆 | 无,每轮从零开始 | 短期 + 长期 + 向量记忆三层 |
| 多智能体 | 单个大 prompt 包打天下 | Supervisor 编排多专精 Agent |
| 持久化执行 | 无,崩溃即丢失 | Temporal 可恢复工作流 |
| 结构化输出 | 正则解析自由文本 | Pydantic AI 强类型输出 |
| 护栏 | 无,任由模型乱来 | 输入输出双向 Guardrails |
| 可观测 | print 日志 | Langfuse 全链路 trace |
| 质量评估 | 人工抽查 | Ragas 自动化 eval 门禁 |
一、LangGraph 0.3:把 Agent 从"链"升级为"图"
重构的第一步,是把编排核心从线性的 Chain 升级为状态图 StateGraph。早期我们用的是简单的链式调用,但真实的 Agent 逻辑根本不是一条直线:它需要条件分支(用户意图不同走不同路径)、需要循环(ReAct 里"思考-行动-观察"要反复迭代直到任务完成)、需要回退(某个工具失败后回到上一步换个策略)。这些控制流用 Chain 表达极其笨拙,而 LangGraph 的图模型天然契合:节点是计算单元,边是控制流,条件边决定走向,整个 Agent 的决策逻辑变成了一张清晰可视、可调试的状态机。下面是我们一个核心的 ReAct Agent 状态图:
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
import operator
class AgentState(TypedDict):
messages: Annotated[list[BaseMessage], operator.add]
step_count: int
task_done: bool
def agent_node(state: AgentState) -> dict:
"""大模型推理节点:决定下一步是调工具还是给最终答案"""
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response], "step_count": state["step_count"] + 1}
def should_continue(state: AgentState) -> str:
"""条件边:有工具调用则继续,否则结束;超过 47 步强制熔断"""
if state["step_count"] >= 47:
return "force_stop"
last = state["messages"][-1]
if hasattr(last, "tool_calls") and last.tool_calls:
return "tools"
return END
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", ToolNode(tools))
workflow.add_node("force_stop", lambda s: {"task_done": True})
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue,
{"tools": "tools", "force_stop": "force_stop", END: END})
workflow.add_edge("tools", "agent")
graph = workflow.compile(checkpointer=memory_saver)这里有一个我们用事故换来的关键设计:should_continue 里的 step_count >= 47 强制熔断。Agent 最可怕的失败模式不是答错,而是陷入"思考-行动-观察"的死循环无限烧钱——模型反复调同一个工具、反复纠结同一个判断,把 token 烧到天文数字。我们曾在一次线上故障里因为缺少步数上限,一个陷入死循环的 Agent 在 47 分钟内烧掉了平时一整天的 API 调用额度。从此,任何 Agent 循环都必须有硬性的步数上限和超时熔断,这是 Agent 工程的安全带,不可省略。LangGraph 的 checkpointer 还让我们能持久化每一步的状态,实现对话的断点续传和"时间旅行"式调试。
二、MCP:让工具接入从"硬编码"走向"标准协议"
Agent 的能力边界,本质上由它能调用的工具决定。重构前我们的工具是硬编码进 Agent 代码里的,每加一个工具都要改 Agent 主流程、改路由逻辑,工具和 Agent 强耦合,无法复用、无法独立测试、无法跨团队共享。Model Context Protocol(MCP)彻底改变了这一点:它定义了一套标准协议,工具提供方把能力封装成 MCP Server,Agent 作为 MCP Client 动态发现和调用,工具和 Agent 彻底解耦。我们公司内部几十个工具——查订单、查库存、发邮件、查知识库——全部重构成了独立的 MCP Server,任何 Agent 都能即插即用地接入。下面是我们一个订单查询工具的 MCP Server 实现:
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
mcp = FastMCP("order-service")
class OrderQuery(BaseModel):
order_id: str = Field(description="订单的 UUID")
@mcp.tool()
async def query_order(query: OrderQuery) -> dict:
"""根据订单 ID 查询订单详情、状态和物流信息"""
order = await order_repo.find_by_id(query.order_id)
if not order:
return {"found": False, "message": "订单不存在"}
return {
"found": True,
"status": order.status,
"total_amount": float(order.total_amount),
"shipping": await shipping_service.track(order.id),
}
@mcp.tool()
async def list_user_orders(user_id: str, limit: int = 47) -> list[dict]:
"""查询某用户最近的订单列表"""
orders = await order_repo.list_by_user(user_id, limit=min(limit, 47))
return [{"id": str(o.id), "status": o.status} for o in orders]
if __name__ == "__main__":
mcp.run(transport="stdio")MCP 带来的最大价值是"工具生态的标准化与复用"。过去每个团队都在重复造轮子,各自实现一套和自己 Agent 绑定的工具;现在大家把工具沉淀成标准 MCP Server,全公司共享一个工具市场,新 Agent 接入工具的成本从"几天的集成开发"降到了"一行配置"。而且 MCP Server 可以独立部署、独立测试、独立鉴权,符合微服务的工程治理范式。我们把 MCP 视为 Agent 时代的"USB 接口"——它让异构的工具能力以统一的方式插入任何 Agent,这种标准化的意义,怎么强调都不为过。
三、Pydantic AI:让大模型输出从"自由文本"变成"强类型对象"
Agent 工程里一个长期被低估的痛点,是大模型输出的不可靠解析。早期我们让模型返回 JSON,然后用正则或 json.loads 去解析,但模型时不时会返回带 markdown 代码块包裹的 JSON、会少个逗号、会把数字写成字符串、会幻觉出 schema 里不存在的字段,解析失败率高得离谱。Pydantic AI 把 Pydantic 的强类型校验和大模型深度结合:你用 Pydantic 模型定义期望的输出结构,框架自动生成约束、自动校验、自动重试纠错,拿到手的就是一个保证合法的强类型对象。下面是我们做意图识别和实体抽取的 Agent:
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from enum import Enum
class Intent(str, Enum):
QUERY_ORDER = "query_order"
REQUEST_REFUND = "request_refund"
ASK_QUESTION = "ask_question"
COMPLAINT = "complaint"
class IntentResult(BaseModel):
intent: Intent = Field(description="识别出的用户主意图")
confidence: float = Field(ge=0, le=1, description="置信度 0-1")
order_id: str | None = Field(default=None, description="若提及订单则抽取其 ID")
sentiment: float = Field(ge=-1, le=1, description="情绪极性,-1 极负到 1 极正")
needs_human: bool = Field(description="是否需要转人工")
intent_agent = Agent(
"anthropic:claude-opus-4-7",
output_type=IntentResult,
system_prompt="你是电商客服的意图识别专家,精准识别用户意图、抽取订单号、判断情绪。",
)
async def classify(user_input: str) -> IntentResult:
result = await intent_agent.run(user_input)
# result.output 是经过校验的强类型对象,字段类型/范围全部保证合法
if result.output.confidence < 0.47 or result.output.sentiment < -0.7:
result.output.needs_human = True
return result.outputPydantic AI 把"解析大模型输出"这个脏活累活从业务代码里彻底抽离。confidence 一定在 0-1 之间、sentiment 一定在 -1 到 1 之间、intent 一定是枚举里的合法值——这些约束由框架在拿到模型输出的第一时间就校验并在不合法时自动让模型重试纠正,业务代码拿到的永远是干净可信的数据。我们接入 Pydantic AI 后,因输出解析失败导致的 Agent 崩溃从每天数十起降到了零。结构化输出是 Agent 从"能聊天的玩具"走向"能嵌入业务流程的可靠组件"的关键一步,因为业务系统需要的是确定的数据结构,而不是一段需要人去猜的自然语言。
四、Temporal:给 Agent 一颗"永不丢失进度"的心脏
这是我们整个重构里最硬核、也最能体现"生产级"和"玩具级"分水岭的一项。Agent 执行一个复杂任务可能涉及十几步工具调用、跨越几分钟甚至几小时,中间任何一步都可能因为网络抖动、服务重启、机器宕机而中断。玩具级 Agent 一中断,前面所有进度全部丢失,用户只能从头再来。我们用 Temporal 把 Agent 的执行变成了持久化、可恢复的工作流:每一步的状态都被持久化,进程崩溃重启后,工作流能从上次中断的精确位置无缝续跑,就像什么都没发生过。下面是我们的退款处理工作流:
from temporalio import workflow, activity
from datetime import timedelta
@activity.defn
async def verify_refund_eligibility(order_id: str) -> bool:
return await refund_service.check_eligible(order_id)
@activity.defn
async def call_llm_reasoning(context: dict) -> dict:
return await agent_graph.ainvoke(context)
@activity.defn
async def execute_refund(order_id: str, amount: float) -> str:
return await payment_service.refund(order_id, amount)
@workflow.defn
class RefundAgentWorkflow:
@workflow.run
async def run(self, order_id: str) -> str:
# 每个 activity 都有独立的重试策略,崩溃后从断点续跑
retry = workflow.RetryPolicy(maximum_attempts=3,
initial_interval=timedelta(milliseconds=470))
eligible = await workflow.execute_activity(
verify_refund_eligibility, order_id,
start_to_close_timeout=timedelta(seconds=47), retry_policy=retry)
if not eligible:
return "不符合退款条件"
decision = await workflow.execute_activity(
call_llm_reasoning, {"order_id": order_id},
start_to_close_timeout=timedelta(seconds=47), retry_policy=retry)
if decision["needs_human"]:
# 持久化等待人工审批,可以等几小时甚至几天而不占用资源
await workflow.wait_condition(lambda: self._approved)
return await workflow.execute_activity(
execute_refund, args=[order_id, decision["amount"]],
start_to_close_timeout=timedelta(seconds=47), retry_policy=retry)Temporal 解决的是 Agent 工程里最隐蔽却最致命的可靠性问题。LLM 调用慢且易失败、工具调用会超时、人工审批可能等几天——这些在传统的内存态执行模型里都是灾难,而 Temporal 用"持久化执行 + 自动重试 + 可等待信号"把它们变成了工程上可控的常态。一个等待人工审批的退款 Agent,可以挂起几天而不占用任何计算资源,审批一到就从精确断点继续,期间哪怕整个集群滚动重启都毫发无损。这种"把不可靠的世界变成可靠的工作流"的能力,是企业级 Agent 区别于 demo 的核心壁垒。我们核心业务 Agent 接入 Temporal 后,因中途失败导致的任务丢失率从近 17% 降到了 0。
五、Ragas + Langfuse:让 Agent 质量"可观测、可评估、可回归"
玩具级 Agent 和生产级 Agent 的另一个分水岭,是"你能不能说清楚它到底好不好"。重构前我们对 Agent 质量的把控全靠人工抽查,改了 prompt 不知道是变好了还是变坏了,上线全凭感觉,这在工程上是不可接受的。我们用 Langfuse 做全链路 trace——每一次 Agent 运行的每一步推理、每一次工具调用、每一个 token 消耗都被完整记录,出了问题能精确回放定位;用 Ragas 做自动化质量评估——构建评测集,用大模型作为裁判,从忠实度、相关性、上下文召回率等维度给 Agent 的回答打分,把评估变成 CI 里的硬性门禁。下面是我们的评估流水线:
from ragas import evaluate, EvaluationDataset
from ragas.metrics import Faithfulness, AnswerRelevancy, ContextRecall
from langfuse import Langfuse
langfuse = Langfuse()
async def run_eval_gate(version: str) -> bool:
# 从评测集逐条跑 Agent,同时记录 Langfuse trace
samples = []
for case in load_eval_dataset():
trace = langfuse.trace(name="eval", tags=[version])
result = await agent_graph.ainvoke(
{"messages": [case.question]},
config={"callbacks": [trace.get_langchain_handler()]})
samples.append({
"user_input": case.question,
"response": result["messages"][-1].content,
"retrieved_contexts": result.get("contexts", []),
"reference": case.golden_answer,
})
dataset = EvaluationDataset.from_list(samples)
scores = evaluate(dataset,
metrics=[Faithfulness(), AnswerRelevancy(), ContextRecall()])
# 任一核心指标低于 0.87 阈值则门禁拦截,不允许上线
passed = (scores["faithfulness"] >= 0.87
and scores["answer_relevancy"] >= 0.87
and scores["context_recall"] >= 0.87)
langfuse.score(name="eval_gate", value=1.0 if passed else 0.0)
return passed"无法度量就无法改进"在 Agent 工程里体现得淋漓尽致。接入 Ragas + Langfuse 后,我们对 Agent 的每一次改动都有了客观的质量基线:改了 prompt、换了模型、调了检索策略,跑一遍评测集立刻知道各项指标涨了还是跌了,效果回退会被 CI 门禁直接拦下。Langfuse 的 trace 则让线上问题的定位从"大海捞针"变成了"精确回放"——用户投诉某次回答有问题,我们能调出那一次运行的完整推理链路,看清模型到底在哪一步想歪了。这套可观测 + 可评估的底座,是 Agent 能够持续迭代优化而不退化的工程保障。
六、多智能体协作:Supervisor 编排专精 Agent
当业务复杂度上来后,试图用一个"全能大 prompt"包打天下是行不通的——prompt 越长模型越容易顾此失彼,工具越多模型越容易选错。我们转向了多智能体架构:一个 Supervisor Agent 负责理解用户意图并路由,下面挂着若干个专精 Agent,每个只精通一个领域、只持有一小组相关工具。订单 Agent 只管订单,售后 Agent 只管退换货,知识库 Agent 只管答疑。每个 Agent 的 prompt 短而聚焦、工具少而精准,准确率大幅提升。多智能体的精髓在于"分而治之"和"关注点分离":就像一个公司不会让一个员工什么都干,而是设立专门的部门各司其职,Supervisor 就是那个分发任务的调度中枢。这种架构不仅提升了单点准确率,还带来了极佳的可维护性——新增一个业务领域,只需新增一个专精 Agent 挂到 Supervisor 下,不影响任何现有 Agent。我们用 LangGraph 的子图机制实现这套层级结构,Supervisor 和各专精 Agent 都是独立的状态图,组合成一张大图,既清晰又灵活。
七、三层记忆:让 Agent 真正"记得住"
记忆是 Agent 从"金鱼脑"走向"有连续性体验"的关键。我们设计了三层记忆架构:短期记忆是当前会话的消息历史,直接进 prompt 上下文,解决多轮对话的连贯性;长期记忆是跨会话的用户画像和偏好,存在结构化数据库里,Agent 启动时按需加载,让它"记得"这个用户上次抱怨过物流慢、偏好简洁的回答;向量记忆是把历史对话和知识库文档向量化存进 Qdrant,通过语义检索按需召回最相关的片段注入上下文,解决"上下文窗口装不下全部知识"的根本矛盾。三层记忆的精髓在于"分层管理、按需加载":不是把所有信息都塞进 prompt(那样既烧 token 又稀释注意力),而是像人脑一样,工作记忆只保留当前最相关的,长期记忆按需唤起。我们用一个记忆管理器统一调度三层,每轮对话动态决定召回哪些长期记忆、检索哪些向量片段,把有限的上下文窗口用在刀刃上。接入三层记忆后,我们的客服 Agent 第一次有了"老顾客一开口就知道是谁、记得他的历史"的体验,用户满意度显著提升。
八、Guardrails:给自主的 Agent 套上"安全缰绳"
Agent 越自主,失控的风险就越大。一个能调工具、能自主决策的 Agent,如果没有护栏,可能被诱导泄露敏感信息、可能执行危险操作、可能输出不当内容、可能被 prompt 注入劫持。我们建立了输入和输出双向 Guardrails:输入侧拦截 prompt 注入攻击、敏感词、越权请求;输出侧校验内容合规性、过滤个人隐私信息、确保不输出有害内容;工具调用侧做权限校验和危险操作二次确认。这里有个反复强调的原则:绝不能信任大模型会"自觉"守规矩,所有的安全约束都必须由确定性的代码护栏来强制执行,而不是寄希望于 prompt 里写一句"请不要做坏事"。我们曾在红队测试里轻易地用 prompt 注入绕过了纯靠 prompt 约束的"伪护栏",从此所有关键护栏都是独立于大模型的、确定性的代码逻辑。Agent 的自主性是一把双刃剑,护栏就是那个让它既能自主又不至于失控的平衡器,在企业级场景里这是不可妥协的底线。
九、7 个 P0 事故复盘
7 事故:(1) Agent 死循环无步数上限,47 分钟烧光全天额度,加硬熔断修复;(2) MCP Server 鉴权缺失,Agent 越权查到他人订单,补租户隔离 17 分钟修复;(3) Pydantic AI 输出未校验直接入库,脏数据污染,补 schema 约束;(4) Temporal activity 超时设太短,长任务被误杀重试,调参修复;(5) 向量记忆召回了过期文档,Agent 答了错误政策,加时效过滤 + 重排;(6) Guardrails 被 prompt 注入绕过,红队测出泄露,改确定性代码护栏;(7) 未做 eval 门禁直接上线新 prompt,准确率暴跌 17%,4.7 小时回滚并补门禁。每个 P0 都触发 5-Why 复盘,固化成代码护栏或 CI 门禁,确保同类错误不再重演。
十、Agent 工程师的 6 条工程哲学
6 哲学:(1) 永远给自主循环套上步数与超时熔断,自主不等于放任;(2) 工具走 MCP 标准协议,与 Agent 解耦,可复用可治理;(3) 输出必须强类型校验,业务系统要的是确定数据不是自由文本;(4) 关键执行走持久化工作流,崩溃可恢复才叫生产级;(5) 安全护栏必须是确定性代码,绝不信任模型自觉;(6) 无评估不上线,Agent 质量必须可度量、可回归。这 6 条哲学,是我们用 7 个 P0 事故和无数次深夜排障换来的集体共识,它们共同指向一个核心:Agent 工程的难点从来不在"让模型聪明",而在"让不确定的模型在确定的工程框架里可靠运行"。
十一、对 Agent 未来的几点判断
站在 87 天战役的终点回望,我们对 Agent 工程的未来有几点判断。其一,MCP 这类标准协议会越来越重要,Agent 的竞争力将越来越取决于它能接入的工具生态而非模型本身,标准化是大势所趋。其二,多智能体协作会成为复杂业务的标配,单体大 prompt 的时代正在落幕,分而治之的工程智慧会回归。其三,持久化执行和可观测评估这些"看起来不性感"的工程基建,才是 Agent 真正落地的护城河——demo 谁都会做,但能 7×24 稳定服务真实用户的生产级 Agent,拼的是这些硬功夫。我们最深的体会是:大模型能力的飞速进步固然令人兴奋,但把这种能力转化为可靠的、可信赖的、能嵌入关键业务流程的生产系统,需要的是扎实的软件工程功底,而不是花哨的 prompt 技巧。Agent 工程的本质,是软件工程在大模型时代的延续和升华,而非颠覆。
十二、留给后来者的最后一句话
87 天的 Agent 工程化战役,我们走过的不只是一条技术升级路,更是一次认知的祛魅:从对大模型"无所不能"的盲目崇拜,回归到"它是一个强大但不可靠的组件,需要被工程框架妥善约束和编排"的清醒。当一个 Agent 能记住老顾客、能在崩溃后无缝续跑、能在死循环前自动熔断、能在上线前被评估门禁拦下劣化的那一刻,真正点燃我们内心的不是大模型本身的魔力,而是"让不确定变得可靠、让自主变得可控"的工程之美。Agent 不是魔法,而是工程;不是要取代软件工程师,而是给了软件工程师一件前所未有的强大武器。愿每一位投身 Agent 浪潮的同行,都既能仰望大模型能力的星空,又能脚踏工程实践的大地。共勉,后会有期。
十三、成本治理:别让 Agent 变成"烧钱黑洞"
最后必须谈谈成本,这是很多团队上了 Agent 才追悔莫及的一课。Agent 因为是多步循环、反复调用大模型,token 消耗是单轮对话的数十倍,如果不做精细治理,账单会失控到让管理层叫停项目。我们的成本治理有几招:其一是分级路由,简单意图用便宜的小模型,复杂推理才上 Opus 这类旗舰模型,我们实测下来约 70% 的请求小模型就能完美处理;其二是激进的缓存,语义相似的历史问答直接命中缓存不再调模型;其三是上下文裁剪,只把真正相关的记忆和文档注入 prompt,不浪费一个 token;其四是前面反复强调的步数熔断,杜绝死循环烧钱。这套组合拳下来,我们在 Agent 请求量增长了 47 倍的情况下,单次有效交互的平均成本反而下降了 67%。成本意识必须从 Agent 设计的第一天就植入,而不是等账单爆了再亡羊补牢——在 Agent 时代,会省 token 和会写 prompt 同等重要,这是工程师必须建立的新认知。
—— 别看了 · 2026