2024 年中我们做一个 AI 财务助手 接入 LangChain Agent 给企业 CFO 做财务分析 我以为很简单 写 5 个 tool 查询数据库 查询 ERP 计算指标 生成报表 邮件发送 用 ZeroShotReactDescription Agent 串起来 demo 跑通效果惊艳 老板拍板上线。然而上线两周后我们陆续踩了一堆坑。第一种最让我傻眼 Agent 调用 LLM 决策下一步 一个简单的 查上月营收并对比上年 任务 LLM tool 选择反复横跳 调了 12 次 LLM 花了 30 秒 OpenAI 费用 0.5 美金一次 月调用 1 万次 1.5 万美金 老板看到账单炸毛。第二种最难缠 LLM 在 tool 之间死循环 调了 query_db 没找到数据 又调 query_db 再没找到 又调 query_db 一直循环 max_iterations 默认 15 用完报错 用户看到 Agent stopped due to iteration limit 直接放弃。第三种最离谱 LLM 调用 tool 时参数解析错误 我们 tool 定义 def query_sales(month: str, year: int) LLM 传 query_sales(month='2024-01', year='2024') 字符串当 int 报错 没做 schema validation 错误直接抛给用户。第四种最致命 我们 tool 里有 send_email 一次 Agent 误判任务 自动发了一封 上月营收 -5000 万 的报表给所有 CFO 邮箱 实际数据正确但报表格式错 引起恐慌 没做 human-in-the-loop 危险 tool 直接执行后果严重。第五种最莫名其妙 我们用 GPT-4 测试都好 上线为了省钱换 GPT-3.5 同样的 prompt Agent 行为完全不一样 tool 选择准确率从 85% 降到 40% 用户体验断崖式下跌。我盯着这一连串问题想了很久才彻底想明白第一版错在一个根本的认知上我以为 Agent 就是 给 LLM 几个 tool LLM 自己决定怎么用 可这个认知是错的真正能用的 Agent 是一个 Tool 设计与 schema 验证 加 决策流程控制 加 错误处理与重试 加 成本控制 加 human-in-the-loop 加 评估与监控 的整套工程方法论 任何一环没做都可能让你的 Agent 烧钱 死循环 误操作或者用户放弃本文从头梳理 LangChain Agent 工程化的要点 tool 怎么设计 决策怎么控制 错误怎么处理 成本怎么压 危险操作怎么兜底 以及一些把 Agent 做扎实要避开的工程坑
问题背景:为什么 Agent 不是写几个 tool 就完事
很多人对 LangChain Agent 的认知是 给 LLM 写几个 tool LLM 自己决定调哪个 完事 实际上 Agent 在生产里你会发现 LLM 决策反复横跳 token 烧爆 死循环 tool 参数解析错 危险 tool 误执行 模型切换行为漂移 任何一个都能让你的 AI 应用翻车。问题的根源在于:
- Tool 设计决定 Agent 智商:tool 描述模糊 LLM 选错 tool 描述清晰且互斥 选对率 90%+。
- schema 严格验证不可省:LLM 经常传错参数类型 必须 pydantic 校验失败重试或转人工。
- 决策流程必须控制:ReAct Agent 容易死循环 必须 max_iterations + early stop 条件。
- 成本是真金白银:GPT-4 一次 Agent 调用 10-20 次 LLM 月调用百万就是百万美金账单。
- 危险 tool 必须人工确认:发邮件 改数据 转账 这些必须 human-in-the-loop。
- 评估比 demo 重要:200 条评估集跑通 Agent 任务成功率 才能上线 不能凭感觉。
一 Tool 设计:决定 Agent 智商
Tool 是 Agent 的能力 tool 设计的好坏直接决定 Agent 智商。tool 描述模糊 LLM 选错 tool 描述清晰且互斥 选对率高。每个 tool 必须有清晰的 description 输入 schema 输出 schema 错误处理 这些都是工程细节不是装饰。
from langchain.tools import StructuredTool
from pydantic import BaseModel, Field
from typing import Literal
# 1 用 pydantic 定义严格 schema
class QuerySalesInput(BaseModel):
"""查询销售数据的输入参数"""
month: str = Field(
description='月份 格式 YYYY-MM 例如 2024-01',
pattern=r'^\d{4}-\d{2}$',
)
region: Literal['north', 'south', 'east', 'west', 'all'] = Field(
default='all',
description='地区 仅支持 north south east west all',
)
metric: Literal['revenue', 'orders', 'customers'] = Field(
description='指标类型 revenue 营收 orders 订单数 customers 客户数',
)
def query_sales(month: str, region: str, metric: str) -> dict:
"""实际查询逻辑"""
# 严格校验日期格式
import datetime
try:
datetime.datetime.strptime(month, '%Y-%m')
except ValueError:
return {'error': f'日期格式错误 应为 YYYY-MM 实际 {month}'}
# 查数据库
result = db.query(f"SELECT * FROM sales WHERE month='{month}' AND region='{region}'")
return {
'month': month, 'region': region, 'metric': metric,
'value': result[metric],
'data_freshness': '截至昨日',
}
# 2 注册为 LangChain tool
query_sales_tool = StructuredTool.from_function(
func=query_sales,
name='query_sales_data',
description=(
'查询历史销售数据 仅支持已结算月份 不支持当月与未来 '
'用于回答 X 月营收 X 月订单数 等问题 '
'不用于实时数据查询 实时数据请用 query_realtime tool'
),
args_schema=QuerySalesInput,
return_direct=False, # 结果给 LLM 继续推理 而非直接返回用户
)
tool 描述的几个关键原则 一是描述要明确 这个 tool 做什么 不做什么 二是与其他 tool 互斥 不要让 LLM 选错 三是 schema 严格 用 pydantic 卡住参数类型 四是错误返回结构化信息让 LLM 知道怎么纠正。
# 3 多个 tool 之间互斥 避免 LLM 混淆
class QueryRealtimeInput(BaseModel):
"""实时数据查询"""
metric: str = Field(description='指标名 例如 today_revenue current_active_users')
query_realtime = StructuredTool.from_function(
name='query_realtime_metric',
description=(
'查询实时指标 例如 今日营收 当前活跃用户 '
'只支持今日及当前状态 不支持历史月份 历史月份请用 query_sales_data'
),
func=lambda metric: {'metric': metric, 'value': cache.get(metric)},
args_schema=QueryRealtimeInput,
)
# 4 tool 输出统一结构化 便于 LLM 解析
class ToolOutput(BaseModel):
"""所有 tool 的统一输出格式"""
success: bool
data: dict | None = None
error: str | None = None
suggestion: str | None = None # 失败时给 LLM 的纠正建议
def safe_tool_wrapper(func):
"""tool 异常包装 失败也返回结构化信息 不抛异常"""
def wrapper(*args, **kwargs):
try:
result = func(*args, **kwargs)
return ToolOutput(success=True, data=result).model_dump()
except ValueError as e:
return ToolOutput(
success=False,
error=str(e),
suggestion='请检查参数格式',
).model_dump()
except Exception as e:
return ToolOutput(
success=False,
error=f'系统错误 {type(e).__name__}',
suggestion='请稍后重试或换一个查询方式',
).model_dump()
return wrapper
Tool 设计的工程经验 用 pydantic 定义严格 schema LLM 传错类型直接拒绝 description 必须明确 做什么不做什么 与其他 tool 互斥 输出统一结构化 success/data/error/suggestion 异常包装绝不抛给 Agent 这套组合能让 tool 调用成功率从 70% 提到 95% LLM 推理流畅。我们公司 Agent 重构 tool schema 后 任务成功率从 60% 提到 88% 用户满意度大涨。
二 决策流程控制:防死循环防失控
Agent 的核心是 LLM 决策下一步 但 LLM 决策不可控 容易死循环 反复调同一 tool 反复换 tool 都跑不出来。必须做决策流程控制 max_iterations early_stop 条件 推理记录 这些都是上线硬规范。
from langchain.agents import AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
# 1 创建 Agent 设置 max_iterations
llm = ChatOpenAI(model='gpt-4-turbo', temperature=0)
REACT_PROMPT = PromptTemplate.from_template("""你是企业财务助手 严格按以下流程工作
可用 tool {tools}
tool 名字 {tool_names}
格式
Question 用户问题
Thought 我应该用什么 tool
Action tool 名字
Action Input tool 输入 JSON
Observation tool 返回
... 最多 5 步
Thought 我已经有足够信息
Final Answer 最终回答
【强制规则】
1 同一 tool 最多调用 2 次 如果 2 次都失败立即返回 Final Answer 转人工
2 总步数不超过 5 步 即使没拿到完美答案也要给出当前最佳回答
3 危险 tool send_email update_database 必须等用户确认 不要直接执行
4 tool 报错时 不要重复调用 改用其他 tool 或转人工
Question {input}
Thought {agent_scratchpad}""")
tools = [query_sales_tool, query_realtime, calculate_metric, generate_report]
agent = create_react_agent(llm, tools, REACT_PROMPT)
# 2 配置 AgentExecutor 防失控
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
max_iterations=5, # 最多 5 步
max_execution_time=60, # 最多 60 秒
early_stopping_method='generate', # 超时返回当前最佳答案
handle_parsing_errors='请用正确的 JSON 格式', # 解析失败给 LLM 提示
return_intermediate_steps=True, # 返回中间步骤便于调试
verbose=False,
)
决策流程的几个关键参数 max_iterations 控制最大步数 max_execution_time 控制最大时间 这两个都是硬上限 触发就停止 不要让 LLM 自己决定何时停 否则要么停太早要么停不下来。
# 3 自定义决策控制 检测死循环
from collections import defaultdict
class LoopDetector:
"""检测 Agent 死循环"""
def __init__(self, max_repeats: int = 2):
self.history: list[tuple] = []
self.max_repeats = max_repeats
def check(self, tool_name: str, tool_input: dict) -> bool:
"""返回 False 表示死循环 需要中止"""
key = (tool_name, str(sorted(tool_input.items())))
self.history.append(key)
# 看最近 5 步是否同一 tool 调用超过 2 次
recent = self.history[-5:]
count = recent.count(key)
if count > self.max_repeats:
return False
return True
# 4 自定义 callback 监控 Agent 行为
from langchain.callbacks.base import BaseCallbackHandler
class AgentMonitor(BaseCallbackHandler):
"""监控 Agent 步数 tool 调用 成本"""
def __init__(self):
self.step_count = 0
self.tool_calls = defaultdict(int)
self.total_tokens = 0
self.loop_detector = LoopDetector()
def on_tool_start(self, serialized, input_str, **kwargs):
tool_name = serialized.get('name', 'unknown')
self.tool_calls[tool_name] += 1
self.step_count += 1
# 检测死循环
if not self.loop_detector.check(tool_name, {'input': input_str}):
raise RuntimeError(f'detected loop on {tool_name} stop')
def on_llm_end(self, response, **kwargs):
# 累计 token 估算成本
usage = response.llm_output.get('token_usage', {})
self.total_tokens += usage.get('total_tokens', 0)
# 5 调用 Agent 带监控
monitor = AgentMonitor()
result = agent_executor.invoke(
{'input': '查询上月营收并对比上年同期'},
config={'callbacks': [monitor]},
)
print(f'步数 {monitor.step_count} token {monitor.total_tokens}')
print(f'tool 调用统计 {dict(monitor.tool_calls)}')
决策流程控制的工程经验 max_iterations=5 max_execution_time=60s 是合理硬上限 超过就停 不要让 LLM 自己决定 用 callback 检测死循环 同一 tool 调用 2 次失败立即中止 记录每次推理步骤便于事后审计 用 handle_parsing_errors 给 LLM 提示而非直接报错 这套组合能让 Agent 永不失控 用户体验稳定。我们公司 Agent 平均 3 步完成任务 比之前默认 15 步快 5 倍 成本降 70%。
三 错误处理与重试:让 Agent 优雅失败
Agent 调用 tool 失败 LLM 决策错误 这些是常态 不是异常。必须做错误处理 让 Agent 失败时优雅降级而不是崩溃。错误信息要结构化让 LLM 能纠正 多次失败要中止避免烧钱 还要日志记录便于追溯。
from langchain.schema import AgentAction, AgentFinish
from langchain_core.exceptions import OutputParserException
import logging
logger = logging.getLogger(__name__)
# 1 tool 错误的分级处理
class ToolError(Exception):
"""业务相关错误 给 LLM 看 让它纠正"""
pass
class SystemError(Exception):
"""系统错误 不要给 LLM 看 直接转人工"""
pass
def query_sales_safe(month: str, region: str, metric: str) -> dict:
"""带错误分级的 tool"""
try:
# 参数验证 给 LLM 纠正机会
if not re.match(r'^\d{4}-\d{2}$', month):
raise ToolError(f'月份格式错误 期望 YYYY-MM 实际 {month}')
# 业务逻辑校验
from datetime import datetime
target = datetime.strptime(month, '%Y-%m')
if target > datetime.now():
raise ToolError('不支持查询未来月份 请查询历史月份')
# 实际查询
result = db.query_sales(month, region, metric)
if not result:
raise ToolError(f'未找到 {month} {region} 数据 可能数据未到位')
return {'success': True, 'data': result}
except ToolError as e:
return {'success': False, 'error': str(e), 'retry_hint': '请修正参数'}
except Exception as e:
# 系统错误 不让 LLM 重试 直接抛出转人工
logger.error(f'system error in query_sales: {e}', exc_info=True)
raise SystemError(f'数据库不可用 {type(e).__name__}')
# 2 Agent 级别的错误处理
def run_agent_safely(question: str, user_id: str) -> dict:
"""带完整错误处理的 Agent 调用"""
try:
result = agent_executor.invoke(
{'input': question},
config={'callbacks': [monitor], 'metadata': {'user_id': user_id}},
)
return {
'success': True,
'answer': result['output'],
'steps': len(result.get('intermediate_steps', [])),
'tokens': monitor.total_tokens,
}
except SystemError as e:
# 系统错误 转人工
logger.error(f'agent system error for user {user_id}: {e}')
return {
'success': False,
'answer': '系统暂时不可用 已为您转人工客服',
'fallback': 'human',
}
except RuntimeError as e:
# 死循环或解析错误
if 'loop' in str(e):
logger.warning(f'agent loop detected for user {user_id}')
return {
'success': False,
'answer': '查询过于复杂 已为您转人工客服',
'fallback': 'human',
}
raise
except Exception as e:
# 兜底
logger.exception(f'agent unknown error for user {user_id}')
return {
'success': False,
'answer': '系统错误 请稍后重试',
'fallback': 'retry',
}
错误处理的关键是区分业务错误与系统错误 业务错误 比如参数错 数据不存在 让 LLM 纠正重试 系统错误 比如数据库挂 直接转人工不要让 LLM 在崩溃的系统上反复尝试 浪费 token。
# 3 LLM 输出解析失败的处理
from langchain.agents.format_scratchpad.openai_tools import format_to_openai_tool_messages
# LangChain 默认 OutputParserException 时 retry 一次
# 我们可以自定义 retry 逻辑
class CustomAgent:
def __init__(self, agent_executor, max_retries=2):
self.executor = agent_executor
self.max_retries = max_retries
def invoke_with_retry(self, query: dict) -> dict:
last_error = None
for attempt in range(self.max_retries):
try:
return self.executor.invoke(query)
except OutputParserException as e:
last_error = e
logger.warning(f'parse error attempt {attempt}: {e}')
# 给 LLM 更明确的格式提示
query['input'] += '\n\n请严格按 JSON 格式返回 tool 输入'
continue
except SystemError:
raise # 系统错误不重试
raise RuntimeError(f'agent failed after {self.max_retries} retries: {last_error}')
# 4 监控指标埋点
def emit_agent_metrics(result: dict, user_id: str):
"""埋点 Agent 调用指标供监控"""
metrics.gauge('agent.tokens', monitor.total_tokens, tags=[f'user:{user_id}'])
metrics.gauge('agent.steps', monitor.step_count, tags=[f'user:{user_id}'])
metrics.increment('agent.calls', tags=[
f'success:{result["success"]}',
f'fallback:{result.get("fallback", "none")}',
])
for tool_name, count in monitor.tool_calls.items():
metrics.gauge(f'agent.tool_calls.{tool_name}', count)
错误处理的工程经验 区分业务错误与系统错误 业务错误让 LLM 重试 系统错误直接转人工 解析失败给 LLM 更明确的格式提示 重试 2 次还失败就中止 所有错误都记录日志带 user_id 便于追溯 埋点 token steps tool_calls 指标实时监控 这套组合能让 Agent 失败率从 30% 降到 5% 而且失败时优雅降级用户体验依然良好。我们公司一次系统故障 数据库挂 Agent 优雅降级到人工 用户没感受到事故 SLA 不破线。
[mermaid]flowchart TD
A[用户问题] --> B[Agent Executor]
B --> C[LLM 推理]
C -->|选 tool| D{tool 是否危险?}
D -->|是 发邮件 改数据| E[human approval]
D -->|否| F[执行 tool]
E -->|批准| F
E -->|拒绝| G[返回拒绝原因]
F -->|成功| H[Observation 给 LLM]
F -->|业务错误| I[结构化错误 LLM 重试]
F -->|系统错误| J[转人工]
H --> K{够回答?}
K -->|是| L[Final Answer]
K -->|否 步数 < 5| C
K -->|步数 = 5| M[返回当前最佳]
I --> C
I -->|同 tool 2 次失败| J
四 成本控制:防止账单爆炸
Agent 一次调用可能 10-20 次 LLM 每次 1000-5000 token GPT-4 单次成本 0.05-0.5 美金 月调用百万就是百万美金账单 这不是危言耸听 我们公司真的差点踩。成本控制必须在 tool 设计 模型选择 缓存 限流 多个层面同时做。
import hashlib
from functools import lru_cache
from langchain_openai import ChatOpenAI
# 1 模型分层 不同任务用不同 LLM
llm_high = ChatOpenAI(model='gpt-4-turbo', temperature=0) # 复杂决策
llm_low = ChatOpenAI(model='gpt-3.5-turbo', temperature=0) # 简单工具调用
llm_cheap = ChatOpenAI(model='gpt-4o-mini', temperature=0) # 极简任务
# 2 路由器先判断难度
ROUTER_PROMPT = """判断用户问题难度
simple 单一指标查询 例如 上月营收
medium 多指标对比 例如 上月与上年同期对比
complex 多步推理与分析 例如 找出营收下降原因
问题 {question}
难度"""
def route_llm(question: str) -> ChatOpenAI:
resp = llm_cheap.invoke(ROUTER_PROMPT.format(question=question))
difficulty = resp.content.strip().lower()
if 'complex' in difficulty:
return llm_high
elif 'medium' in difficulty:
return llm_low
return llm_cheap
# 3 tool 结果缓存
import redis
r = redis.Redis(host='localhost', port=6379, db=2)
def cached_tool(func):
"""tool 结果缓存到 Redis 1 小时"""
def wrapper(**kwargs):
cache_key = f'tool:{func.__name__}:' + hashlib.md5(
json.dumps(kwargs, sort_keys=True).encode()
).hexdigest()
cached = r.get(cache_key)
if cached:
return json.loads(cached)
result = func(**kwargs)
r.setex(cache_key, 3600, json.dumps(result))
return result
return wrapper
@cached_tool
def query_sales_cached(month: str, region: str, metric: str) -> dict:
return db.query_sales(month, region, metric)
成本控制还要考虑 LLM 提示词压缩 system prompt 越短越省 历史对话裁剪 不要带 10 轮历史 用 summary 压缩 这两个细节能让 token 降 50%。
# 4 token 用量监控与限流
from datetime import datetime, timedelta
class TokenBudget:
"""token 预算控制 防止单个用户烧爆"""
def __init__(self, daily_limit: int = 100000, monthly_limit: int = 2000000):
self.daily_limit = daily_limit
self.monthly_limit = monthly_limit
def check(self, user_id: str, estimated_tokens: int) -> bool:
"""检查是否超预算"""
today = datetime.now().strftime('%Y-%m-%d')
month = datetime.now().strftime('%Y-%m')
day_used = int(r.get(f'tokens:day:{user_id}:{today}') or 0)
month_used = int(r.get(f'tokens:month:{user_id}:{month}') or 0)
if day_used + estimated_tokens > self.daily_limit:
return False
if month_used + estimated_tokens > self.monthly_limit:
return False
return True
def consume(self, user_id: str, tokens: int):
"""消费 token 累计"""
today = datetime.now().strftime('%Y-%m-%d')
month = datetime.now().strftime('%Y-%m')
r.incrby(f'tokens:day:{user_id}:{today}', tokens)
r.expire(f'tokens:day:{user_id}:{today}', 86400 * 2)
r.incrby(f'tokens:month:{user_id}:{month}', tokens)
r.expire(f'tokens:month:{user_id}:{month}', 86400 * 35)
# 5 Agent 调用前预算检查
def safe_agent_invoke(question: str, user_id: str, budget: TokenBudget) -> dict:
estimated = len(question) * 10 # 粗估 1 字符 = 10 token (Agent 倍数)
if not budget.check(user_id, estimated):
return {'success': False, 'error': '今日额度已用完 请明日再试'}
result = run_agent_safely(question, user_id)
if result['success']:
budget.consume(user_id, monitor.total_tokens)
return result
# 6 成本核算面板
def daily_cost_report():
today = datetime.now().strftime('%Y-%m-%d')
keys = r.keys(f'tokens:day:*:{today}')
total_tokens = sum(int(r.get(k) or 0) for k in keys)
# GPT-4-turbo 输入 0.01/1k output 0.03/1k 假设 1:3
cost_usd = total_tokens / 1000 * 0.02 # 平均
return {
'date': today,
'total_tokens': total_tokens,
'users': len(keys),
'estimated_cost_usd': round(cost_usd, 2),
'cost_per_user_usd': round(cost_usd / max(len(keys), 1), 2),
}
成本控制的工程经验 模型分层 简单任务用 GPT-3.5/gpt-4o-mini 复杂任务用 GPT-4 用 router 自动选 tool 结果缓存 1 小时 避免重复查询 token 预算 每用户每日上限 1 万 token 每月 200 万 实时监控成本面板 这套组合能让 Agent 月成本从 5 万美金降到 5 千美金 降 90%。我们公司一次 GPT-3.5 替代 70% 简单任务 月省 3 万美金 CEO 笑得合不拢嘴。
五 Human-in-the-Loop:危险操作必须人工确认
Agent 自主决策很爽 但有些 tool 是危险的 发邮件 改数据 转账 删除文件 这些操作不能让 Agent 直接执行 必须 human-in-the-loop 人工确认。Agent 误判 不可逆操作 后果严重 我们公司就出过 Agent 自动发错报表的事 之后必加 HITL。
from enum import Enum
from datetime import datetime
class ToolRisk(Enum):
"""tool 危险等级"""
SAFE = 0 # 只读查询 无副作用
NOTICE = 1 # 读+少量写 比如日志
DANGER = 2 # 不可逆操作 比如发邮件 改数据
CRITICAL = 3 # 涉及钱与权限 比如转账 删除
# 1 tool 标注危险等级
TOOL_RISKS = {
'query_sales_data': ToolRisk.SAFE,
'query_realtime': ToolRisk.SAFE,
'calculate_metric': ToolRisk.SAFE,
'generate_report': ToolRisk.SAFE,
'send_email': ToolRisk.DANGER, # 必须人工确认
'update_kpi': ToolRisk.DANGER,
'transfer_money': ToolRisk.CRITICAL, # 必须 2 人确认
'delete_record': ToolRisk.CRITICAL,
}
# 2 待确认 tool 队列
class PendingApproval(BaseModel):
id: str
user_id: str
tool_name: str
tool_input: dict
risk: ToolRisk
created_at: datetime
status: Literal['pending', 'approved', 'rejected', 'expired']
approver: str | None = None
class ApprovalQueue:
"""等待人工确认的队列"""
def __init__(self):
self.queue: dict[str, PendingApproval] = {}
def submit(self, user_id: str, tool_name: str, tool_input: dict) -> str:
risk = TOOL_RISKS.get(tool_name, ToolRisk.SAFE)
if risk == ToolRisk.SAFE:
return None # 安全 tool 不需要审批
approval_id = f'app_{int(datetime.now().timestamp() * 1000)}'
self.queue[approval_id] = PendingApproval(
id=approval_id, user_id=user_id,
tool_name=tool_name, tool_input=tool_input,
risk=risk, created_at=datetime.now(),
status='pending',
)
# 通知审批人 Slack 邮件等
self.notify_approver(approval_id, risk)
return approval_id
def approve(self, approval_id: str, approver: str) -> bool:
if approval_id not in self.queue:
return False
item = self.queue[approval_id]
# CRITICAL 需要 2 人确认
if item.risk == ToolRisk.CRITICAL:
if not self.has_second_approver(approval_id, approver):
return False
item.status = 'approved'
item.approver = approver
return True
# 3 在 Agent 执行 tool 前检查审批
queue = ApprovalQueue()
def wrap_tool_with_approval(tool_func, tool_name: str):
"""给 tool 加审批层"""
def wrapped(**kwargs):
risk = TOOL_RISKS.get(tool_name, ToolRisk.SAFE)
if risk == ToolRisk.SAFE:
return tool_func(**kwargs)
# 提交审批请求
approval_id = queue.submit(
user_id=current_user_id,
tool_name=tool_name,
tool_input=kwargs,
)
# 返回给 Agent pending 状态 Agent 应该结束当前轮
return {
'success': False,
'pending_approval': True,
'approval_id': approval_id,
'message': f'此操作 ({tool_name}) 需要人工确认 已提交审批 ID {approval_id}',
}
return wrapped
# 4 等待审批 batch 处理
def poll_approvals():
"""轮询待审批的请求"""
while True:
for aid, item in list(queue.queue.items()):
if item.status == 'pending':
# 超时自动拒绝 默认 1 小时
if (datetime.now() - item.created_at).seconds > 3600:
item.status = 'expired'
notify_user(item.user_id, f'审批超时 操作未执行')
continue
elif item.status == 'approved':
# 真正执行 tool
result = execute_real_tool(item.tool_name, item.tool_input)
notify_user(item.user_id, f'操作已执行 {result}')
del queue.queue[aid]
time.sleep(10)
Human-in-the-Loop 的工程经验 tool 必须按危险等级分类 SAFE NOTICE DANGER CRITICAL 危险 tool 必须人工确认 不要让 Agent 直接执行 不可逆操作 审批超时自动拒绝 不要无限挂起 CRITICAL 操作 转账删除 必须 2 人确认 4 眼原则 审批记录全量日志便于审计 这套组合能让你的 Agent 既智能又安全 避免误操作灾难。我们公司加 HITL 后 Agent 误操作零事故 用户对 Agent 信任度大幅提升。
六 Agent 的工程坑:那些 demo 时学不到的
讲完原理来说几个真实生产里踩过的坑。第一个坑是 Agent 任务必须有评估集 不能凭感觉 我们做 200 条评估问题 覆盖简单 中等 复杂 各种 tool 组合 任何改动都跑完整评估 任务成功率下降 5% 拒绝上线 这套评估让 Agent 上线后零回滚。第二个坑是 模型切换不能直接换 GPT-4 跑的 Agent 换 GPT-3.5 行为完全不一样 必须每个模型一个版本 切换前重跑评估 GPT-3.5 tool 选择准确率比 GPT-4 低 20% 简单任务 OK 复杂任务不行。第三个坑是 多轮对话的 context 管理 Agent 默认带历史对话 对话越长 token 越爆 必须做 summary memory 把历史压缩成几句话不要全文本 这能省 50% token。第四个坑是 Agent 不要做所有事 有些任务直接 RAG 或 SQL 模板更好 不是所有 LLM 调用都要 Agent 我们把 60% 简单查询改成 SQL 模板 30% 中等复杂用 RAG 只剩 10% 复杂任务用 Agent 总成本降 80%。第五个坑是 Agent 上线前必做压测 一次 Agent 调用并发 100 OpenAI 限流报错满天飞 必须 rate limiter exponential backoff 重试 在客户端做好限流不要靠 OpenAI 自己限。
关键概念速查
| 概念 | 含义 | 工程价值 |
|---|---|---|
| tool description | tool 用途说明 | LLM 选择正确率 90% |
| pydantic schema | tool 参数校验 | 参数错误拦截 |
| 互斥 tool | 避免选择歧义 | 明确 do/dont |
| max_iterations | 最大步数 | 防死循环 |
| early_stopping | 提前结束 | 超时返回最佳 |
| loop detector | 同 tool 重复检测 | 2 次失败中止 |
| 结构化错误 | success/error/suggestion | LLM 可纠正 |
| 模型路由 | 分层 LLM | 成本降 70% |
| tool 缓存 | 结果 Redis | 避免重复调用 |
| HITL | 人工确认 | 危险 tool 必备 |
避坑清单
- Tool 用 pydantic 严格 schema description 明确互斥 不要让 LLM 猜。
- tool 输出统一结构化 success/data/error/suggestion 异常绝不抛给 Agent。
- 设置 max_iterations=5 max_execution_time=60s 不要让 LLM 自己决定。
- 用 callback 检测死循环 同 tool 2 次失败立即中止。
- 区分业务错误与系统错误 业务错误让 LLM 重试 系统错误转人工。
- 模型分层 简单用 GPT-3.5 复杂用 GPT-4 用 router 自动选 省 70%。
- tool 结果缓存 Redis 1 小时 避免重复查询同样问题。
- 设置 token 预算 每用户每日上限 防止恶意刷爆账单。
- 危险 tool 发邮件 改数据 必须 HITL 人工确认 CRITICAL 操作 2 人确认。
- 200 条评估集必跑 任何改动指标下降拒绝上线 不能凭感觉。
总结
LangChain Agent 这事 很多人的直觉是 写几个 tool 让 LLM 自己决定怎么用 完事。这其实是把 我能跑通 LangChain demo 和 我能在生产用 Agent 撑住企业财务助手 每月 10 万次调用 不烧爆账单不死循环不误操作 混为一谈。前者是会用 Agent 后者是懂 Agent 工程。中间隔着的是 tool 设计 决策控制 错误处理 成本控制 HITL 评估与监控 整整一套工程方法论。
从 demo 到生产 你需要做的事远不止 from langchain import AgentExecutor。你要懂 tool 怎么 schema 验证 怎么互斥设计 max_iterations 怎么定 死循环怎么检测 错误怎么分级 模型怎么路由 缓存怎么做 token 怎么限额 HITL 怎么实现 评估集怎么搭。每一项单独看都不复杂 但它们组合在一起 才是一个能上线撑住生产的 Agent。少任何一项 都可能让你的 AI 应用烧钱 死循环 误操作 用户放弃。
我经常用一个比喻来理解 Agent 它有点像一个新入职的实习生。tool 是公司给他的权限与工具 描述要清晰 不然他不知道用哪个。pydantic schema 是表格要填的字段 写错格式直接打回不让乱填。max_iterations 是上班时间 最多 5 步就要交付 不能磨蹭一整天。loop detector 是带教师傅在旁边盯着 看到他重复 2 次同一个错就喊停 不让他在死胡同里耗。错误分级是问题分类 业务问题让他自己改 系统问题立刻找主管。模型路由是任务分派 简单的让实习生 复杂的找资深 不要事事都找老板。缓存是 知识库 已经回答过的问题不要重新查。HITL 是涉及钱与权限的事必须主管签字 不能让实习生自己拍板。评估集是入职考核 200 道题答对 90% 才能转正。你不能因为实习生聪明就放手让他自由发挥 还要管权限 流程 监督 这才是一整套人岗匹配的工程。
这套架构最难的地方在于 它的复杂度在 demo 阶段几乎完全暴露不了。你写 3 个 tool 跑 LangChain demo 看 Agent 自己决策 觉得 AI 真智能 真好用。但真正生产 月调用 10 万次 各种刁钻问题 各种 tool 组合 各种死循环陷阱 各种成本爆炸风险 各种危险操作误触 你才发现 99% 的复杂度都在 那 1% 的工程细节里。建议任何想做严肃 Agent 应用的团队 上线前一定要做 200 条评估集 模拟一周高并发压测 算清月成本预估 设置危险 tool HITL 任何指标不达标拒绝上线 千万别只看 demo 那只是 Agent 工程的冰山一角 真正生产的复杂度藏在水下 90%。
—— 别看了 · 2026