LangGraph 多 Agent 协作 token 成本暴涨 6 倍的 7 天复盘:5 个反模式与 6 套修法

2026 年 3 月,我们一个企业级 AI Agent 平台(LangGraph 0.2.50 + OpenAI GPT-5 + Anthropic Claude Sonnet 4.6 + 自研 Tool Hub 87 个工具 + 23 个业务 Agent + 日均 18 万次 Agent 会话、每会话平均 12 轮、上下文窗口 200K)从单 Agent 切换到"主从多 Agent 协作"架构第 11 天起,财务部门发出红色警报:OpenAI + Anthropic 月度账单从稳态 8.2 万美金暴涨到 47 万美金、单次会话 token 消耗中位数从 4800 飙到 38000、tool call 调用次数从平均 6 次飙到 47 次、用户投诉响应时间从 14 秒涨到 110 秒、9% 的会话陷入"Agent 自循环"无法退出。表面是"Agent 变多了所以贵",实际打开 LangSmith trace + Prometheus token metrics + 87 个 tool 的调用分布之后才定位到根因:多 Agent 之间没有共享上下文 cache、每个 sub-agent 重复拉全量历史消息、Tool call 结果没去重导致同样的 API 调用被发起 14 次、LangGraph cycle 检测阈值过宽导致 Agent 互相 ping-pong、prompt template 没用 Anthropic prompt caching、context window 突破 100K 后 cost-per-token 实际跳了 2 倍、没有 token budget 熔断,这是教科书级的"多 Agent 协作时 token 成本指数膨胀"故障。修复路径用共享 conversation state + tool call 结果 cache + LangGraph cycle 限制 + Anthropic prompt caching + context compression + token budget 三层熔断 + 模型路由(简单任务路由到 Haiku 4.5)6 套手段组合落地。本文复盘 7 天里的所有踩坑、五个反模式、六套修法以及最终沉淀的 13 条 AI Agent 系统工程纪律。

一、背景:为什么从单 Agent 切到多 Agent

这套 AI Agent 平台 2025 Q3 上线,单 Agent 架构(一个 LLM + 87 个 tool 全暴露),主要服务客服场景。2026 Q1 业务扩到"运营分析 / 数据查询 / 知识问答 / 合同审阅 / 客服 / 工单分流 / 销售助手 / BI 报表 / 文档生成"9 大类,单 Agent 实在覆盖不了——一个 prompt 塞 87 个 tool description 已经占 8000 token,加上 system prompt 总开销 12000 token,响应慢、tool 选错率高。架构组决定切多 Agent 协作:1 个 Coordinator(主管)+ 9 个 Domain Agent(每个领域专家)+ 共享 Tool Hub。Coordinator 接到用户请求后判断该路由到哪个 Domain Agent,Domain Agent 调用 Tool Hub 完成工作,结果回到 Coordinator 整合后返回用户。

团队规模 9 人:AI 工程师 4 人 + 后端 3 人 + SRE 2 人。技术栈 LangGraph 0.2.50 + LangChain 0.3 + Pydantic 2.10 + FastAPI 0.115 + Redis 7.4(用于会话状态)+ PostgreSQL 16(持久化对话)+ LangSmith(trace)+ Anthropic + OpenAI 双供应商。

二、故障时间线:7 天从切换到全量治理

Day1-3 多 Agent 架构灰度 20%,各项指标看起来正常(P95 响应 22 秒、错误率 0.8%)——但 token 账单还没回传,大家以为没问题。Day4 灰度 80%,运维上午看到 OpenAI dashboard 显示"昨日费用 1.4 万美金"(往日 0.27 万),立刻警觉。Day5 全量切到 100%,Day6 财务报警"账单已突破月度预算的 380%",CEO 直接喊停问"为什么 token 烧这么快"。Day6 - Day7 紧急复盘,SRE 拉 LangSmith trace 看,单个用户问"上海昨天销售额"这种简单问题,Coordinator 路由到 BI Agent,BI Agent 调用 query_db tool 一次,但 trace 显示整个会话累计 token 消耗 47000、tool call 12 次、Coordinator 与 BI Agent 之间消息往返 8 次。问题原因瞬间清晰:多 Agent 之间没有共享 state,Coordinator 每次向 BI Agent 发请求都把"完整历史 + 完整 user query + 87 个 tool description"塞进去,BI Agent 回复时又把"完整结果 + 完整 reasoning"塞回 Coordinator,一来一回多了 6-12 倍 token。

