LangChain Agent 工程化完全指南:从一次"Agent 死循环烧了几百美元 token"看懂为什么 demo 时聪明的 Agent 上线就崩

2024 年我做了一个内部的 AI 助手项目核心需求是让模型能调一些公司内部的工具查 CRM 看订单生成报表执行 SQL 我第一反应是这事现在用 LangChain Agent 一周就能搞定 ReAct 框架自动会决定调哪个工具调几次最后输出答案给用户原型确实一周做完测了几个 case 模型行为很聪明老板看了还挺满意可等真把这套面向几百个内部用户开放一串问题冒了出来第一种最先把我打懵某天用户问帮我看一下昨天的销售汇总模型啰嗦地调用了 8 次工具 SQL 跑了 5 次 CRM 查了 3 次最后给出了一段我看半天没看懂的描述用户也看不懂结账时被 OpenAI token 账单震了一下第二种最难缠某些用户的请求模型死活进入死循环我看 trace 模型在 ReAct 里不停 thought 加 action 加 observation 同一个工具反复调几十次到 max iterations 才停下来用户等了 3 分钟看到的是一段道歉错误第三种最离谱模型调工具时把参数搞错了把销售部门传成了销售导致工具报错模型看到报错以为是工具的问题又调了一次还是错的来回十几次第四种最莫名其妙我换了一个更便宜的模型 GPT 3.5 跑同样的 prompt 工具调用准确率掉了 60% 我才意识到 Agent 对模型能力的依赖比纯对话场景敏感得多第五种最致命有用户故意问帮我把所有用户的密码导出模型很贴心地调了 SQL 工具直接执行了 select 我吓出一身冷汗才知道 Agent 必须做工具调用的权限边界我盯着这一连串问题想了很久才彻底想明白第一版错在一个根本的认知上我以为 LangChain Agent 就是 LLM 加一堆工具描述 Agent 自己会用得很好可这个认知是错的真正能在生产用的 Agent 系统是一个工具粒度设计加 Prompt 工程加调用次数控制加失败重试加权限隔离加可观测的工程系统少任何一环都会爆炸本文从头梳理工具应该怎么拆 ReAct vs Function Calling 怎么选调用次数和成本怎么控制失败处理怎么做权限边界怎么设以及一些把 Agent 做扎实要避开的工程坑

2024 年我做了一个内部的 AI 助手项目核心需求是让模型能调一些公司内部的工具查 CRM 看订单生成报表执行 SQL 我第一反应是这事现在用 LangChain Agent 一周就能搞定 ReAct 框架自动会决定调哪个工具调几次最后输出答案给用户原型确实一周做完测了几个 case 模型行为很聪明老板看了还挺满意可等真把这套面向几百个内部用户开放一串问题冒了出来第一种最先把我打懵某天用户问"帮我看一下昨天的销售汇总"模型啰嗦地调用了 8 次工具 SQL 跑了 5 次 CRM 查了 3 次最后给出了一段我看半天没看懂的描述用户也看不懂结账时被 OpenAI token 账单震了一下第二种最难缠某些用户的请求模型死活进入死循环我看 trace 模型在 ReAct 里不停 thought + action + observation 同一个工具反复调几十次到 max_iterations 才停下来用户等了 3 分钟看到的是一段道歉错误第三种最离谱模型调工具时把参数搞错了把"销售部门"传成了"销售"导致工具报错模型看到报错以为是工具的问题又调了一次还是错的来回十几次第四种最莫名其妙我换了一个更便宜的模型 GPT-3.5 跑同样的 prompt 工具调用准确率掉了 60% 我才意识到 Agent 对模型能力的依赖比纯对话场景敏感得多第五种最致命有用户故意问"帮我把所有用户的密码导出"模型很贴心地调了 SQL 工具直接执行了 select 我吓出一身冷汗才知道 Agent 必须做工具调用的权限边界我盯着这一连串问题想了很久才彻底想明白第一版错在一个根本的认知上我以为 LangChain Agent 就是 LLM 加一堆工具描述 Agent 自己会用得很好可这个认知是错的真正能在生产用的 Agent 系统是一个"工具粒度设计 + Prompt 工程 + 调用次数控制 + 失败重试 + 权限隔离 + 可观测"的工程系统少任何一环都会爆炸本文从头梳理工具应该怎么拆 ReAct vs Function Calling 怎么选调用次数和成本怎么控制失败处理怎么做权限边界怎么设以及一些把 Agent 做扎实要避开的工程坑

问题背景

