2024 年我们给一家电商做智能客服 Agent 系统 需求是 用户问问题 Agent 自己判断要不要查订单 要不要查物流 要不要查退款政策 要不要转人工 一个 Agent 完成全链路。第一版我们用 LangChain 的 ReAct Agent 跑通 demo 给业务看 客户问"我上周买的耳机什么时候到"Agent 自动调 query_order + query_logistics 返回精确答案 老板看了直夸 AI 真是革命性的。但真上线一周就开始翻车 平均每天 30+ 投诉 有的 Agent 卡在 thought-action 循环 30 轮停不下来 有的 Agent 把"查订单"工具调成"取消订单"差点真退了客户的款 有的 Agent 上下文累积到 50K token 一次调用 5 块钱。然后我们陆续踩了一堆坑。第一种最让我傻眼 Agent 进入 thought-action-observation 循环 一直 thought 然后 action 然后 thought 然后 action 30 轮停不下来 我们没设 max_iterations 也没设环检测 一个 query 烧 20 块 token 钱。第二种最难缠 ReAct prompt 让 LLM 自己挑工具 但 LLM 偶尔会幻觉出一个不存在的工具名 比如 cancel_order_async 然后我们的 tool router 找不到工具直接抛异常 整个 Agent 崩。第三种最离谱 Agent 拿到工具返回的 JSON 它"消化"成自然语言时 偶尔会把"订单状态:已发货"理解成"订单状态:已取消"完全相反的意思 这是 LLM 处理结构化数据的著名幻觉。第四种最致命 多步骤任务上下文越累积越长 第 10 步时已经 30K token 不光慢还贵 而且 LLM 对早期步骤的关注度急剧下降经常忘了用户最初问的是什么。第五种最莫名其妙 多个 Agent 协作时(产品 Agent + 订单 Agent + 物流 Agent)一个 Agent 输出"订单号 ORD123"另一个 Agent 接到却变成"ORD-123"或"订单 ORD123"格式不统一下游 Agent 解析失败 全链路雪崩。我盯着这一连串问题想了很久才彻底想明白第一版错在一个根本的认知上我以为 Agent 就是 LLM + tools + 循环 简单 ReAct 模式就够了 可这个认知是错的真正能投产的 Agent 系统是一个 状态图编排 加 工具调用守护 加 错误恢复策略 加 上下文裁剪 加 多 Agent 通信契约 加 可观测性 的整套工程方法论 任何一环没做都可能让你的 Agent 从"智能助手"变成"烧钱黑洞"或"事故制造机"本文从头梳理 LangGraph 多 Agent 工程化的要点 状态图怎么搭 工具怎么调 错误怎么恢复 上下文怎么压 多 Agent 怎么协作 监控怎么做 以及一些把 Agent 用扎实要避开的工程坑
问题背景:为什么 ReAct Agent 总翻车
2023 年 ReAct 论文火了之后 每个团队都想搭 Agent。LangChain 的 AgentExecutor 让你 10 行代码就能跑通 demo 但 demo Agent 与生产 Agent 是两个量级。生产场景的坑主要来自:
- 循环失控:LLM 决定下一步动作 没有外部约束就会无限循环 烧钱无底洞。
- 工具调用幻觉:LLM 偶尔幻觉出不存在的工具名或参数 没有 validation 直接崩。
- 状态管理混乱:多步骤任务的中间状态散落在 prompt 里 难以追踪 难以恢复。
- 上下文累积爆炸:每轮把全部历史塞 prompt 第 10 轮 token 30K 慢且贵。
- 多 Agent 通信契约缺失:Agent 之间用自然语言传递 格式不统一下游解析失败。
- 可观测性差:Agent 链路长 出问题不知道哪一步错 难排查难优化。
一 用 LangGraph 替代 AgentExecutor:从循环到状态图
LangChain 的 AgentExecutor 是黑盒循环 出问题难调试。LangGraph 把 Agent 显式建模成有向图 每个节点是一个步骤 每条边是状态转移 整个流程可追踪可回放可暂停。这是从 demo 到生产的第一步。
# 1 LangGraph 基础 StateGraph
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator
# 1.1 定义状态(整个 Agent 共享的内存)
class AgentState(TypedDict):
messages: Annotated[List, operator.add] # 消息历史 自动累加
user_query: str
intent: str # 意图分类
tool_results: dict # 工具调用结果
final_answer: str
error: str
iter_count: int # 迭代计数 防循环
# 1.2 定义节点(每个节点是一个处理函数)
def classify_intent(state: AgentState):
"""分类用户意图"""
query = state["user_query"]
intent = llm.classify(query, options=["query_order", "query_logistics", "refund", "other"])
return {"intent": intent}
def query_order(state: AgentState):
"""查订单工具节点"""
result = order_service.query(state["user_query"])
return {"tool_results": {**state["tool_results"], "order": result}}
def query_logistics(state: AgentState):
"""查物流节点"""
result = logistics_service.query(state["tool_results"]["order"]["id"])
return {"tool_results": {**state["tool_results"], "logistics": result}}
def generate_answer(state: AgentState):
"""生成最终答案"""
answer = llm.generate_answer(state["user_query"], state["tool_results"])
return {"final_answer": answer}
def route_human(state: AgentState):
"""转人工"""
return {"final_answer": "已为您转接人工客服"}
# 1.3 构建图
workflow = StateGraph(AgentState)
workflow.add_node("classify", classify_intent)
workflow.add_node("query_order", query_order)
workflow.add_node("query_logistics", query_logistics)
workflow.add_node("generate", generate_answer)
workflow.add_node("human", route_human)
# 1.4 定义边(状态转移)
workflow.set_entry_point("classify")
def route_after_classify(state):
"""根据意图分发"""
intent = state["intent"]
if intent == "query_order":
return "query_order"
elif intent == "query_logistics":
return "query_order" # 物流也需要先查订单
elif intent == "refund":
return "human"
else:
return "generate"
workflow.add_conditional_edges("classify", route_after_classify, {
"query_order": "query_order",
"human": "human",
"generate": "generate"
})
workflow.add_edge("query_order", "query_logistics")
workflow.add_edge("query_logistics", "generate")
workflow.add_edge("generate", END)
workflow.add_edge("human", END)
# 1.5 编译执行
app = workflow.compile()
result = app.invoke({"user_query": "我上周的订单到哪了", "tool_results": {}, "iter_count": 0})
实战经验:LangGraph 把隐式的 Agent 循环变成显式的状态图 每一步骤都可视化可调试;状态共享通过 TypedDict 强类型约束 避免 dict 乱传;条件边让流程分支清晰 不需要 LLM 自己决定下一步。我们从 AgentExecutor 迁到 LangGraph 后 故障定位时间从 30 分钟降到 3 分钟。
二 工具调用守护:防幻觉防参数错
LLM 调用工具时偶尔会幻觉工具名 编造参数 传错类型 没有守护层直接传给后端就是事故。生产 Agent 必须在 tool router 这层做严格 validation 失败要优雅降级而不是抛异常。
# 1 严格的工具签名定义 用 Pydantic
from pydantic import BaseModel, Field, ValidationError
from typing import Literal
class QueryOrderArgs(BaseModel):
order_id: str = Field(..., pattern=r"^ORD\d{6,}$", description="订单号 格式 ORD开头+6位数字")
user_id: int = Field(..., gt=0)
class RefundOrderArgs(BaseModel):
order_id: str = Field(..., pattern=r"^ORD\d{6,}$")
reason: Literal["damaged", "wrong_item", "no_longer_wanted"]
amount: float = Field(..., gt=0, le=10000)
# 2 工具注册表 显式声明
TOOLS = {
"query_order": {
"func": query_order_impl,
"args_schema": QueryOrderArgs,
"description": "查询订单状态 输入订单号"
},
"query_logistics": {
"func": query_logistics_impl,
"args_schema": QueryLogisticsArgs,
"description": "查询物流信息"
}
}
# 3 安全的工具调用 带 validation
def safe_tool_call(tool_name: str, raw_args: dict):
"""带防护的工具调用"""
# 检查工具是否存在 防幻觉工具名
if tool_name not in TOOLS:
return {
"success": False,
"error": f"工具 {tool_name} 不存在 可用工具: {list(TOOLS.keys())}",
"fallback": "ask_llm_to_pick_again" # 让 LLM 重选
}
tool = TOOLS[tool_name]
# 验证参数 防类型错
try:
validated_args = tool["args_schema"](**raw_args)
except ValidationError as e:
return {
"success": False,
"error": f"参数错误: {e.errors()}",
"fallback": "ask_llm_to_fix_args"
}
# 调用工具 带超时
try:
result = tool["func"](**validated_args.dict(), timeout=10)
return {"success": True, "data": result}
except TimeoutError:
return {"success": False, "error": "工具调用超时", "fallback": "retry_or_skip"}
except Exception as e:
return {"success": False, "error": str(e), "fallback": "skip"}
# 4 重试与降级策略
def call_tool_with_retry(tool_name, args, max_retries=2):
for attempt in range(max_retries + 1):
result = safe_tool_call(tool_name, args)
if result["success"]:
return result
if result["fallback"] == "ask_llm_to_fix_args":
# 让 LLM 看错误信息修参数
args = llm.fix_args(tool_name, args, result["error"])
continue
if result["fallback"] == "retry_or_skip" and attempt < max_retries:
continue
break
return {"success": False, "data": None, "error": result["error"]}
# 5 危险操作必须 human-in-loop
def execute_action(tool_name, args, context):
"""敏感操作走人工审批"""
DANGEROUS_TOOLS = {"refund_order", "cancel_order", "update_user_data"}
if tool_name in DANGEROUS_TOOLS:
approval = request_human_approval(tool_name, args, context)
if not approval["approved"]:
return {"success": False, "error": "人工拒绝执行"}
return safe_tool_call(tool_name, args)
实战经验:工具参数必须 Pydantic schema 验证 LLM 写错参数立即拦截;不存在的工具名要给 LLM 列表让它重选 不能抛异常崩溃;敏感操作(退款 取消 删除)必须 human-in-loop 不能让 LLM 自动执行。我们曾经踩过 Agent 自动调 cancel_order 的坑 一个 prompt injection 攻击就能让 Agent 帮攻击者取消别人的订单。
三 循环控制:防失控烧钱
Agent 最大的成本黑洞是循环失控。LLM 决定下一步 如果没有外部约束 它可能进入"想-做-想-做"无限循环 一次对话烧 100 美元。生产 Agent 必须设多层熔断。
# 1 LangGraph 内置递归限制
app = workflow.compile()
result = app.invoke(
{"user_query": "..."},
config={"recursion_limit": 25} # 最多 25 个节点跳转
)
# 2 自定义迭代计数 + 熔断
def with_iter_guard(node_func, max_iter=10):
"""节点装饰器 加迭代限制"""
def wrapper(state):
iter_count = state.get("iter_count", 0)
if iter_count >= max_iter:
return {
"final_answer": "对不起 我处理这个问题花了太多时间 请联系人工客服",
"error": "max_iterations_exceeded"
}
result = node_func(state)
result["iter_count"] = iter_count + 1
return result
return wrapper
迭代次数熔断防止无限循环 但还有一类更隐蔽的成本失控 LLM 调用单步消耗 token 暴涨。同一个 Agent 任务 第 1 步可能 2K token 第 10 步可能 30K token 单纯计数熔断不够 必须 token 与 cost 双维度监控。
# 3 token 预算控制
class TokenBudgetTracker:
def __init__(self, max_tokens=50000, max_cost_usd=1.0):
self.tokens_used = 0
self.cost_used = 0
self.max_tokens = max_tokens
self.max_cost = max_cost_usd
def track(self, prompt_tokens, completion_tokens, model="gpt-4o"):
self.tokens_used += prompt_tokens + completion_tokens
# 按 model 计价
price = {"gpt-4o": (0.005, 0.015), "gpt-4o-mini": (0.00015, 0.0006)}
in_price, out_price = price[model]
self.cost_used += prompt_tokens / 1000 * in_price + completion_tokens / 1000 * out_price
def check(self):
if self.tokens_used > self.max_tokens:
raise BudgetExceeded(f"token 超预算 {self.tokens_used}")
if self.cost_used > self.max_cost:
raise BudgetExceeded(f"成本超预算 ${self.cost_used:.2f}")
# 4 环检测 防止真正的循环
def detect_loop(state, window=5):
"""检测最近 N 步是否在重复同样的动作"""
recent_actions = [m["action"] for m in state["messages"][-window:] if m.get("action")]
if len(set(recent_actions)) == 1 and len(recent_actions) == window:
return True
return False
# 5 总时长限制 防止单次请求卡几分钟
import time
class TimeoutGuard:
def __init__(self, max_seconds=60):
self.start = time.time()
self.max = max_seconds
def check(self):
if time.time() - self.start > self.max:
raise TimeoutError(f"Agent 执行超时 {self.max}s")
实战经验:LangGraph recursion_limit 是第一道防线 但不够细 必须加自定义迭代计数 + token 预算 + 时间预算多层熔断;环检测在 ReAct 模式很有用 Agent 经常陷入"再试一次相同工具"的死循环;熔断后必须给用户友好提示 不能直接 500。我们设的标准:max_iter=10 max_tokens=30K max_cost=$0.5 max_time=30s 任何一个超就熔断。
四 上下文管理:解决累积爆炸
多步 Agent 第 10 步上下文可能 30K token 不光慢还贵 而且 LLM 注意力开始涣散经常忘记用户最初的问题。上下文管理是 Agent 工程化的关键技能 不是简单 truncate 那么粗暴。
# 1 滑动窗口 + summary 双策略
class ContextManager:
def __init__(self, max_recent=5, summary_threshold=10):
self.max_recent = max_recent
self.summary_threshold = summary_threshold
def manage(self, messages):
"""超过阈值时压缩老消息为 summary"""
if len(messages) <= self.summary_threshold:
return messages
# 保留最近 N 条原文
recent = messages[-self.max_recent:]
# 老消息生成 summary
old = messages[:-self.max_recent]
summary = llm.summarize(old, max_tokens=500)
return [
{"role": "system", "content": f"以下是之前对话的摘要: {summary}"}
] + recent
# 2 关键信息提取并固定保留
class StickyContext:
"""关键信息永不被压缩"""
def __init__(self):
self.sticky = {} # user_id, order_id, original_query 等
def extract_and_stick(self, message):
"""从消息中抽取关键字段"""
import re
order_match = re.search(r'ORD\d{6,}', str(message))
if order_match:
self.sticky["order_id"] = order_match.group()
# ...
def get_sticky_prompt(self):
return "重要上下文(始终保留): " + json.dumps(self.sticky, ensure_ascii=False)
# 3 按消息类型分级保留
def smart_truncate(messages, budget_tokens=8000):
"""优先保留 system + 最新 user + 最近 tool results"""
result = []
used = 0
# 强制保留
must_keep = [m for m in messages if m["role"] == "system"]
must_keep += [messages[-1]] # 最新用户消息
for m in must_keep:
result.append(m)
used += count_tokens(m)
# 倒序填充 直到预算
for m in reversed(messages[:-1]):
if m["role"] == "system":
continue
t = count_tokens(m)
if used + t > budget_tokens:
break
result.insert(-1, m)
used += t
return result
滑动窗口 + summary 解决了消息历史的累积问题 但 Agent 工程化最大的上下文杀手其实是 tool results。一次 query_order 可能返回 5KB JSON 几次工具调用下来轻松 20K token 全是数据。把 tool result 从 LLM 上下文里搬出去 用 reference id 引用 是节省 token 的杀手锏。
# 4 tool results 单独缓存 不进 LLM 上下文
class ToolResultCache:
"""工具结果存外部 Agent 用 reference id 引用"""
def __init__(self):
self.cache = {}
def store(self, result):
ref = f"tool_result_{uuid.uuid4().hex[:8]}"
self.cache[ref] = result
return ref
def get_summary_for_llm(self, ref):
"""给 LLM 只看摘要 详情用 reference 取"""
result = self.cache[ref]
return {
"ref": ref,
"summary": llm.summarize(result, max_tokens=200),
"schema": describe_schema(result)
}
# 5 上下文 token 实时监控
def with_context_monitor(state):
total_tokens = sum(count_tokens(m) for m in state["messages"])
if total_tokens > 20000:
state["messages"] = ContextManager().manage(state["messages"])
return state
实战经验:简单 truncate 会丢关键信息 必须 sticky context 固定订单号 用户 ID;长 tool result(几 KB JSON)单独存外部 LLM 只看 summary + reference id 节省 70% token;summary 阈值设 10 轮以上 太频繁压缩反而丢失上下文。我们用这套上下文管理后 单次对话平均 token 从 25K 降到 8K 成本降 70%。
[mermaid]
flowchart TD
A[用户输入] --> B[ContextManager 检查]
B --> C{token > 20K}
C -->|是| D[滑动窗口 + 老消息 summary]
C -->|否| E[保留原样]
D --> F[StickyContext 抽取关键字段]
E --> F
F --> G[LLM 决策下一步]
G --> H{动作类型}
H -->|tool call| I[safe_tool_call 验证]
H -->|finish| J[返回最终答案]
I --> K{success}
K -->|是| L[结果存 ToolResultCache]
K -->|否| M[fallback 让 LLM 修正]
L --> N[更新 state iter+1]
M --> N
N --> O{达到熔断}
O -->|是| P[friendly error + 转人工]
O -->|否| B
五 多 Agent 协作:契约式通信
单 Agent 能力有限 复杂业务必须多 Agent 协作。产品 Agent 负责选品 订单 Agent 负责下单 客服 Agent 负责答疑 三个 Agent 之间必须有严格的数据契约 否则一个 Agent 输出"订单号 ORD123"另一个解析成"ORD-123"全链路雪崩。
# 1 Supervisor 模式:中央调度多 Agent
from langgraph.graph import StateGraph
from typing import Literal
class TeamState(TypedDict):
messages: list
next_agent: str
artifacts: dict # 各 Agent 的结构化产出
def supervisor(state):
"""决定下一个调用哪个 Agent"""
prompt = f"""根据对话历史 决定下一步调用哪个 Agent
可选: product_agent order_agent service_agent finish
历史: {state['messages']}
artifacts: {state['artifacts']}"""
next_agent = llm.decide(prompt, options=["product_agent", "order_agent", "service_agent", "finish"])
return {"next_agent": next_agent}
# 2 结构化输出契约 每个 Agent 必须返回 schema
class ProductRecommendation(BaseModel):
"""产品 Agent 必须返回这个 schema"""
products: List[dict] = Field(..., description="推荐的产品列表")
reason: str
confidence: float = Field(ge=0, le=1)
class OrderInfo(BaseModel):
"""订单 Agent 必须返回这个 schema"""
order_id: str = Field(..., pattern=r"^ORD\d{6,}$")
status: Literal["pending", "paid", "shipped", "delivered", "cancelled"]
amount: float
items: List[dict]
def product_agent(state: TeamState):
"""产品 Agent"""
response = llm.generate_structured(
prompt=build_prompt(state),
schema=ProductRecommendation # 强制 JSON schema
)
return {
"artifacts": {**state["artifacts"], "products": response.dict()},
"messages": state["messages"] + [{"role": "product_agent", "content": response.json()}]
}
# 3 Agent 间消息格式标准化
class AgentMessage(BaseModel):
"""所有 Agent 间消息的标准格式"""
from_agent: str
to_agent: str
intent: Literal["request", "response", "broadcast"]
payload: dict
refs: List[str] = [] # 引用之前的 artifact id
timestamp: float
# 4 Agent 间冲突解决
def merge_agent_outputs(outputs: List[dict]):
"""多个 Agent 同时输出时合并 冲突时投票"""
merged = {}
for output in outputs:
for key, value in output.items():
if key not in merged:
merged[key] = [value]
else:
merged[key].append(value)
# 投票决定
final = {}
for key, values in merged.items():
if len(set(map(str, values))) == 1:
final[key] = values[0]
else:
# 冲突 让 supervisor 决定
final[key] = supervisor_decide(key, values)
return final
# 5 Handoff 显式交接
def handoff(from_agent: str, to_agent: str, payload: dict, reason: str):
"""显式交接 留痕"""
handoff_log.append({
"from": from_agent,
"to": to_agent,
"payload": payload,
"reason": reason,
"timestamp": time.time()
})
return {"next_agent": to_agent, "handoff_payload": payload}
实战经验:多 Agent 通信必须用结构化 schema 别让 LLM 用自然语言传字段格式必崩;Supervisor 模式比 free-form 协作好调试 一个中央调度让 trace 清晰;handoff 必须留痕方便排查到底哪个 Agent 出错。我们把客服系统从 ReAct 单 Agent 改成 Supervisor + 3 个专业 Agent 后 准确率从 75% 升到 92% trace 清晰多了。
六 可观测性:让 Agent 黑盒变白盒
Agent 链路长 节点多 每步都可能出错 没有可观测性等于盲飞。生产 Agent 必须有 tracing + metrics + logging 三位一体 出问题 1 分钟定位。
# 1 LangSmith / LangFuse 集成
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__..." # 从 env 读
os.environ["LANGCHAIN_PROJECT"] = "ecommerce-agent-prod"
# 每次 invoke 自动上报到 LangSmith
result = app.invoke({"user_query": "..."})
# 在 LangSmith UI 看完整 trace 每个节点 input output 耗时
# 2 自定义 OpenTelemetry trace
from opentelemetry import trace
tracer = trace.get_tracer(__name__)
def traced_node(node_func):
def wrapper(state):
with tracer.start_as_current_span(node_func.__name__) as span:
span.set_attribute("user_query", state.get("user_query", "")[:100])
try:
result = node_func(state)
span.set_attribute("success", True)
return result
except Exception as e:
span.set_attribute("success", False)
span.set_attribute("error", str(e))
raise
return wrapper
# 3 关键指标采集
class AgentMetrics:
def __init__(self):
self.histogram_latency = Histogram("agent_latency_seconds", ["node"])
self.counter_iter = Counter("agent_iterations", ["status"])
self.counter_tokens = Counter("agent_tokens", ["model", "type"])
self.histogram_cost = Histogram("agent_cost_usd")
def record_invocation(self, state, latency, cost):
self.histogram_latency.labels(node="total").observe(latency)
self.counter_iter.labels(status=state.get("error", "success")).inc()
self.histogram_cost.observe(cost)
# 4 关键事件日志结构化
import structlog
log = structlog.get_logger()
def log_agent_event(event_type, **kwargs):
log.info(
event_type,
trace_id=trace.get_current_span().get_span_context().trace_id,
**kwargs
)
# 例如:
log_agent_event("tool_call", tool="query_order", args={"order_id": "ORD123"})
log_agent_event("llm_call", model="gpt-4o", prompt_tokens=1500, completion_tokens=300)
log_agent_event("circuit_breaker", reason="max_iter_exceeded", iter_count=10)
# 5 异常检测告警
class AnomalyDetector:
def __init__(self):
self.baseline_latency_p99 = 8.0 # 秒
self.baseline_iter_avg = 3.5
self.baseline_cost_avg = 0.05
def check(self, metrics):
if metrics["latency"] > self.baseline_latency_p99 * 2:
alert("Agent latency 2x baseline")
if metrics["iter"] > self.baseline_iter_avg * 3:
alert("Agent iteration 3x baseline 可能循环失控")
if metrics["cost"] > self.baseline_cost_avg * 5:
alert("Agent cost 5x baseline 可能 token 爆炸")
# 6 回放与调试
def replay_session(session_id):
"""从 trace 还原一次 Agent 会话 用于复现 bug"""
trace_data = langfuse.get_trace(session_id)
initial_state = trace_data["initial_state"]
# 设置 LLM 回放模式 返回原 trace 中的 LLM 输出
with replay_mode(trace_data["llm_calls"]):
result = app.invoke(initial_state)
return result
实战经验:LangSmith / LangFuse 必接 不接基本盲飞;trace 必须能从用户输入一路追到每个工具调用 每个 LLM 调用;metric 关注 latency / iter / cost 三个核心指标 baseline 比对自动告警;replay 是定位疑难 bug 的神器 客户投诉时直接回放会话 5 分钟定位问题。
关键概念速查
| 概念 | 说明 | 推荐 | 备注 |
|---|---|---|---|
| LangGraph | 状态图编排 | 替代 AgentExecutor | 可调试可回放 |
| StateGraph | 共享状态 | TypedDict 强类型 | 避免 dict 乱传 |
| tool 守护 | 参数验证 | Pydantic schema | 防幻觉防错 |
| 循环熔断 | 多层防护 | iter+token+time | 防烧钱黑洞 |
| 上下文压缩 | summary+sticky | 10 轮触发 | 省 70% token |
| ToolResultCache | 结果外置 | ref id 引用 | 不占 LLM 上下文 |
| Supervisor 模式 | 多 Agent 调度 | 优于 free-form | trace 清晰 |
| 结构化契约 | Agent 间通信 | Pydantic schema | 防格式不统一 |
| human-in-loop | 敏感操作审批 | 必加 | 防误操作 |
| LangSmith | tracing 平台 | 必接 | 可观测性核心 |
避坑清单
- 不要用 AgentExecutor 跑生产 必须 LangGraph 显式状态图 可调试可回放可暂停。
- 不要让 LLM 自由选工具 必须 Pydantic schema 验证参数 不存在工具要列表让重选不能抛异常。
- 不要不设循环熔断 必须 iter + token + cost + time 四层防护 任一超就停。
- 不要每轮都把全部历史塞 prompt 必须滑动窗口 + summary + sticky context 长上下文必爆炸。
- 不要 tool result 直接塞上下文 必须外部 cache + reference id 单 result 几 KB JSON 太浪费。
- 不要敏感操作让 Agent 自动执行 退款 取消 删除必须 human-in-loop 否则 prompt injection 就崩。
- 不要多 Agent 自然语言通信 必须 Pydantic 契约 不然格式不统一下游解析必崩。
- 不要多 Agent free-form 协作 必须 Supervisor 模式中央调度 trace 才清晰。
- 不要不接 LangSmith/LangFuse 盲飞 出问题排查时间从 5 分钟变 5 小时。
- 不要忽视 baseline 告警 latency 2x iter 3x cost 5x 偏离基线必有问题立即报警。
总结
把 LangGraph 多 Agent 这套从我们踩过的所有坑里反过来看 你会发现真正影响 Agent 投产成败的不是 LLM 能力 而是工程化的全栈能力。同样一个 GPT-4o ReAct AgentExecutor 模式循环失控烧钱 LangGraph 状态图编排 + 多层熔断 + 工具守护 + 上下文压缩稳如老狗;同样一组工具 自然语言传递格式必崩 Pydantic 契约 + Supervisor 调度精准协作。Agent 不是 LLM + tools + while loop 几行代码的玩具 它是一个 状态图编排 + 工具守护 + 错误恢复 + 上下文管理 + 多 Agent 协作 + 可观测性 的完整系统工程。
另一个常见的认知误区是把 Agent 当成自主智能体 觉得 LLM 自己能搞定一切 我们工程师只要"给个目标"它就能完成。但事实是 LLM 是工具 不是引擎 真正驱动 Agent 投产的是工程方法论 是 schema 约束 是熔断防护 是契约通信 是可观测性。LLM 的能力上限决定 Agent 的天花板 但工程化的扎实程度决定 Agent 能不能落地。
打个比方 LangGraph Agent 像一个外包客服中心。StateGraph 是工作流系统(谁负责哪步) 工具守护是合同审核员(参数对不对) 循环熔断是值班经理(防新员工死磕一个客户) 上下文压缩是会议纪要(每轮总结防遗忘) Supervisor 是接待主管(分发问题给专业组) 结构化契约是工单系统(各部门统一格式) human-in-loop 是法务复核(敏感操作必审) LangSmith 是质检中心(每通电话录音回放)。哪一项缺了 这个客服中心就有事故 要么响应慢 要么交付错 要么成本爆 要么客户投诉。
所以下一次再有人跟你说 Agent 就是 LLM + tools + 循环 你可以反问他 状态图建了吗 工具 schema 验了吗 熔断设了吗 上下文压了吗 多 Agent 契约定了吗 human-in-loop 加了吗 LangSmith 接了吗 baseline 告警设了吗 这些工作没做完 Agent 只是一个能跑通 demo 的玩具 不是一个能扛业务的智能系统。从踩坑到投产 中间隔着一整套工程方法论 这条路没有捷径 但走完之后 你的 Agent 系统会从烧钱黑洞变成业务利器 从每天 30 投诉变成准确率 92% 从盲飞排查 5 小时变成 trace 定位 5 分钟。
—— 别看了 · 2026