Day7 紧急上线 6 套手段:共享 state + tool cache + cycle 限制 + prompt caching + context compression + budget 熔断。Day8 全量监控 24 小时,账单回落到稳态 9.8 万美金/月(略高于单 Agent 时代 8.2 万,因为多 Agent 本身有协作开销,但完全可接受),P95 响应时间 21 秒、cycle 死锁 0、用户体验明显改善。

三、五个反模式

反模式 1:多 Agent 之间没有共享 conversation state,每次都传全量上下文

原始 LangGraph 代码:

from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic

def coordinator_node(state):
    user_msg = state["user_message"]
    history = state["history"]
    full_context = "\n".join([f"{m['role']}: {m['content']}" for m in history])
    response = coordinator_llm.invoke([
        SystemMessage(content=COORDINATOR_PROMPT),  # 4200 token
        HumanMessage(content=f"History:\n{full_context}\n\nUser: {user_msg}")
    ])
    target_agent = parse_routing(response.content)
    return {"target_agent": target_agent, "coordinator_response": response.content}

def bi_agent_node(state):
    user_msg = state["user_message"]
    history = state["history"]
    full_context = "\n".join([f"{m['role']}: {m['content']}" for m in history])
    response = bi_agent_llm.invoke([
        SystemMessage(content=BI_AGENT_PROMPT + ALL_BI_TOOL_DESCRIPTIONS),  # 6800 token
        HumanMessage(content=f"History:\n{full_context}\n\nUser: {user_msg}")
    ])
    return {"bi_response": response.content}

问题显而易见:每个 Agent 都重新组装一遍完整 history + user message,Coordinator 用一次、BI Agent 又用一次,完全没共享。9 个 Domain Agent 各自都这么干,token 翻 10 倍。

反模式 2:Tool call 结果没缓存,同一参数被调 14 次

@tool
def query_sales_db(date: str, region: str) -> str:
    """查询销售数据"""
    result = db.execute(f"SELECT * FROM sales WHERE date='{date}' AND region='{region}'")
    return json.dumps(result)

一个会话里用户问"上海昨天销售额",BI Agent 调一次 query_sales_db,返回结果后 LLM 想"我需要 confirm 这个数字",又调一次;Coordinator 想"我要 verify BI Agent 的结果",再调一次;最终回答用户时再调一次确认……一次会话调 14 次同样参数的 query_sales_db,数据库压力 + LLM token 都翻 14 倍。Tool call 没有任何 memoization 机制是多 Agent 系统的标准反模式。

反模式 3:LangGraph cycle 检测阈值过宽,Agent 互相 ping-pong

workflow = StateGraph(AgentState)
workflow.add_node("coordinator", coordinator_node)
workflow.add_node("bi_agent", bi_agent_node)
workflow.add_node("customer_agent", customer_agent_node)
workflow.add_edge("coordinator", "bi_agent")
workflow.add_edge("bi_agent", "coordinator")
workflow.add_conditional_edges(
    "coordinator",
    routing_fn,
    {"bi": "bi_agent", "customer": "customer_agent", "end": END}
)
# ❌ 没设 recursion_limit
graph = workflow.compile()

LangGraph 默认 recursion_limit=25,意味着 Coordinator 和 BI Agent 之间可以来回 25 次才被打断。9% 的会话(主要是模棱两可的用户问题)会陷入 ping-pong:Coordinator 觉得需要 BI Agent,BI Agent 觉得需要 Coordinator 澄清,Coordinator 又转给 BI Agent……循环 20 次才被 limit 拦住,每轮都消耗 4000+ token,单次会话累计 100000+ token。

反模式 4:Anthropic Claude prompt caching 没用

Anthropic 在 2024 年底推出 prompt caching,缓存命中价格降 90%。但我们的代码完全没用:

response = anthropic_client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=2000,
    system=BIG_SYSTEM_PROMPT,  # 12000 token,每次重发
    messages=conversation,
)

BIG_SYSTEM_PROMPT 12000 token 每次都全量发,实际可以通过 cache_control 缓存:

response = anthropic_client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=2000,
    system=[
        {"type": "text", "text": BIG_SYSTEM_PROMPT, "cache_control": {"type": "ephemeral"}}
    ],
    messages=conversation,
)

缓存命中后这 12000 token 只需要支付 1200 token 的费用。我们没用 prompt caching 等于每次会话多付 90% 的 system prompt 钱,18 万次会话 × 12000 token × 90% = 大量浪费。