Agent 是当下 LLM 应用里最热但也最难做扎实的形态之一。它的承诺很迷人——给模型一堆工具,模型能自己决定先调什么后调什么,完成多步任务。但真正把 Agent 推到生产用户面前,你会发现几乎所有 demo 时的"丝滑表现"都在真实场景里崩坏。常见的几类失败模式:

  • 调用次数失控:一个简单请求模型啰嗦地调了十几次工具,token 烧爆,延迟飙到分钟级。
  • 死循环:模型在 ReAct 里反复调同一个工具,直到 max_iterations 才停,用户等到天荒地老。
  • 参数错位:模型把"销售部门"传成"销售",工具报错,模型不会修正只会重试。
  • 幻觉调用:模型调了一个根本不存在的工具,或者编造了不存在的参数。
  • 权限缺失:用户问什么模型就调什么,没有"哪些工具允许调"的边界,容易出大事故。
  • 失败无重试:工具一次失败 Agent 整条链就崩了,没有降级和兜底。
  • 观测不到:用户说"它今天答得不对",你打开日志只有最终输出,中间的 thought/action 全没存。

一、工具设计:Agent 性能的天花板

Agent 的能力上限是被"工具设计"锁死的。模型能否选对工具、能否传对参数、能否少调几次,完全取决于工具的粒度、命名、参数描述写得好不好。大多数团队第一版工具设计得很差,然后把效果不好归因到"模型不够聪明"——实际上换模型也救不了。

工具设计的三个核心原则:第一,粒度恰当。工具粒度太粗(一个"万能查询"工具),模型不知道用它能干啥;粒度太细(每个 SQL 字段一个工具),模型组合调用次数爆炸。理想的粒度是"每个工具完成一件人类能用一句话描述的事"。第二,命名直观。工具名要让模型一看就知道这个工具干啥,不要用 query_v3 这种代号。第三,参数描述详尽。每个参数说明用途、类型、合法范围、示例,模型才知道怎么填。

# 反例:粒度太粗 + 描述太简
{
    "name": "query",
    "description": "查询数据",
    "parameters": {
        "type": "object",
        "properties": {
            "type": {"type": "string"},
            "params": {"type": "object"}
        }
    }
}
# 问题:模型不知道 type 能填什么, params 也不知道结构, 必然乱填

# 正例:粒度恰当 + 描述详尽
{
    "name": "get_daily_sales_summary",
    "description": (
        "获取指定日期的销售汇总数据, 包括总销售额、订单数、客单价、退款金额。"
        "用于回答'昨天/上周销售情况'类问题, 不要用于查询单个订单详情。"
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "date": {
                "type": "string",
                "description": "日期, 格式 YYYY-MM-DD, 例如 2026-05-23",
                "pattern": "^\\d{4}-\\d{2}-\\d{2}$"
            },
            "department": {
                "type": "string",
                "description": "部门, 可选值: sales_north / sales_south / sales_east / sales_west / all",
                "enum": ["sales_north", "sales_south", "sales_east", "sales_west", "all"]
            },
            "include_refunds": {
                "type": "boolean",
                "description": "是否在销售额中扣除退款, 默认 false",
                "default": False
            }
        },
        "required": ["date"]
    }
}
# 优势:模型知道 date 格式, 知道 department 只有 5 个合法值,
# 不会乱编参数, 也不会用这个工具查其他类型数据

另一个关键设计是"工具数量要控制"。OpenAI 的 Function Calling 在 10-15 个工具内表现稳定,超过 20 个工具召回准确率明显下降。所以不要把"所有可能用到的功能"都塞进 Agent,而是"按场景分流",不同场景的 Agent 配不同的工具子集。比如客服场景的 Agent 配 5 个查询类工具,运营场景的 Agent 配 5 个报表类工具,而不是给一个 Agent 配 50 个工具让它自己挑。

from langchain.tools import StructuredTool
from langchain.agents import AgentExecutor, create_openai_tools_agent

# 按场景拆 Agent
def build_customer_service_agent():
    tools = [
        StructuredTool.from_function(get_order_status,
            description="按订单 ID 查询订单状态"),
        StructuredTool.from_function(get_user_orders,
            description="按用户 ID 查询该用户最近 N 个订单"),
        StructuredTool.from_function(create_refund_request,
            description="为已支付订单创建退款申请"),
        StructuredTool.from_function(get_logistics,
            description="按订单 ID 查询物流轨迹"),
        StructuredTool.from_function(get_faq,
            description="检索 FAQ 知识库回答常见问题"),
    ]
    return create_agent(tools)

def build_ops_agent():
    tools = [
        StructuredTool.from_function(get_daily_sales_summary,
            description="按日期 + 部门查询销售汇总"),
        StructuredTool.from_function(get_top_products,
            description="按时间段查询销售 TopN 商品"),
        StructuredTool.from_function(get_funnel_conversion,
            description="查询某个漏斗的转化率统计"),
        StructuredTool.from_function(export_report,
            description="将查询结果导出为 Excel 报表"),
    ]
    return create_agent(tools)

# 入口路由:根据用户意图选 Agent, 而不是给一个 Agent 配所有工具
def route(user_query: str, user_role: str):
    if user_role == "customer_service":
        return build_customer_service_agent()
    elif user_role == "ops":
        return build_ops_agent()
    # ...

