一、为什么我要自己写一个Agent
先说结论：LangChain的initialize_agent和AutoGPT这类框架，开箱即用是真的，但出了问题你基本是瞎的。
上个月我们内部要做一个"运维问答助手"，需求很朴素：用户问"昨天订单服务的错误率为什么涨了"，Agent要能自己决定先查监控、再查日志、最后查发布记录，然后综合回答。一开始我直接用LangChain的AgentExecutor + create_openai_tools_agent，两天跑通demo，然后上线测试，问题全来了：
- 工具参数传错格式，模型给了个字符串，工具要dict，直接抛异常，整个链路挂掉；
- 对话超过8轮后，prompt token飙到6000+，成本和延迟都顶不住；
- 有一次模型陷入"调用工具A→结果不满意→再调用工具A"的死循环，烧了12万token才被我们手动kill；
- 工具内部报错（比如监控API 502），异常直接冒泡，用户看到一个500。
这些问题框架不是不能配，但配置项散落在max_iterations、handle_parsing_errors、memory、early_stopping_method各处，文档还经常对不上版本。于是我决定手写一个，把控制权拿回来。最终成品约180行核心代码，跑在LangChain 0.1.16 + openai 1.30.1 + Python 3.11，下面把实现细节掰开讲。
二、环境与版本
先把版本钉死，LangChain 0.1.x和0.2.x的BaseTool签名有差异，别混用：
python==3.11.8
langchain==0.1.16
langchain-core==0.1.46
langchain-openai==0.1.7
openai==1.30.1
pydantic==2.7.1
tiktoken==0.6.0
模型用gpt-3.5-turbo-0125（便宜、够用），temperature=0保证工具调用稳定。如果你用GPT-4o，下面所有逻辑不变，只是延迟数字会变。
三、方案设计：ReAct + 显式状态机
我选的是ReAct范式（Reasoning + Acting），但不用LangChain的agent executor，而是自己写一个显式循环。整体结构：
用户输入 → 拼装system prompt(工具描述+记忆) → LLM
↓
LLM返回：要么 final_answer，要么 tool_call
↓ (tool_call)
参数校验 → 执行工具 → 捕获异常 → 结果写回记忆 → 回到LLM
↓ (final_answer 或 达到max_steps)
输出
四个核心模块对应四个类：ToolRegistry（工具定义与校验）、MemoryManager（记忆管理）、AgentLoop（循环控制）、RetryPolicy（错误处理）。这样拆的好处是每一块都能单独单测，出问题一眼定位。
# agent_core.py 骨架
from dataclasses import dataclass, field
from typing import Any
import json, time, logging
log = logging.getLogger("agent")
@dataclass
class Step:
role: str # "user" / "assistant" / "tool"
content: str
tool_name: str | None = None
tool_input: dict | None = None
ts: float = field(default_factory=time.time)
class ToolRegistry:
def __init__(self): self._tools: dict[str, dict] = {}
def register(self, name, func, description, schema): ...
def call(self, name, raw_args) -> str: ... # 含校验+异常捕获
class MemoryManager:
def __init__(self, max_tokens=2500, keep_recent=6): ...
def add(self, step: Step): ...
def build_context(self) -> list[dict]: ... # 滑动窗口
class AgentLoop:
def __init__(self, llm, tools, memory, max_steps=8): ...
def run(self, user_input: str) -> str: ...
四、核心实现
4.1 工具定义：JSON Schema + 显式校验
工具定义我不用LangChain的@tool装饰器，因为它对参数校验太松（args_schema不匹配时经常静默通过）。我要求每个工具必须带一份JSON Schema，调用前用jsonschema校验：
import jsonschema
class ToolRegistry:
def __init__(self):
self._tools = {}
def register(self, name: str, func, description: str, schema: dict):
jsonschema.Draft7Validator.check_schema(schema)
self._tools[name] = {"func": func, "desc": description, "schema": schema}
def specs(self) -> list[dict]:
# 转成 OpenAI function calling 格式
return [{
"type": "function",
"function": {"name": n, "description": t["desc"], "parameters": t["schema"]}
} for n, t in self._tools.items()]
def call(self, name: str, raw_args: str) -> str:
t = self._tools.get(name)
if not t:
return f"[ERROR] 未知工具: {name}，可用工具: {list(self._tools)}"
try:
args = json.loads(raw_args) if isinstance(raw_args, str) else raw_args
jsonschema.validate(args, t["schema"])
except (json.JSONDecodeError, jsonschema.ValidationError) as e:
return f"[ERROR] 工具 {name} 参数非法: {e}. 请修正后重试。"
try:
return str(t["func"](**args))[:2000] # 截断，防止爆token
except Exception as e:
log.exception("tool %s failed", name)
return f"[ERROR] 工具 {name} 执行失败: {type(e).__name__}: {e}"
关键设计：工具永远不抛异常给上层，而是把错误作为字符串返回给LLM。这是Agent和自我修复能力的来源——模型看到[ERROR] 参数非法，下一轮通常会自己改对。实测这一步让参数错误导致的整链路失败率从17%降到0。
注册示例：
tools = ToolRegistry()
tools.register(
name="query_metrics",
func=lambda service, metric, window="1h": fake_prometheus(service, metric, window),
description="查询指定服务在时间窗口内的监控指标，如错误率、QPS、P99延迟",
schema={
"type": "object",
"properties": {
"service": {"type": "string", "description": "服务名，如 order-service"},
"metric": {"type": "string", "enum": ["error_rate", "qps", "p99_latency"]},
"window": {"type": "string", "pattern": r"^\d+[mh]$", "default": "1h"}
},
"required": ["service", "metric"]
}
)
注意schema里我用了enum和pattern，模型看到这些约束后参数合规率明显更高（实测从72%到94%）。
4.2 记忆管理：token预算下的滑动窗口 + 工具结果压缩
记忆不能无脑append，gpt-3.5-turbo 16K上下文，多轮工具调用很容易顶到上限。我的策略：
- 用
tiktoken精确算token，预算2500； - 保留最近6条消息（
keep_recent），保证当前推理的连贯性； - 更早的
tool消息压缩成一行摘要（工具名+结果前80字符）； - system prompt里始终保留"已调用过的工具清单"，防止模型重复调用。
import tiktoken
enc = tiktoken.encoding_for_model("gpt-3.5-turbo")
class MemoryManager:
def __init__(self, max_tokens=2500, keep_recent=6):
self.steps: list[Step] = []
self.max_tokens = max_tokens
self.keep_recent = keep_recent
def _count(self, msgs) -> int:
return sum(len(enc.encode(m["content"])) for m in msgs)
def add(self, step: Step):
self.steps.append(step)
def build_context(self) -> list[dict]:
recent = self.steps[-self.keep_recent:]
older = self.steps[:-self.keep_recent]
# 老的工具结果压成一行
compressed = []
for s in older:
if s.role == "tool":
compressed.append({"role": "system",
"content": f"[历史] {s.tool_name} -> {s.content[:80]}"})
else:
compressed.append({"role": s.role, "content": s.content[:200]})
msgs = compressed + [{"role": s.role, "content": s.content} for s in recent]
# 超预算从最老的开始丢
while self._count(msgs) > self.max_tokens and len(msgs) > 2:
msgs.pop(0)
return msgs
这个策略在20条测试用例上，平均prompt token从4100降到1900，单次调用成本降了约54%，而任务完成率只掉了2个百分点（87%→85%），我认为划算。
4.3 循环控制：max_steps + 重复调用检测
循环控制两个保险：硬上限max_steps=8，以及软检测——连续两次调用同一工具且参数完全相同，直接判定死循环，强制让模型给最终答案。
class AgentLoop:
def __init__(self, llm, tools: ToolRegistry, memory: MemoryManager,
max_steps=8, max_retries=3):
self.llm, self.tools, self.memory = llm, tools, memory
self.max_steps, self.max_retries = max_steps, max_retries
def _call_llm(self, messages):
for i in range(self.max_retries):
try:
return self.llm.invoke(messages)
except Exception as e:
wait = 2 ** i # 1s, 2s, 4s 指数退避
log.warning("LLM call failed (%s), retry in %ss", e, wait)
time.sleep(wait)
raise RuntimeError("LLM 调用重试耗尽")
def run(self, user_input: str) -> str:
self.memory.add(Step(role="user", content=user_input))
last_sig = None
for step_i in range(self.max_steps):
msgs = [{"role": "system", "content": SYSTEM_PROMPT}] + self.memory.build_context()
resp = self._call_llm(msgs)
msg = resp
# 1) 模型直接给答案
if not getattr(msg, "tool_calls", None):
self.memory.add(Step(role="assistant", content=msg.content))
return msg.content
# 2) 处理工具调用
for tc in msg.tool_calls:
sig = (tc.function.name, tc.function.arguments)
if sig == last_sig:
log.warning("检测到重复调用，强制收敛")
self.memory.add(Step(role="system",
content="你已重复调用同一工具，请基于现有信息直接给出最终答案。"))
break
last_sig = sig
result = self.tools.call(tc.function.name, tc.function.arguments)
self.memory.add(Step(role="tool", content=result,
tool_name=tc.function.name))
self.memory.add(Step(role="assistant",
content=f"调用 {tc.function.name}，结果已记录。"))
# 3) 达到步数上限
return "[未收敛] 已达到最大推理步数，请缩小问题范围或补充信息。"
SYSTEM_PROMPT里我明确写了"最多调用工具8次，信息足够就回答"，配合硬上限，双保险。
五、踩坑与优化
坑1：tool_calls的arguments是字符串不是dict。 OpenAI返回的tc.function.arguments是JSON字符串，直接**解包会TypeError，必须先json.loads，且模型偶尔会输出带尾逗号的非法JSON。我在ToolRegistry.call里做了兜底，解析失败就把原始字符串和错误一起返回给模型，让它重试——比直接崩溃好太多。
坑2：并行tool_calls。 模型有时一次返回2-3个tool_call，我一开始只处理第一个，导致状态不一致。正确做法是遍历msg.tool_calls全部执行，并且每个结果都要以role="tool"消息回填（OpenAI要求tool消息必须紧跟assistant的tool_calls，顺序不能乱）。
坑3：记忆里的assistant消息格式。 如果你把带tool_calls的assistant消息原样塞回context，某些版本会报400。我的简化方案是回填一条纯文本"调用X，结果已记录"，牺牲一点保真度换取稳定性，实测对效果影响很小。
坑4：延迟。 单次GPT-3.5调用约1.1s，4轮工具调用就是4-5s。优化手段：工具内部加10分钟本地缓存（functools.lru_cache改造成带TTL的），重复查询直接命中；prompt压缩（见4.2）。P95从6.1s降到3.8s。
六、效果数据
20条多步运维查询测试集（每条平均需要2-4次工具调用），对比LangChain AgentExecutor默认配置：
| 指标 | LangChain默认 | 手写Agent |
|---|---|---|
| 任务完成率 | 87% | 85% |
| 参数错误致失败 | 17% | 0% |
| 死循环发生次数 | 3/20 | 0/20 |
| 平均prompt token | 4100 | 1900 |
| P95延迟 | 6.1s | 3.8s |
| 代码可控性 | 黑盒 | 全白盒 |
完成率略低2个点，是因为我砍了部分历史上下文，属于预期内的取舍；但稳定性、成本和可调试性提升明显，线上跑了三周没再出现烧token的事故。
七、总结
手写Agent的核心不是"调LLM"，而是把不确定性关进笼子：工具层用JSON Schema把参数关进笼子，记忆层用token预算把上下文关进笼子，循环层用max_steps和重复检测把死循环关进笼子，错误处理层把异常转成模型能读懂的反馈。这四件事做好了，180行代码就够用，而且每一行你都看得懂、改得动。框架该用还得用，但至少下次它出问题时，你知道该去哪个文件里找答案。
代码已整理成单文件，依赖装好即可跑，建议你先用假工具跑通循环，再替换成真实API。