反模式 5:context window 突破 100K 后没压缩,长会话越走越贵

Claude / GPT 在 context > 100K 时单 token 价格实际跳了 2 倍(模型架构原因 + provider 定价策略)。我们某些"运营分析"会话用户连续问 30 多轮,context 累计到 180K,后续每轮成本是正常的 4-6 倍。没有任何 context compression 策略(摘要、滑动窗口、关键信息抽取)就是放任 token 成本失控。

四、问题本质:多 Agent 是 token 放大器

核心矛盾是"多 Agent 协作的本质是'用更多 LLM 调用换取更专业的领域处理',但如果没有共享 state + cache + compression,每次协作会乘以 N 倍 token 成本"。单 Agent 时一次会话 token = T,多 Agent 协作时 token = T × (agents involved) × (rounds) × (no_cache_factor)。9 个 Domain Agent × 8 轮 × 6 倍无 cache = 432 倍?当然实际不会这么极端,因为不是每个 Agent 都参与每轮,但 6-12 倍放大是常态。

flowchart LR
    A[User 1 句话] --> B[Coordinator
12000 system prompt]
    B --> C[BI Agent
6800 system prompt]
    C --> D[Tool: query_db]
    D --> C2[BI Agent re-think
6800 system prompt]
    C2 --> B2[Coordinator confirm
12000 system prompt]
    B2 --> E[User]
    style B fill:#f99
    style C fill:#f99
    style C2 fill:#f99
    style B2 fill:#f99

[mermaid]
flowchart TD
    Start[多 Agent token 飙升?] --> Q1{shared state?}
    Q1 -->|否| Share[加 LangGraph shared state]
    Q1 -->|是| Q2{tool 结果 cache?}
    Q2 -->|否| Cache[加 tool result memoize]
    Q2 -->|是| Q3{cycle 限制?}
    Q3 -->|否| Limit[设 recursion_limit=8]
    Q3 -->|是| Q4{prompt cache?}
    Q4 -->|否| PCache[开 Anthropic cache_control]
    Q4 -->|是| Compress[context compression]
[/mermaid]

五、六套修法

修法 1:LangGraph 全局共享 state + 引用传递

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, MessagesState
from langgraph.checkpoint.postgres import PostgresSaver

class SharedAgentState(MessagesState):
    user_id: str
    session_id: str
    tool_results: dict  # 全局 tool cache,所有 agent 共享
    routing_history: list
    token_budget_remaining: int
    summary: str  # 滚动摘要,长会话替代全量历史

def coordinator_node(state: SharedAgentState):
    summary = state.get("summary") or "(no prior summary)"
    recent = state["messages"][-4:]  # 只用最近 4 条
    response = coordinator_llm.invoke([
        SystemMessage(content=COORDINATOR_PROMPT, cache_control={"type": "ephemeral"}),
        HumanMessage(content=f"Summary so far: {summary}\nRecent messages: {recent}")
    ])
    return {"messages": [AIMessage(content=response.content)]}

def bi_agent_node(state: SharedAgentState):
    summary = state.get("summary", "")
    recent = state["messages"][-3:]
    response = bi_agent_llm.invoke([
        SystemMessage(content=BI_AGENT_PROMPT, cache_control={"type": "ephemeral"}),
        HumanMessage(content=f"Summary: {summary}\nLatest: {recent[-1].content}")
    ])
    return {"messages": [AIMessage(content=response.content)]}

关键改进:每个 Agent 只看"摘要 + 最近 3-4 条消息",不重传完整 history;state 用 MessagesState 自动 reducer,Coordinator 写的 Agent 都能读到。token 单轮从 47000 降到 7800,降幅 84%。
修法 2:Tool call 结果 cache(Redis,会话级)
import hashlib, json, redis
r = redis.Redis(host="redis.internal", decode_responses=True)

def tool_cache(ttl=300):
    def decorator(fn):
        def wrapper(*args, **kwargs):
            session_id = kwargs.pop("session_id", "global")
            key_data = f"{fn.__name__}:{json.dumps(args, sort_keys=True)}:{json.dumps(kwargs, sort_keys=True)}"
            key = f"toolcache:{session_id}:{hashlib.md5(key_data.encode()).hexdigest()}"
            cached = r.get(key)
            if cached:
                return cached
            result = fn(*args, **kwargs)
            r.setex(key, ttl, result)
            return result
        return wrapper
    return decorator