认知翻转:Agent 的能力天花板不是"用什么模型",而是"工具设计得好不好"。一个工具描述清晰、参数明确、按场景拆分的 Agent,用 GPT-3.5 都能跑得稳;一个工具描述含糊、参数随便、几十个工具堆在一起的 Agent,GPT-4 也救不了。投在工具设计上的工程时间是 Agent 项目里 ROI 最高的部分——多写一个清晰的 description 比换大模型便宜得多,效果还更稳。先在工具设计上打磨,再考虑模型升级,顺序反了多半事倍功半。

二、ReAct vs Function Calling:别再用旧范式

LangChain 早期 Agent 框架是基于 ReAct(Reasoning + Acting)的:模型输出 "Thought: ..." + "Action: tool_name" + "Action Input: ..." 这种纯文本格式,框架去解析。这套机制是 2022 年模型还没原生支持工具调用时的产物,在 2024 年的今天已经过时——OpenAI、Anthropic、Google 主流模型都原生支持 Function Calling / Tools,稳定性远胜 ReAct 文本解析。

ReAct 的核心问题是格式不稳定。模型有时候不按格式输出,框架解析失败,要么报错要么走奇怪的回退路径。下面是 ReAct 经常出问题的几种格式:

ReAct 输出格式不稳定的典型场景

预期格式:
  Thought: 我需要查昨天的销售
  Action: get_daily_sales_summary
  Action Input: {"date": "2026-05-23"}

实际可能输出:
1. 多了空格或换行
   Thought:我需要查昨天的销售
   Action :get_daily_sales_summary
   Action Input :{"date": "2026-05-23"}

2. JSON 写成单引号或拼写错
   Action Input: {'date': '2026-05-23'}
   或
   Action Input: {date: "2026-05-23"}

3. 多调一个非法工具(幻觉)
   Action: get_sales  ← 实际工具叫 get_daily_sales_summary

4. 输出完 Thought 直接给答案不调工具
   Thought: 我可以直接回答
   Final Answer: 昨天销售大约 100 万

5. 思考链跑飞
   Thought: 我需要查昨天的销售, 但首先让我想想昨天是几号,
            昨天是周几, 这周的销售一般什么走势, 我应该考虑...
   (跑了 800 字才进入 Action)

每一种都要 LangChain 的 parser 用正则去兜底, 失败率永远不可能为零。
Function Calling 完全没这些问题, 因为是模型 API 层强约束的 JSON。

Function Calling 把"决定调哪个工具 + 参数是什么"做成了模型 API 的原生能力,返回的是结构化 JSON 而不是自由文本。模型训练时就专门优化过这套调用,稳定性远高于让它生成文本再解析。LangChain 0.1+ 推荐用 create_openai_tools_agent 或 create_anthropic_tools_agent,它们内部走 Function Calling,不再依赖文本解析:

from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 用 Tools API 而不是 ReAct
llm = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", (
        "你是一个内部数据助手, 通过调用工具回答用户问题。"
        "原则: 1. 优先复用已有工具结果, 不要重复调用相同参数。"
        "2. 一次回答最多调 5 次工具, 超出请告诉用户问题需要拆分。"
        "3. 工具返回 error 时检查参数是否正确, 修正后再调一次, 仍失败就告诉用户。"
        "4. 涉及修改/删除数据的工具必须先向用户确认。"
    )),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent, tools=tools,
    max_iterations=5,            # 硬限制调用次数
    max_execution_time=30,       # 硬限制执行时间(秒)
    return_intermediate_steps=True,
    handle_parsing_errors=True,
    verbose=True,
)

result = executor.invoke({"input": "帮我看一下昨天的销售汇总"})
print(result["output"])
for step in result["intermediate_steps"]:
    print(f"调用了 {step[0].tool}({step[0].tool_input}) -> {step[1][:100]}")

关于 max_iterations 的设置,我的经验值是 5-7 比较合适。低于 3 很多正常多步任务完不成,高于 10 基本就是模型在死循环。同时设 max_execution_time 兜底超时,避免某个工具 hang 死整个 Agent。

认知翻转:2024 年还在用 ReAct 文本解析的 Agent 框架基本都是历史遗留。Function Calling 是 OpenAI / Anthropic / Google 主流模型 API 层支持的原生能力,稳定性、准确性、调试体验全面碾压 ReAct。新项目应该直接用 create_openai_tools_agent / create_anthropic_tools_agent;老项目用 ReAct 的,只要模型支持 Tools 就应该尽快迁移,这是 Agent 稳定性的基础。但 Function Calling 也不是万能,工具设计本身的好坏仍然是天花板,这俩不冲突——好工具描述 + Function Calling 才是当前推荐姿势。

三、调用次数与成本控制:别让 Agent 烧光预算

Agent 跟普通 LLM 调用最大的区别是"一个用户请求可能触发多次模型调用 + 多次工具调用",成本是线性的甚至指数级。一个简单问题如果走 5 步 ReAct,模型就被调了 5 次,每次都要把"系统 prompt + 工具描述 + 历史对话 + 工具结果"全送进去,token 飞涨。生产里失控的 Agent 成本经常是预算的 5-10 倍。

