1. 问题背景：为什么还要手写Agent？
上个月我负责的客服机器人项目需要接入第三方API——查订单、退换货、改地址。最初直接用LangChain的AgentExecutor + OpenAI Function Calling，跑了一周发现几个问题：
- 工具参数幻觉：模型偶尔会编造不存在的参数（比如把
order_id传成orderId），导致API 400错误率高达32%。 - 记忆无边界：对话超过10轮后，
ConversationBufferMemory会把历史全塞进Prompt，实测单次请求Token从1.8K涨到7.5K，延迟从800ms飙到2.4s。 - 循环失控：当模型反复调用同一个失败工具时，
max_iterations=5根本不够，经常死循环到超时。
AutoGPT的思路（任务分解+工具注册表+自我反思）给了我启发，但它的代码太重——我们只需要一个轻量级、可嵌入现有服务的Agent内核。于是决定基于LangChain的底层原语（BaseTool、BaseChatMessageHistory），手写一个可控的Agent循环。
2. 环境与版本
Python 3.11.8
langchain 0.3.7
langchain-openai 0.2.6
openai 1.40.3
redis 5.0.4 (用于记忆持久化)
模型：gpt-4o-mini-2024-07-18，temperature=0.2。
3. 方案设计：核心架构
整体分为四层：
- 工具注册表（ToolRegistry）：维护工具名 → 工具实例的映射，并提供JSON Schema校验器。
- 记忆管理器（MemoryManager）：基于Redis的
ZSET存储对话历史，按时间戳排序，淘汰最旧消息。 - Agent循环控制器：标准
while循环，包含最大迭代限制、错误捕获、重复调用检测。 - 输出解析器：强制模型以
{"action": "...", "action_input": {...}}格式输出。
关键决策：
- 不用AgentExecutor，改用RunnableLambda封装我们的循环，方便注入自定义逻辑。
- 工具校验用jsonschema库（版本4.22.0），不用LangChain自带的Tool参数校验——那个只检查必填字段，不检查类型。
4. 核心实现：工具定义与注册表
先看工具定义。以查询订单为例：
from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
import json, jsonschema
class OrderQueryInput(BaseModel):
order_id: str = Field(description="订单号，格式如：SO-2024-001")
include_items: bool = Field(default=False, description="是否返回商品明细")
class QueryOrderTool(BaseTool):
name: str = "query_order"
description: str = "根据订单号查询订单状态和金额，当用户提到'订单'、'物流'时使用"
args_schema: type[BaseModel] = OrderQueryInput
def _run(self, order_id: str, include_items: bool = False) -> str:
# 模拟API调用，实际项目中替换为requests.post(...)
if not order_id.startswith("SO-"):
raise ValueError(f"非法订单号格式: {order_id}")
result = {"status": "shipped", "amount": 299.00}
return json.dumps(result, ensure_ascii=False)
async def _arun(self, *args, **kwargs):
return self._run(*args, **kwargs)
这里踩过一个坑：description必须写清楚“什么时候用”，否则模型会在用户问“退款”时去调query_order。后来我加了一个“工具选择提示词”模板，把工具描述按相关度排序，这个问题才缓解。
工具注册表与校验器：
class ToolRegistry:
def __init__(self):
self._tools = {}
self._schemas = {}
def register(self, tool: BaseTool):
schema = tool.args_schema.model_json_schema()
self._schemas[tool.name] = schema
self._tools[tool.name] = tool
def validate_and_call(self, tool_name: str, action_input: dict) -> str:
if tool_name not in self._tools:
return f"错误: 未知工具 '{tool_name}'，可用工具: {list(self._tools.keys())}"
schema = self._schemas[tool_name]
try:
# 关键：jsonschema校验，阻止参数幻觉
jsonschema.validate(instance=action_input, schema=schema)
except jsonschema.ValidationError as e:
# 返回具体错误，让模型自我纠正
return f"参数校验失败: {e.message}。请检查参数类型和必填字段。"
try:
result = self._tools[tool_name].invoke(action_input)
return result
except Exception as e:
return f"工具执行异常: {str(e)}"
注意validate_and_call的返回值——它不是抛出异常，而是返回错误字符串。这样模型在下一次循环中能看到错误信息并自我修正。这是将工具调用失败率从32%降到9%的关键。
5. 核心实现：记忆管理与循环控制
记忆管理使用Redis ZSET，score为Unix时间戳：
import redis, time, json
from typing import List, Dict
class RedisMemoryManager:
def __init__(self, session_id: str, max_messages: int = 12, redis_url: str = "redis://localhost:6379/0"):
self.r = redis.Redis.from_url(redis_url, decode_responses=True)
self.session_key = f"agent:memory:{session_id}"
self.max_messages = max_messages
def add_message(self, role: str, content: str):
msg = {"role": role, "content": content, "ts": time.time()}
self.r.zadd(self.session_key, {json.dumps(msg): time.time()})
self._trim()
def _trim(self):
# 保留最近max_messages条
count = self.r.zcard(self.session_key)
if count > self.max_messages:
# 移除最旧的 (count - max_messages) 条
self.r.zremrangebyrank(self.session_key, 0, count - self.max_messages - 1)
def get_history(self) -> List[Dict]:
raw_msgs = self.r.zrange(self.session_key, 0, -1)
return [json.loads(m) for m in raw_msgs]
这里有一个设计细节：zadd的score用time.time()，但同一毫秒内可能有多条消息，所以score会有冲突。Redis ZSET对于相同score按字典序排列，这会导致顺序错乱。我的解法是：score = time.time() * 1000 + self._seq，_seq是一个自增计数器，确保严格有序。
Agent循环控制器：
def run_agent(user_input: str, session_id: str) -> str:
memory = RedisMemoryManager(session_id)
memory.add_message("user", user_input)
history = memory.get_history()
prompt = build_prompt(history, registry.list_tools())
max_iters = 6
last_error = None
for i in range(max_iters):
response = llm.invoke(prompt)
parsed = parse_action(response.content) # 解析 {"action": "...", "action_input": {...}}
if parsed["action"] == "FINISH":
final_answer = parsed["action_input"]["answer"]
memory.add_message("assistant", final_answer)
return final_answer
if parsed["action"] == "THINK":
# 允许模型输出思考过程，但不计入最终结果
continue
# 重复调用检测
if i > 0 and parsed == last_call:
return "检测到重复调用，终止循环。请重试或换一个问法。"
last_call = parsed
tool_result = registry.validate_and_call(parsed["action"], parsed["action_input"])
# 将工具结果追加到上下文
prompt += f"\n\n工具返回: {tool_result}"
return "达到最大迭代次数(6)，未能完成任务。"
踩坑记录：parse_action最初用json.loads，但模型偶尔输出多余的前缀文字（如“好的，我来处理：{...}”）。后来改用正则r'\{.*\}'提取第一个JSON对象，解析成功率从87%提到99%。
另一个坑：llm.invoke(prompt)每次循环都会重新计算整个Prompt，导致Token浪费。优化方案是使用LangChain的RunnableWithFallbacks + 流式输出，但为了代码简洁，这里暂时用字符串拼接。
6. 效果数据与总结
在MacBook M1（16GB）上，用100条真实客服对话测试：
| 指标 | 使用AgentExecutor | 手写Agent |
|---|---|---|
| 工具调用失败率 | 32% | 9% |
| 平均响应延迟 | 1.8s | 1.2s |
| 单次交互Token消耗 | 5.7K | 2.3K |
| 最大迭代次数 | 5（常超时） | 6（无超时） |
Token消耗降低的主要贡献是记忆淘汰机制——只保留最近12条消息。但如果用户问“我之前说的第一句话是什么”，系统会答不上来。这是一个业务取舍：客服场景下，上下文窗口比长记忆更重要。
最终建议：如果只是简单调用几个API，直接用AgentExecutor就够了；但如果你的工具数量超过5个、或者需要精细控制错误恢复，手写Agent循环是值得的。成本不过300行代码，换来的是完全可控的决策流程。
后续计划：将记忆管理器替换为langchain_community.chat_message_histories.RedisChatMessageHistory，再对比一波性能差异，届时再来更新。