@tool
@tool_cache(ttl=600)
def query_sales_db(date: str, region: str, session_id: str = None) -> str:
    """查询销售数据"""
    result = db.execute(f"SELECT * FROM sales WHERE date='{date}' AND region='{region}'")
    return json.dumps(result)

会话级 cache TTL 10 分钟,同会话同参数 tool call 只执行 1 次,后续 13 次直接命中 cache。query_db 类工具调用从 14 次降到 1 次,数据库压力同步下降 90%。
修法 3:LangGraph recursion_limit + 路由 history 检测
graph = workflow.compile(checkpointer=PostgresSaver(...))

def routing_fn(state):
    history = state.get("routing_history", [])
    # 检测 ping-pong: 相同两节点交替超过 3 次
    if len(history) >= 6 and history[-6::2] == history[-5::2]:
        return "end_with_escalation"
    return parse_target(state["messages"][-1].content)

# 调用时传 recursion_limit
result = graph.invoke(
    {"messages": [HumanMessage(content=user_msg)]},
    config={"recursion_limit": 8, "configurable": {"thread_id": session_id}}
)

recursion_limit=8 + ping-pong 检测,任何 Agent 互相 bouncing 超过 3 次自动 escalate 到人工或返回"无法处理"。9% 的死循环会话归零,token 浪费立刻消失。
修法 4:Anthropic prompt caching 全量改造
from anthropic import Anthropic
client = Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=2000,
    system=[
        {
            "type": "text",
            "text": COORDINATOR_BIG_SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"}
        },
        {
            "type": "text",
            "text": ALL_TOOL_DESCRIPTIONS,
            "cache_control": {"type": "ephemeral"}
        }
    ],
    messages=conversation
)

# 监控 cache 命中率
print(response.usage.cache_creation_input_tokens)
print(response.usage.cache_read_input_tokens)

把 system prompt 和 tool descriptions 都标 cache_control,命中后费用降 90%,我们 cache 命中率从 0 飙到 87%,这部分单项节省约 18 万美金/月。同样原理 OpenAI 的 prompt caching 也开了,GPT-5 cache 命中后省 50%。
修法 5:Context compression(滚动摘要)
SUMMARIZE_PROMPT = """请把以下对话历史压缩成一段摘要,保留:用户意图、关键事实、未解决问题、已调用工具及结果。控制在 300 token 以内。"""

def maybe_summarize(state: SharedAgentState):
    msgs = state["messages"]
    total_tokens = sum(estimate_tokens(m.content) for m in msgs)
    if total_tokens > 30000:  # 超过 30K 触发摘要
        recent = msgs[-6:]  # 保留最近 6 条
        old = msgs[:-6]
        summary_response = summary_llm.invoke([
            SystemMessage(content=SUMMARIZE_PROMPT),
            HumanMessage(content="\n".join([f"{m.type}: {m.content}" for m in old]))
        ])
        return {
            "summary": summary_response.content,
            "messages": [SystemMessage(content=f"[Summary] {summary_response.content}")] + recent
        }
    return {}

workflow.add_node("maybe_summarize", maybe_summarize)

超过 30K token 自动摘要前面的对话,保留最近 6 条原文。长会话从 180K context 压到 40K,成本降低 4-6 倍,响应速度也快很多。
修法 6:Token budget 三层熔断 + 模型路由
DAILY_BUDGET = 12000  # USD
SESSION_BUDGET = 0.5  # USD per session
REQUEST_BUDGET = 0.08  # USD per LLM call

def select_model(complexity: str, urgency: str):
    """模型路由:简单任务用便宜模型"""
    if complexity == "simple" and urgency != "high":
        return "claude-haiku-4-5"  # 1/10 价格
    if complexity == "medium":
        return "claude-sonnet-4-6"
    return "claude-opus-4-7"

class TokenBudgetGuard:
    def __init__(self):
        self.daily_spent = 0
        self.session_spent = defaultdict(float)

    def check(self, session_id: str, estimated_cost: float) -> bool:
        if self.daily_spent + estimated_cost > DAILY_BUDGET:
            raise BudgetExceeded("daily budget exhausted")
        if self.session_spent[session_id] + estimated_cost > SESSION_BUDGET:
            raise BudgetExceeded(f"session {session_id} budget exhausted")
        if estimated_cost > REQUEST_BUDGET:
            raise BudgetExceeded("single request too expensive")
        return True

    def record(self, session_id: str, actual_cost: float):
        self.daily_spent += actual_cost
        self.session_spent[session_id] += actual_cost