控制成本的几个核心策略:

第一,设硬上限。max_iterations、max_execution_time、单次最大 token 数,这三个一定要设。预期外的死循环至少不会无限烧钱。同时给整个 Agent 调用设月度预算告警,超过阈值自动降级或拒服务。

第二,工具结果裁剪。工具返回的内容会拼到下一轮 Agent 的 context 里参与下一次模型调用。如果工具返回的是 10000 行 SQL 结果,下一轮 prompt 就会塞进 10000 行,token 爆炸。所以工具内部要做"返回长度上限",比如最多返回 50 行,告诉模型"还有 9950 行已省略"。

def get_user_orders(user_id: str, limit: int = 10) -> dict:
    """查询用户最近的订单(返回给 Agent)"""
    orders = db.query(
        "SELECT id, status, amount, created_at FROM orders "
        "WHERE user_id = %s ORDER BY created_at DESC LIMIT %s",
        (user_id, limit + 1)        # 多查一条用于判断"是否还有更多"
    )

    has_more = len(orders) > limit
    orders = orders[:limit]

    # 关键:返回给 Agent 的内容要短, 不要把全部字段都塞回去
    summary = [
        {
            "id": o["id"],
            "status": o["status"],
            "amount": float(o["amount"]),
            "created_at": o["created_at"].strftime("%Y-%m-%d"),
        }
        for o in orders
    ]

    return {
        "count": len(summary),
        "has_more": has_more,
        "orders": summary,
        "note": (f"该用户还有更多订单, 共 {limit}+ 条" if has_more
                 else "已返回该用户全部订单"),
    }

第三,缓存工具结果。同一个用户在同一个对话里,Agent 经常会重复调同样参数的工具(模型短期记忆不可靠)。在工具调用层加一个 in-memory cache 按 (tool_name, params) 哈希存最近 5 分钟的结果,能省掉 30%-50% 的重复调用。

from functools import lru_cache
import hashlib, json, time

class ToolCache:
    def __init__(self, ttl_seconds=300):
        self.cache = {}
        self.ttl = ttl_seconds

    def _key(self, tool_name, params):
        s = f"{tool_name}:{json.dumps(params, sort_keys=True, ensure_ascii=False)}"
        return hashlib.md5(s.encode()).hexdigest()

    def get(self, tool_name, params):
        k = self._key(tool_name, params)
        entry = self.cache.get(k)
        if entry and time.time() - entry["t"] < self.ttl:
            return entry["v"]
        return None

    def put(self, tool_name, params, value):
        k = self._key(tool_name, params)
        self.cache[k] = {"v": value, "t": time.time()}

tool_cache = ToolCache(ttl_seconds=300)

def cached_tool_wrapper(tool_fn):
    """包装工具函数, 自动走 cache"""
    def wrapper(**params):
        cached = tool_cache.get(tool_fn.__name__, params)
        if cached is not None:
            return cached
        result = tool_fn(**params)
        tool_cache.put(tool_fn.__name__, params, result)
        return result
    wrapper.__name__ = tool_fn.__name__
    wrapper.__doc__ = tool_fn.__doc__
    return wrapper

# 使用
get_user_orders_cached = cached_tool_wrapper(get_user_orders)

第四,选合适的模型。Agent 调度类任务对模型能力敏感,但不是所有任务都需要 GPT-4。可以做混合模型策略:外层 Agent 用强模型(GPT-4o)做规划和最终回复,内部某些简单工具调用(如格式化、翻译)用便宜模型(GPT-3.5 / Claude Haiku)。这种"模型分级"在生产里能省 50% 成本而效果几乎不下降。

第五,做"用户请求复杂度预判"。简单问题(查个订单、看个状态)根本不需要 Agent,直接拼个 prompt 让模型回答更快更便宜。复杂问题(跨多个数据源 + 多步推理)再走 Agent。可以用一个 cheap 模型先判断"这个请求要不要走 Agent",省下 80% 的不必要 Agent 调用。

认知翻转:Agent 是 LLM 应用里成本最容易失控的形态。一个 Agent 请求可能等于 5-10 次普通对话的 token,没有硬上限和成本控制就是自杀。把 max_iterations / max_execution_time / 工具结果裁剪 / 缓存 / 模型分级 这五件事做到位,同等效果下成本能压到原来的 1/3。生产上线前一定要算清楚单次 Agent 请求的最坏 token 估算,乘以预期 QPS 看月成本,有没有超预算,再决定上不上。先算账再上线,Agent 最容易在你不知道的时候烧光预算。

四、失败处理与重试策略

真实生产里,工具调用失败是常态而不是异常。SQL 超时、API 限流、参数校验不通过、外部服务挂掉,每一种都可能发生。Agent 框架默认对失败的处理很粗暴——要么直接把 error 抛给模型让它自己处理,要么直接终止。真正能用的 Agent 必须自己设计完整的失败处理流程。

