1. 为什么直接调LangChain的Agent还是不够用

先说背景。我们有个电商售后场景，用户会问“我的订单为什么还没发货”，这需要Agent调用订单查询、物流查询、退款规则三个工具，并且要根据不同阶段的用户情绪选择不同回答策略。我最初直接用LangChain的AgentExecutor，配上OpenAI函数调用，跑通demo没问题——但上了生产就发现两个致命问题：

  • 对话历史无限累积：每个请求都携带全部历史，2000字对话后Token直接爆掉，单次调用成本从0.02美元飙到0.18美元。
  • 工具陷入死循环：当订单状态是“已发货”但物流信息未更新时，Agent会反复调用物流查询工具，最多一次循环了14次，花了12秒才返回超时。

所以这事必须手写。LangChain只保留它作为工具调用链的基座，循环控制、记忆截断、错误恢复全部自己实现。

2. 环境与版本：锁死版本，别追新

我的环境如下，如果你用更新的版本，某些API可能已废弃：

Python 3.10.12
langchain==0.1.0
openai==1.6.1
tiktoken==0.5.1

特别注意：langchain.agents里的AgentExecutor在0.1.0之后已经重构，我用的create_react_agent函数在0.2.x被移除了。所以如果你看到代码报错ModuleNotFoundError，先查版本。

3. 方案设计：三层架构，把控制权拿回自己手里

我的设计思路是三层分离：

  • 工具层：用@tool装饰器包装业务函数，统一输入输出为字符串，方便LLM解析。
  • 记忆管理层：自定义一个CircularBufferMemory类，按Token数裁剪旧消息，而不是按条数。
  • 控制层：手写while循环，包含最大迭代限制、错误重试、结果判定三个逻辑。

核心代码框架如下（节选）：

from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
import tiktoken

@tool
def query_order(order_id: str) -> str:
    """查询订单状态，返回发货时间、物流单号"""
    # 模拟真实API，实际会调HTTP接口
    return f"订单{order_id}已发货，物流单号SF123456"

@tool
def query_logistics(tracking_no: str) -> str:
    """查询物流轨迹，返回最新一条物流信息"""
    # 模拟延迟，实际平均200ms
    return "2024-01-15 10:30 已到达【上海转运中心】"

class CircularBufferMemory:
    def __init__(self, max_tokens=800):
        self.max_tokens = max_tokens
        self.messages = []
        self.encoder = tiktoken.encoding_for_model("gpt-4")

    def add(self, message):
        self.messages.append(message)
        self._trim()

    def _trim(self):
        # 从最旧的消息开始删，直到总Token数低于阈值
        while self._count_tokens() > self.max_tokens and len(self.messages) > 2:
            self.messages.pop(0)

    def _count_tokens(self):
        return sum(len(self.encoder.encode(m.content)) for m in self.messages)

4. 核心实现：手写循环，每一步都可控

这里是最关键的部分。我不用AgentExecutor，而是自己控制循环。逻辑是：

  1. 把当前记忆+用户问题拼成Prompt，发给LLM。
  2. 如果返回Action，就执行对应工具，把结果加入记忆。
  3. 如果返回Final Answer，就退出循环。
  4. 如果超过max_iterations，强制终止并返回最后一条消息。

代码实现：

```python
def run_agent(user_query: str, tools: dict, max_iterations=5):
llm = ChatOpenAI(model="gpt-4-1106-preview", temperature=0)
memory = CircularBufferMemory(max_tokens=800)
memory.add(SystemMessage(content="你是电商客服助手，只能使用提供的工具。"))
memory.add(HumanMessage(content=user_query))

iteration = 0
while iteration  2`保护，并且始终保留第一条SystemMessage。

坑3：循环里异常处理不够细
工具调用可能因为网络超时抛出TimeoutError，也可能因为参数类型错误抛出TypeError。我之前用except Exception一把抓，结果LLM把错误信息当作正常工具结果，导致后续决策混乱。现在分两层：网络错误重试一次，业务错误直接返回给LLM让其换方案。

6. 压测数据与优化对比

用2000条真实咨询测试，对比三个版本：

版本 完成率 平均迭代次数 平均耗时 单次成本
纯Prompt工程 63% - 2.1s $0.04
LangChain AgentExecutor 78% 4.2 3.4s $0.09
本手写Agent 89% 2.1 1.9s $0.05

完成率提升的关键在于错误恢复：当工具返回异常时，我的代码会明确告诉LLM“这个工具不可用，换一个”，而不是让它傻傻重试。另外，把max_iterations从默认的15降到5，不仅没降低完成率，反而减少了无效推理。

7. 总结与适用边界

这套手写方案适合工具数量在3-8个、调用链深度不超过3层的场景。如果你的工具超过10个，建议用LangChain的ToolRouter做分层路由。另外，我的记忆裁剪是按Token数做的，但如果你处理的是长文档，建议改成按语义相关性裁剪，具体可以参考langchain.memory.SummaryBufferMemory的实现。

最后说一句：AutoGPT那套自动拆解任务的想法很好，但在生产环境里，控制比智能更重要。给Agent加上硬性循环上限、错误隔离、预算控制，它才能真正落地。

有问题欢迎评论区交流，代码细节可以私信我发完整版。