guard = TokenBudgetGuard()

三层 budget(请求级 / 会话级 / 日级)+ 模型路由,简单问题(如"今天几号"、"上海昨天销售额"这种简单查询)路由到 Haiku 4.5,只在复杂任务才用 Sonnet 4.6 / Opus 4.7。模型路由单项节省约 12 万美金/月。
六、性能与成本对比

七、13 条 AI Agent 系统工程纪律

1. 多 Agent 协作必须用 LangGraph 共享 state,严禁全量 history 传递;
2. 所有 tool 必须有结果 cache(会话级或全局,TTL 按业务);
3. LangGraph recursion_limit 必须设(推荐 ≤ 10),禁止默认 25;
4. 必须有 ping-pong 检测 + 自动 escalation,死循环立即中断;
5. Anthropic / OpenAI prompt caching 必须开,system + tool 都标 cache_control;
6. context > 30K 必须触发摘要压缩,长会话不能让上下文无限增长;
7. 必须有三层 token budget(请求 / 会话 / 日),触发即熔断;
8. 必须有模型路由(简单任务路由到 Haiku/Mini,复杂才用 Sonnet/Opus);
9. LangSmith trace 必须 100% 采样关键链路,异常会话立即告警;
10. 每个 Agent 的 system prompt + tool description 必须有 token 预算上限(建议 ≤ 8000);
11. tool call 必须有重试 + 超时(API 失败不能拖垮整个 Agent);
12. 跨 Agent 通信必须有 schema(Pydantic 强类型),避免乱传字段;
13. 新接入 Agent / Tool 必须经过"成本压测"(模拟 1000 会话评估月度成本影响)。

八、引申一:LangGraph vs LangChain vs AutoGen vs CrewAI 选型

我们选 LangGraph 因为 state machine 模型最适合"工作流 + 协作"组合场景。2026 年新项目可以直接看 claude-agent-sdk(2025 Q4 发布),原生 prompt caching + tool use 设计极致优雅。

九、引申二:Tool Hub 设计模式

87 个 tool 不能全暴露给每个 Agent。我们做了"分层暴露":(1) 基础工具(read_file、http_get、now)所有 Agent 都可调;(2) 领域工具按 Agent 类别注册(BI Agent 只能调 query_db / chart_gen,客服 Agent 只能调 ticket_search / kb_search);(3) 高危工具(send_email、execute_code、delete_record)需要二次确认。这种"权限分层"减少 tool description 总 token + 提高 LLM 选 tool 的准确率(选项少了选错率降低)。

十、引申三:LLM 输出结构化的几种方案

Agent 让 LLM 输出 JSON 时常见三种方式:(1) 普通 prompt 描述要 JSON + 后处理(脆弱);(2) JSON mode(GPT/Claude 原生支持);(3) function calling / tool use(最稳)。我们最终统一用 tool use,所有结构化输出都用 Pydantic schema 自动转 tool definition,LLM 通过调用 tool 的方式返回结构化结果。tool use 比 prompt 工程稳 10 倍,生产 Agent 必选。

十一、引申四:Agent 的可观测性体系

LangSmith trace 只是基础,我们额外加了 4 层:(1) Prometheus metrics(每 Agent / 每 tool 调用次数、延迟、错误率);(2) OpenTelemetry trace(跨 Agent + tool 全链路);(3) 业务指标(用户满意度、任务完成率);(4) 成本指标(per-session / per-user / per-feature 成本)。这 4 层 + LangSmith 让我们对每个 Agent 决策都"可解释、可追溯、可优化"。

十二、引申五:Prompt engineering 与 Agent 设计的边界

Prompt engineering 解决"如何让 LLM 做对单次任务",Agent 设计解决"如何让 LLM 处理多步工作流"。前者关注 prompt 模板、few-shot example、output format;后者关注 state machine、tool 编排、retry/escalation 策略。两者都要懂,但 Agent 设计才是企业级 LLM 应用的核心能力,2026 年这是最稀缺的工程岗位之一。

十三、引申六:Agent 的安全性挑战

多 Agent 系统的安全风险:(1) Prompt injection(用户输入诱导 Agent 越权);(2) Tool poisoning(恶意 tool 返回内容污染上下文);(3) 数据泄露(Agent 把敏感数据写到响应);(4) 资源耗尽(用户故意触发 token 风暴);(5) 越权调用(Agent 调用了不该调的 tool)。我们的对策:input sanitization + tool sandboxing + output filtering + budget 熔断 + RBAC tool permission。这 5 层防护必须全开。