[mermaid]
flowchart TD
A[Agent 调用工具] --> B{工具返回}
B -->|成功| C[结果送回模型 进入下一轮]
B -->|参数错误| D[把详细错误送回模型 让它修正参数]
B -->|限流 临时错| E[退避重试 最多 3 次]
B -->|权限不足| F[告诉模型该工具该用户无权 不要重试]
B -->|外部服务挂| G[降级到备用工具或直接告诉用户暂时不可用]
E -->|重试成功| C
E -->|仍失败| G
D --> A
F --> H[Agent 综合现有信息回答 不再调该工具]

下面是一个分层的失败处理实现,把不同错误类型映射到不同的策略:

import time
from enum import Enum
from typing import Any

class ToolErrorType(Enum):
    PARAM_INVALID = "param_invalid"    # 参数错, 让模型修正
    RATE_LIMITED  = "rate_limited"     # 限流, 退避重试
    PERMISSION    = "permission"       # 权限不足, 不重试
    SERVICE_DOWN  = "service_down"     # 服务挂, 降级
    UNKNOWN       = "unknown"          # 未知, 重试一次

class ToolResult:
    def __init__(self, success: bool, data: Any = None,
                 error_type: ToolErrorType = None, message: str = ""):
        self.success = success
        self.data = data
        self.error_type = error_type
        self.message = message

    def to_agent_message(self) -> str:
        """转成给模型看的内容"""
        if self.success:
            return f"工具执行成功, 返回: {self.data}"
        if self.error_type == ToolErrorType.PARAM_INVALID:
            return f"参数错误: {self.message}。请检查参数后重试。"
        if self.error_type == ToolErrorType.PERMISSION:
            return f"权限不足: {self.message}。请告诉用户没有权限执行此操作。"
        if self.error_type == ToolErrorType.SERVICE_DOWN:
            return f"服务暂不可用: {self.message}。请告诉用户稍后再试。"
        return f"工具执行失败: {self.message}"

def safe_tool_call(tool_fn, params: dict, max_retries: int = 3) -> ToolResult:
    """带重试的工具调用包装"""
    for attempt in range(max_retries):
        try:
            data = tool_fn(**params)
            return ToolResult(success=True, data=data)
        except ValueError as e:
            # 参数错误, 不重试, 让模型修正
            return ToolResult(success=False,
                              error_type=ToolErrorType.PARAM_INVALID,
                              message=str(e))
        except PermissionError as e:
            return ToolResult(success=False,
                              error_type=ToolErrorType.PERMISSION,
                              message=str(e))
        except RateLimitError as e:
            # 退避重试
            time.sleep(2 ** attempt)
            continue
        except (ConnectionError, TimeoutError) as e:
            if attempt == max_retries - 1:
                return ToolResult(success=False,
                                  error_type=ToolErrorType.SERVICE_DOWN,
                                  message=str(e))
            time.sleep(2 ** attempt)
            continue
        except Exception as e:
            if attempt == max_retries - 1:
                return ToolResult(success=False,
                                  error_type=ToolErrorType.UNKNOWN,
                                  message=str(e))
    return ToolResult(success=False,
                      error_type=ToolErrorType.UNKNOWN,
                      message="超过最大重试次数")

另一个被忽略的设计是"幻觉调用的处理"。模型偶尔会编造一个不存在的工具名,Function Calling API 层会防住一部分但不是全部。Agent 框架应该在分发工具调用前先校验:如果调用的工具名不在注册列表里,直接返回错误让模型重新决策,而不是把这个错误吞掉:

def dispatch_tool_call(tool_name: str, params: dict,
                        registered_tools: dict) -> ToolResult:
    if tool_name not in registered_tools:
        # 幻觉工具, 告诉模型可用的工具列表
        available = ", ".join(registered_tools.keys())
        return ToolResult(
            success=False,
            error_type=ToolErrorType.PARAM_INVALID,
            message=f"工具 {tool_name} 不存在。可用工具: {available}",
        )
    tool_fn = registered_tools[tool_name]
    return safe_tool_call(tool_fn, params)

认知翻转:Agent 框架默认对失败的处理是"抛错或忽略",生产里都不够用。真正能用的 Agent 必须自己把失败分类(参数错/限流/权限/服务挂/未知),每种走不同的策略(修正/退避/拒绝/降级/重试)。这一层失败处理的代码量经常超过 Agent 业务逻辑本身,但它决定了 Agent 在真实世界里能不能撑住——demo 跑得稳的 Agent 上线后崩,80% 是这一层没做。

五、权限边界:别让 Agent 替用户干坏事

Agent 最大的安全风险是"它会按用户指令调工具",而工具是可以改数据库、可以发邮件、可以调外部 API 的。一个没有权限边界的 Agent 就是一把对内的双刃剑——好人用它提效,坏人用它批量做坏事。我前面提到的"导出所有用户密码"事故就是典型,如果当时 SQL 工具有 readonly 限制 + 表级别白名单,模型就算想干也干不了。

