Agent开发框架对比:LangChain vs 原生实现——一个RAG系统的真实重构
背景:从“玩具Agent”到生产级RAG
某电商客服系统需要构建一个Agent,能根据用户问题查询订单状态、退换货规则,并调用外部物流API。初期团队使用LangChain快速搭建原型,但上线后发现:对话上下文丢失、工具调用超时、追踪困难。这迫使团队重新审视框架选择——LangChain的抽象是否值得引入,还是用原生Python更可控?
架构设计:两种方案的对比框架
LangChain方案(初始版本)
- 组件链:
LLMChain+RetrievalQA+Tool,通过AgentExecutor串联。 - 工具定义:用
@tool装饰器封装API调用。 - 记忆机制:
ConversationSummaryMemory+BufferWindowMemory。
原生实现方案(重构版本)
- 核心循环:手动实现
while loop,每次迭代打印当前状态(工具名、参数、结果)。 - 工具注册:用
dict[str, callable]管理,每个工具返回(status, data)元组。 - 状态管理:Python原生
list存储消息历史,每次调用前截断到最近N轮。
关键对比:三个真实踩坑点
1. 工具调用的超时与重试
问题:LangChain的AgentExecutor默认不处理工具超时。当物流API响应超过5秒,Agent会卡死,最终抛出TimeoutError导致整个对话失败。
原生实现解决:
import asyncio
async def call_tool_with_retry(tool_name, params, max_retries=3, timeout=5):
for attempt in range(max_retries):
try:
result = await asyncio.wait_for(
tools[tool_name](**params), timeout=timeout
)
return result
except asyncio.TimeoutError:
if attempt == max_retries - 1:
return {"status": "error", "message": "第三方API超时"}
await asyncio.sleep(0.5) # 退避
关键点:asyncio.wait_for提供了精确的毫秒级超时控制。LangChain的Tool类虽然支持return_direct=True,但无法在工具内部实现重试逻辑——必须改写AgentExecutor的_atool_call方法,这反而破坏了框架的“开箱即用”承诺。
2. 上下文管理:记忆窗口的坑
问题:LangChain的ConversationSummaryMemory在对话超过10轮后,摘要会丢失细节(如用户提到的具体SKU编号)。而BufferWindowMemory截断后,Agent无法理解“刚才那单”指的是哪个订单。
原生实现解决:
class SlidingWindowMemory:
def __init__(self, max_rounds=6, max_tokens=2000):
self.history = []
self.max_rounds = max_rounds
self.max_tokens = max_tokens
def add(self, user_msg, assistant_msg, tool_calls=None):
# 存储原始消息,附带工具调用记录
self.history.append({
"user": user_msg,
"assistant": assistant_msg,
"tools": tool_calls or [] # 关键:保留工具调用上下文
})
if len(self.history) > self.max_rounds:
self.history = self.history[-self.max_rounds:]
关键点:原生实现可以精确控制保留哪些字段(如工具调用的原始参数),而LangChain的Memory类将消息和工具调用合并为BaseMessage对象,导致无法区分“用户说”和“工具返回”。这个差异在需要将工具结果回传给LLM时尤其致命——LangChain需要额外配置return_messages=True,且无法在记忆对象中按时间戳排序。
3. 追踪与调试:黑盒 vs 白盒
问题:LangChain的AgentExecutor每次迭代输出的是已格式化的AgentAction和AgentFinish,无法在中间步骤打印变量。当工具返回错误时,只能看到“工具调用失败”,看不到具体参数。
原生实现解决:
def agent_loop(user_input, tools, llm, memory, max_steps=5):
memory.add(user_input, "", [])
for step in range(max_steps):
# 每次迭代打印当前状态
print(f"[STEP {step}] memory length: {len(memory.history)}")
# 手动构造提示词,包含工具描述和记忆
prompt = build_prompt(memory, tools)
response = llm.invoke(prompt)
action = parse_action(response)
if action["type"] == "final":
memory.add("", action["answer"], [])
return action["answer"]
# 调用工具并打印参数
print(f"[TOOL] {action['name']} with params: {action['params']}")
result = call_tool_with_retry(action["name"], action["params"])
print(f"[RESULT] {result}")
# 将结果追加到记忆
memory.add("", f"tool_{action['name']}_result", [action])
关键点:原生实现通过print语句在循环中暴露每一步的变量,这在调试阶段可以将log_level设为DEBUG。LangChain虽然支持callbacks(如StdOutCallbackHandler),但回调只能拿到Chain和Tool的输入输出,无法在工具调用前插入断点或修改参数——这在需要动态调整工具参数(如根据用户IP切换API端点)时,必须重写Tool的_run方法,比原生实现更复杂。
性能对比:数字背后的真相
| 指标 | LangChain | 原生实现 | 差异原因 |
|---|---|---|---|
| 首次响应时间 | 1.2s | 0.8s | LangChain的AgentExecutor需要序列化/反序列化工具描述 |
| 连续对话稳定率 | 78% | 92% | 原生实现的记忆窗口可精确控制上下文 |
| 代码行数 | 150行 | 280行 | LangChain的抽象节省了工具注册和循环代码 |
注意:数据来自内部压测(1000次对话,单轮调用3个工具)。原生实现的代码行数多,但每行都是可调试的——LangChain的“少代码”本质是“隐藏复杂度”,而非“减少复杂度”。
踩坑总结与建议
- 不要用LangChain的默认Agent:当工具数量超过5个或API有重试需求时,原生实现更可控。LangChain的
StructuredOutputParser在多工具场景下容易返回格式错误,且修复成本高。 - 记忆是核心瓶颈:LangChain的
Memory类设计面向对话,而非Agent。原生实现应自行管理history结构,建议用dataclass存储每条消息的role、content、tool_calls字段。 - 调试是框架选择的第一指标:如果团队没有完善的APM工具(如OpenTelemetry),原生实现的自定义
print和断点调试远比LangChain的回调机制高效。
下一步行动建议
对于新项目,建议先用原生Python实现一个最小Agent循环(约100行),验证工具调用、记忆管理、错误处理三个核心流程。只有当需要多LLM切换、复杂的Prompt模板(如Few-shot示例)、或分布式Agent编排时,再引入LangChain的Hub和Runnable组件——并且只引入具体组件,避免使用AgentExecutor这个“全家桶”。记住:框架是加速器,不是救生衣。能徒手写Agent的团队,才有资格选择框架。
本文关键词:Agent开发框架、LangChain、RAG、工具调用、记忆管理