十四、引申七:Agent 协作模式的几种典型

(1) Hierarchical(我们用的,Coordinator + Workers);(2) Sequential(顺序流水线,Agent A → B → C);(3) Parallel(多个 Agent 并行 + Reducer 合并);(4) Debate(两个 Agent 对辩,第三个仲裁);(5) Swarm(无中心,Agent 自由通信);(6) Hybrid(组合上述模式)。不同场景选不同模式,Hierarchical 是企业级最稳妥的起点。

十五、引申八:模型路由的工程化

路由逻辑可以是 rule-based(关键词、长度阈值)、ML-based(用 embedding + classifier)、LLM-based(用一个便宜 LLM 决定路由)。我们最终用 ML-based(fine-tuned BERT classifier,准确率 94%、延迟 8ms、成本几乎为 0)。路由本身不应该是性能瓶颈,选准方法是关键。

十六、引申九:Agent 评测与回归测试

传统软件单测验证"input → output",Agent 评测要验证"多步对话 → 任务完成"。我们用 LangSmith Datasets + 自研评测框架:(1) 100 条 golden 对话(各业务场景);(2) 每次部署前自动跑 golden,任务完成率 < 90% 阻断;(3) 月度对账(抽样真实用户对话,人工评分);(4) A/B 测试(线上 5% 用新版本,对比指标)。这套评测让我们升级 Agent 时心里有底。

十七、引申十:开源 LLM vs 闭源 LLM 的成本平衡

有些场景(简单分类、信息抽取)用开源 Llama 3.3 / Qwen 3 + 自部署也行,成本可以再降 10 倍。但开源 LLM 维护成本(GPU、运维、模型升级)对小团队不划算。我们的策略:核心 Agent 用闭源(质量保证),边缘任务(批量翻译、关键词提取)用开源自部署,综合 ROI 最高。

十八、引申十一:Streaming 与 用户体验

多 Agent 协作时如何 stream 给用户?Coordinator 第一步路由就给反馈("正在分析您的问题")、BI Agent 调 tool 时给反馈("正在查询数据库")、最终结果 stream 输出。LangGraph 支持 stream_mode="values" / "updates" / "messages",我们用 "messages" 模式 push 给前端 SSE。21 秒的总响应里,用户能持续看到进度,实际"感知响应时间"只有 1.5 秒。

十九、引申十二:LLM 提供商多供应商策略

不能 100% 依赖一家供应商,任何家挂掉(API 502 / 配额 / 区域故障)整个 Agent 平台就瘫。我们的多供应商策略:核心 Coordinator 用 Anthropic Claude Sonnet 4.6(首选)+ GPT-5(备选);BI / 复杂分析用 Claude Opus 4.7;简单任务用 Haiku 4.5 / GPT-5-mini;失败自动切供应商,SLO 99.95%。这套多供应商架构 2026 Q1 帮我们扛过了 Anthropic 一次 4 小时区域故障。

二十、引申十三:成本与 ROI

这次治理直接成本:AI 工程师 4 人 × 7 天 + 后端 3 人 × 4 天 + SRE 2 人 × 3 天,折算约 18 万人民币;直接收益:月度账单从 47 万美金降到 9.8 万美金,年度节省约 450 万美金 = 3200 万人民币,ROI 约 178 倍。AI Agent 平台的成本治理是 2026 年最 ROI 的工程投入之一,因为 LLM 账单天然指数膨胀,稍不留神就会爆。

二十一、引申十四:对 AI Agent 未来 2-3 年的展望

(1) Agent 标准化协议(Anthropic MCP、OpenAI Agents SDK、Google A2A)逐步统一,Agent 间通信将像 HTTP/REST 一样标准;(2) Token 价格继续下降(预期 2027 年再降 10 倍),但用户期望也会增高,成本博弈持续;(3) 多模态 Agent(文本 + 图像 + 视频)成为标配;(4) Agent + RAG + KG 三件套是企业知识 Agent 的标准栈;(5) On-device Agent(手机 / 边缘)崛起,延迟敏感场景下沉到本地。这五个方向是未来 Agent 工程师必须跟踪的趋势。

二十二、引申十五:团队能力建设