权限边界的设计要分多层:

第一层:工具白名单。给每个用户角色一份"允许调用的工具列表",在 Agent 初始化时只注入这些工具,模型连看到都看不到其他工具,自然调不了。客服角色配查询类工具,运营角色配查询 + 报表工具,管理员才有修改工具。

第二层:工具内部权限校验。即使工具被调到了,工具实现里也要做参数级别的权限校验。比如 get_user_orders 工具,普通员工只能查自己负责的客户,不能查别人的;管理员才能查全部。这一层是"工具白名单"的补充,因为同一个工具不同人调,可见的数据范围可能不同。

from dataclasses import dataclass

@dataclass
class UserContext:
    user_id: str
    role: str               # customer_service / ops / admin
    permissions: set        # 细粒度权限标识

def get_user_orders(user_id: str, limit: int = 10,
                    _ctx: UserContext = None) -> dict:
    # 权限校验:普通客服只能查自己关联的用户
    if _ctx.role == "customer_service":
        if not is_assigned_to(_ctx.user_id, user_id):
            raise PermissionError(
                f"客服 {_ctx.user_id} 没有查询用户 {user_id} 订单的权限"
            )
    # 管理员可以查所有, 不校验
    return _get_orders_impl(user_id, limit)

def execute_sql(query: str, _ctx: UserContext = None) -> dict:
    # SQL 工具的强权限:只允许 readonly 且只能查白名单表
    if not query.strip().upper().startswith("SELECT"):
        raise PermissionError("execute_sql 只允许 SELECT 查询")

    allowed_tables = {"orders", "products", "users_public_view"}
    used_tables = extract_table_names(query)
    if not used_tables.issubset(allowed_tables):
        forbidden = used_tables - allowed_tables
        raise PermissionError(f"不允许查询表: {forbidden}")

    if _ctx.role not in ("ops", "admin"):
        raise PermissionError("execute_sql 仅限运营和管理员使用")

    return run_readonly_query(query)

第三层:危险操作二次确认。涉及"修改/删除/对外发送"的工具,模型调之前必须先 ask_user_to_confirm。这一层不能依赖模型"自觉"——Agent 框架要在工具上打 confirm_required=True 标记,运行时强制弹出确认按钮给用户,用户点了才执行。

@dataclass
class Tool:
    name: str
    fn: callable
    description: str
    confirm_required: bool = False     # 是否需要用户确认

TOOLS = {
    "get_user_orders": Tool(
        name="get_user_orders", fn=get_user_orders,
        description="查询用户最近订单",
        confirm_required=False,        # 只读, 不需要确认
    ),
    "create_refund_request": Tool(
        name="create_refund_request", fn=create_refund_request,
        description="为已支付订单创建退款申请",
        confirm_required=True,         # 涉及钱, 必须确认
    ),
    "send_notification": Tool(
        name="send_notification", fn=send_notification,
        description="给用户发短信通知",
        confirm_required=True,         # 涉及对外, 必须确认
    ),
}

def execute_agent_tool_call(tool_name: str, params: dict,
                             user_ctx: UserContext) -> ToolResult:
    tool = TOOLS.get(tool_name)
    if tool is None:
        return ToolResult(False, error_type=ToolErrorType.PARAM_INVALID,
                          message=f"工具 {tool_name} 不存在")

    if tool.confirm_required:
        # 阻塞等用户在 UI 上点确认, 拒绝就返回拒绝
        if not request_user_confirmation(tool_name, params):
            return ToolResult(False, error_type=ToolErrorType.PERMISSION,
                              message="用户拒绝执行该操作")

    return safe_tool_call(tool.fn, {**params, "_ctx": user_ctx})

第四层:审计日志。所有 Agent 的工具调用都要留痕——谁(user)、什么时候(time)、用什么意图(query)、调了什么(tool + params)、结果是什么(success / error)。一旦出问题能追溯,日常也能审计有没有滥用。这一层是合规和事故处置的基础,生产 Agent 不可省。

认知翻转:Agent 的权限边界不是"加一层 if 判断"就够,是一套"白名单 + 工具内校验 + 二次确认 + 审计"的多层防御。LLM 是天然的"按指令执行者",你不告诉它什么不能做,它就什么都可能去做。生产 Agent 上线前必须画一张"威胁建模图":想象一个恶意用户能用 Agent 干哪些坏事,每条都要有对应的拦截层。没做过威胁建模就上线的 Agent 等于无防的内网门户,被滥用是时间问题。

六、可观测:出问题要能查出来

Agent 的可观测比普通 LLM 应用复杂得多,因为一个用户请求里有多次模型调用 + 多次工具调用 + 多个中间 thought,任何一个环节出问题都可能影响最终结果。出事的时候你需要能完整回放整个链路,看模型每一步在想什么、调了什么、收到什么。

核心可观测要素:

第一,trace 级别的完整记录。每个用户请求生成一个 trace_id,这个 trace 下挂所有的 LLM 调用 + tool 调用 + 中间状态,按时间顺序串起来。LangSmith / LangFuse / Phoenix 这些 LLM 专用可观测工具都做了这层,生产强烈推荐接入,自己写费时还容易漏。

第二,每次 LLM 调用要存完整 prompt 和 response。出 bug 时你需要知道"模型那一刻看到了什么文字,输出了什么",这是排查幻觉、参数错位、调错工具的唯一线索。不存的话用户说"它今天答得不对"你只能干瞪眼。

第三,工具调用的输入输出要存全。包括失败时的 error stacktrace。同一个工具同一个参数能不能复现失败,是排查工具 bug 的关键。

第四,关键指标按业务分桶。tools_per_request(平均工具调用次数)、agent_iterations(平均循环次数)、success_rate(任务完成率)、p95_latency、avg_tokens_per_request,按场景 / 用户角色 / 工具类型分桶统计。一旦某个桶的指标恶化就报警。

import time, uuid, json
from contextvars import ContextVar

current_trace = ContextVar("current_trace", default=None)

class AgentTrace:
    def __init__(self, user_id: str, query: str):
        self.trace_id = str(uuid.uuid4())
        self.user_id = user_id
        self.query = query
        self.start_time = time.time()
        self.steps = []

    def log_llm_call(self, model: str, prompt: str, response: str,
                      tokens_in: int, tokens_out: int, cost: float):
        self.steps.append({
            "type": "llm",
            "ts": time.time(),
            "model": model,
            "prompt": prompt[:5000],     # 截断防止过长
            "response": response[:5000],
            "tokens_in": tokens_in,
            "tokens_out": tokens_out,
            "cost_usd": cost,
        })

    def log_tool_call(self, tool: str, params: dict,
                       result: Any, success: bool, error: str = None):
        self.steps.append({
            "type": "tool",
            "ts": time.time(),
            "tool": tool,
            "params": params,
            "result": str(result)[:5000],
            "success": success,
            "error": error,
        })

    def finalize(self, output: str, status: str = "success"):
        self.end_time = time.time()
        record = {
            "trace_id": self.trace_id,
            "user_id": self.user_id,
            "query": self.query,
            "output": output,
            "status": status,
            "duration_ms": int((self.end_time - self.start_time) * 1000),
            "total_llm_calls": sum(1 for s in self.steps if s["type"] == "llm"),
            "total_tool_calls": sum(1 for s in self.steps if s["type"] == "tool"),
            "total_cost_usd": sum(s.get("cost_usd", 0)
                                  for s in self.steps if s["type"] == "llm"),
            "steps": self.steps,
        }
        # 写入存储(ClickHouse / S3 / 日志服务)
        trace_storage.put(record)
        # 同时上报指标到 Prometheus
        metrics.observe("agent_duration_ms", record["duration_ms"])
        metrics.observe("agent_tool_calls", record["total_tool_calls"])
        metrics.observe("agent_cost_usd", record["total_cost_usd"])
        return record

def run_agent_with_trace(user_id: str, query: str):
    trace = AgentTrace(user_id, query)
    current_trace.set(trace)
    try:
        result = agent_executor.invoke({"input": query})
        trace.finalize(result["output"], "success")
        return result
    except Exception as e:
        trace.finalize(str(e), "error")
        raise

除了 trace,生产里还要做"回放沙箱"。把生产某个失败 trace 完整 dump 下来,在测试环境用同样的 prompt + 同样的工具 mock 回放,看是不是能复现,改完 prompt 或工具后再回放确认修好。这套回放能力是 Agent 持续优化的基础工具,没有它每次改 prompt 都是赌运气。

认知翻转:Agent 的可观测不是"打几条日志"就够,而是要做"全链路 trace + 关键指标 + 回放沙箱"。出问题时没有 trace 你完全没法排查,看到指标恶化没有 trace 你不知道哪些请求出了什么问题,改完 prompt 没有回放你不知道有没有改对。LangSmith / LangFuse / Phoenix 这类工具不是锦上添花,是 Agent 生产化的基础设施,生产上线前必须接入。先有可观测,再上 Agent,顺序反了等出事故时哭都来不及。

关键概念速查

概念 含义 常见误区 正确做法
工具粒度 单个工具完成的事的大小 太粗或太细 每个工具完成一件人能一句话说清的事
工具描述 给模型看的工具说明 一句话带过 详细说用途/参数/反例
Function Calling 原生 API 工具调用 还用 ReAct 文本解析 2024 年新项目首选
max_iterations Agent 循环上限 用默认 15 5-7 适合大多数场景
工具结果裁剪 限制工具返回长度 原样返回 SQL 结果几千行 限制条数 + 说明 has_more
工具缓存 相同参数缓存结果 不做 5 分钟 cache 省 30% 调用
失败分类 不同错误不同策略 统一抛给模型 参数错让修正 服务挂降级
权限白名单 角色对应工具子集 所有工具对所有人开放 按角色注入工具
二次确认 修改类操作需用户确认 模型决定就执行 confirm_required 标记 + UI 确认
Trace 可观测 完整链路记录 只存最终输出 LangSmith/LangFuse 接入