AI Agent 工程师不是"会调 API 的 Python 工程师",而是"懂 LLM 行为 + 懂分布式系统 + 懂工程纪律"的复合型人才。我们后续的人才培养路线:(1) 每月一次 Agent 内部分享(论文 + 实战);(2) 必读 Anthropic engineering blog / OpenAI cookbook;(3) 每季度一次"Agent 黑客松";(4) 鼓励参与 LangGraph / LangChain 开源贡献。一年下来团队整体水位从"会写 prompt"涨到"能设计 Agent 系统",企业竞争力质变。

总结

这 7 天踩坑让我最深刻的感受是:"AI Agent 是 token 成本的指数放大器,如果不在第一天就设计 budget + cache + compression,等账单飙起来就是 5-10 倍的烧钱"。47 万美金一个月听起来吓人,但其中 80% 都是"重复传 history + 重复 tool call + 没用 prompt cache"的纯浪费——这些浪费用 7 天工程治理就能消除,留下的 9.8 万美金才是真正的"业务价值开销"。

更深层次的反思是"AI Agent 工程化的核心不是 prompt,是系统设计"。Coordinator 决定何时调谁、Agent 共享 state 而不重复传、tool 结果可缓存、cycle 必须熔断、context 必须压缩、token 必须熔断、模型必须路由——这些都是经典分布式系统的思想,只是搬到了 LLM 时代。会写 prompt 的人很多,会设计 Agent 系统的人极少,这就是为什么 2026 年 AI 工程师岗位薪资是普通后端的 2-3 倍。

给所有正在做 AI Agent 平台的团队三条建议:(1) 第一天就接 LangSmith / OpenTelemetry trace,可观测性是 Agent 工程化的基础;(2) prompt caching + tool cache + recursion_limit 是多 Agent 系统的"三件套",缺一就是定时炸弹;(3) 模型路由 + token budget 是成本控制的最后一道防线,必须强制嵌入每个 Agent 节点。希望这篇 5000+ 字的复盘能让你少烧 30 万美金,欢迎在评论区交流你们的 Agent 实战经验。

二十三、引申十二:事件驱动(Event-Driven)Agent vs 请求-应答(Request-Reply)Agent

本平台 23 个 Agent 里有 19 个是请求式的,也就是用户来请求 Agent 一次再回应应答。但有的任务本不适合请求式:报业务负责用户最近七天的销售 Agent 每隔 5 分钟主动跑批分析、舆情监控 Agent 每隔 1 分钟感知数据库几条数据的变化。一开始我们粗暴用请求式 Agent + 外部定时器,每次调用都重新塞 38000 token 对话历史,成本暴涨也是本次事故来源之一。

真实策略是将这些后台任务切到 event-driven Agent 模式:Agent 作为 Kafka consumer 启动,收到事件后触发一次性会话(无人规定状态机,不需要维护 MessagesState),结果落小键到 Redis,成本直接降下 92%。这说明了思想:Agent 有状态化不是必然的。只有"用户对话类"的任务,才需要 MessagesState;定时性、事件触发的任务更适合"无状态、只回函数化"。2026 年的 AI Agent 架构开发,event-driven Agent 主要一面,跟 request-reply 一样重要。

二十四、引申十三:团队协作信息流改造

AI Agent 平台跨多个"Agent 间需直接协作的,是多团队接触面"。BI Agent 跟 Coordinator 是业务核心的对接面,改变 Coordinator 的路由逻辑时,要通知 BI 团队 review schema 变化;Tool Hub 里一个更新工具的对外 schema,要通知调用 Agent 探明一次,否则又来一篇 JSON 格式麻烦。我们的基础设施:跑 Slack 机器人话题 git hook 注入,Agent / Tool / Coordinator 改变备自动 @ 对应负责人。基础设施降低了下游意外的后续冲突率 71%。这说明 AI Agent 系统架构改变化,真正难计算的事是变更对齐机制。

二十五、引申十四:Agent 与应用分层(Application Boundary)

有一段我们的冲突:Agent 平台运算组与运算业务团队上做一个 Agent,谁来负责其边界与命名要紧?第一段的方法:Agent 平台只提供"Agent runtime + Tool Hub + LLM gateway + 观察网",业务 Agent 一律交业务团队在各自的 Git 仓库维护,在 CI 中注入 Agent registry。Agent 平台跑跑在 Kubernetes,业务 Agent 各自 deployment。这样压力交给了 6 次,业务 Agent 改变。这是我们将 Agent 平台作为很高的内容服务而前台团队。AI 平台跟其业务运行包络间要这种限制和基础设施,否则平台都将成为"Agent 命令 shop"。