避坑清单

  1. 不要把所有工具都塞给一个 Agent,按场景拆 Agent,每个 Agent 配 5-10 个紧贴场景的工具。
  2. 不要工具描述写一句话就完事,要详细写用途、参数、enum 合法值、何时不该用,详细描述比换大模型便宜。
  3. 不要还在用 ReAct 文本解析,Function Calling 是当前主流模型 API 层原生能力,稳定性碾压。
  4. 不要不设 max_iterations 和 max_execution_time,死循环能把账单和延迟同时打爆。
  5. 不要工具结果原样返回,大结果集要裁剪 + 标记 has_more,避免下一轮 prompt 爆炸。
  6. 不要不做工具缓存,Agent 在一个对话里重复调相同参数工具是常态,缓存能省 30%-50% 调用。
  7. 不要把所有错误一股脑抛给模型,要按错误类型分流——参数错让模型修正、限流退避重试、服务挂降级。
  8. 不要不做权限边界,Agent 没有白名单 + 工具内校验 + 二次确认就是裸奔,被滥用是时间问题。
  9. 不要忽视幻觉调用,工具分发前要校验工具名是否注册,不在的直接返回错让模型重选。
  10. 不要不接 LLM 可观测工具,生产 Agent 没有 trace + 回放就完全没法排查问题。

总结

LangChain Agent 是 LLM 应用里"看起来最自由实际最难驯服"的形态。自由是因为你可以给它任意工具,它会自己规划怎么用;难驯服也是因为它太自由——一次调用十几个工具、参数乱填、死循环、调错工具、烧光预算、绕过权限,任何一种都可能在生产里发生。一份能上线的 Agent 不取决于代码多少行,取决于"你对每一种可能的失败都做了预案"。

另一层被严重低估的是,Agent 的效果"上限"不是模型能力决定的,是周边工程决定的。工具设计、Function Calling、max_iterations、失败分流、权限边界、可观测,这些每一项都能让 Agent 的稳定性浮动一个数量级,而它们都跟"你用了 GPT-4 还是 Claude Opus"没直接关系。换更强的模型只能让你的 Agent 从 70 分涨到 75 分,把工程链路补完整能让你从 70 分涨到 90 分。前者花钱,后者花心思,后者性价比高得多。

打个不太严谨的比方,做 Agent 有点像招一个新员工。这个员工很聪明(LLM 能力强)但完全不了解公司业务。你直接扔给他一份"全公司所有工具的清单"让他自己干活,他大概率会乱用、用错权限的工具、做重复的事。靠谱的做法是按岗位给他配一份精挑细选的工具(白名单),每个工具写清楚怎么用什么时候用(详细描述),涉及大动作前必须找你确认(二次确认),做事过程留下完整记录(trace),做错了你能教他改(可观测 + 回放)。一个新员工能不能成长为靠谱的同事这些环节缺一不可,Agent 同样如此。

所以做 Agent,本地跑通几个 demo case 永远暴露不了真正的问题。它暴露不了用户的真实请求分布跟你想象的天差地别,暴露不了某些请求让模型陷入死循环烧光 token,暴露不了模型把"销售部门"传成"销售"导致工具反复报错,暴露不了恶意用户用一句巧妙的 prompt 让 Agent 干危险事,更暴露不了某一天工具上游 API 挂了 Agent 全线崩溃没有降级。真正的检验在生产环境,在第一周用户大规模使用的下午,在一次模型升级后某些 prompt 突然失效的早晨,在一次恶意输入差点让 Agent 干坏事的午后,在一次月底账单出来比预算高三倍的尴尬时刻。把上面六节里的功夫提前做扎实,等那些时刻到来时,你会感谢自己当初没图省事。如果你正在做或者准备做 Agent,请把它当成一个"工具 + 调度 + 控制 + 观测"的多阶段工程系统,而不是"调几个 LangChain API 的脚本"——这是从 demo 到生产最关键也最容易被忽略的认知差。

—— 别看了 · 2026
声明:本站所有文章,如无特殊说明或标注,均为本站原创发布。任何个人或组织,在未征得本站同意时,禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益,可联系我们进行处理 邮箱1846861578@qq.com。
技术教程

Kafka 消费者组与 rebalance 工程实战:从一次"凌晨 3 点整组消费停顿 20 分钟"看懂为什么慢消息能拖垮你的消费链路

2026-5-24 14:25:42

技术教程

Redis 缓存穿透/击穿/雪崩三大场景实战指南:从一次"大促零点缓存命中率从 98% 掉到 12%"看懂为什么 Redis 加 TTL 远远不够

2026-5-24 14:39:26

0 条回复 A文章作者 M管理员
    暂无讨论,说说你的看法吧
个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
搜索