二十六、引申十五:可观察性 Agent 的"金"

传统服务问发往里说的是几种指标:延迟、错误率、CPU、内存,媒体要为 Agent 系统加一些指标:(1) Agent token rate(token / s 每 Agent);(2) tool cache hit ratio(我目标 70%+);(3) recursion depth distribution(P95 < 6,P99 < 10);(4) Anthropic cache hit ratio(我目标 60%+);(5) summarization 触发数量和 token compression ratio(我目标 80%+);(6) 每会话 cost(P50 / P95);(7) 每 Agent cost(排行牢前十的 Agent);(8) 模型路由分配(Haiku / Sonnet / Opus 占比)。8 类指标结合传入测试关联跑业务 trace,Grafana 一行串起来,精确告诉跟平台什么时候要要爆了。

二十七、引申十六:LLM 调用层的成本归因(Cost Attribution)

事故的初期最让人崩溃的是"知道月账单暴涨,但不知道是哪个 Agent / 哪个用户 / 哪个功能在烧钱"。Anthropic 和 OpenAI 的账单只给一个汇总,粒度不到 API 调用级别。我们后来在 LLM gateway 层拦截所有调用,给每次 request 打上 (tenant_id, user_id, agent_name, session_id, tool_chain) 五元组标签,写到 ClickHouse,日均 1800 万行,压缩后 8GB。定位"哪 5% 的用户消费了 60% 的 token"用了不到 30 秒 SQL,定位"BI Agent 调用 search_documents 工具占了 23% 总成本"用了 1 分钟。没有 cost attribution,任何成本治理都是盲人摸象——这条经验比任何具体的代码 fix 都重要。

二十八、引申十七:Tool Hub 的版本管理与灰度

87 个 tool 都不是一蹴而就,每周新增 2-3 个、修改 5-7 个 schema 是常态。早期我们直接 hot-deploy,一次某个 tool 的 input schema 改了字段名,导致 9 个调用它的 Agent 在 LLM 端生成的参数全部 schema 不匹配,2 小时大面积报错。后来强制 Tool Hub 走语义版本号(search_documents@v1, search_documents@v2),Agent 在 prompt 里明确声明依赖版本,新版本灰度 5% 流量观察 24 小时再全量,有问题立刻回滚 v1。这个流程让我们半年内 0 次 tool 升级事故,可见 AI Agent 系统的"API 契约管理"比传统微服务还要严格——因为 LLM 不会像编译器一样在 schema 不匹配时抛错,它会默默猜一个参数然后跑出错误结果。

二十九、引申十八:Agent 失败模式的分类与应对

七天复盘里我们整理了多 Agent 系统的 6 类典型失败模式:(1) Ping-pong 循环(Agent A 和 B 互相调用对方)用 recursion_limit + 路径检测;(2) Tool 选错(LLM 选了不该用的工具)用 tool description 增强 + few-shot 示例;(3) 参数幻觉(LLM 编造了不存在的字段)用 structured output + JSON Schema 校验;(4) 上下文丢失(summarization 把关键信息压掉了)用 critical info 标记 + 不参与摘要;(5) 成本爆炸(token 失控)用三层 budget;(6) 模型降级失败(Haiku 处理不了任务但没回退到 Sonnet)用 confidence score + fallback。这 6 类覆盖了 95% 的实际故障,把每一类的检测 + 修复都做成 playbook,新人接手 Agent 平台运维的学习曲线从 3 个月压到 3 周。

三十、引申十九:AI Agent 时代的工程师角色重塑

这次事故让我重新理解了"AI Agent 工程师"的真正含义。它不是"写 prompt 的人",而是"设计一个跨越 LLM 推理、分布式状态机、流式 IO、成本计量、可观测性、灰度发布、容灾切换的复杂分布式系统的人"。LLM 只是其中一个组件,且是最不可靠的组件——它会幻觉、会循环、会超预算、会失败。工程师的核心工作是用"传统软件工程的严谨约束 LLM 的不确定性":用 schema 校验约束输出、用 cache 约束成本、用 recursion limit 约束循环、用 budget 约束失控、用 trace 约束盲区。把这些"约束"想清楚、做扎实,Agent 系统才能跑稳;只想着"换个更聪明的模型"或"调个更好的 prompt",永远走不出"demo 漂亮、生产爆炸"的怪圈。这是这次 7 天复盘最值得带走的认知。

—— 别看了 · 2026