手写AI Agent核心循环：工具调用、记忆压缩与错误重试的LangChain实现**

一、为什么我要自己写Agent循环

上个月接了个内部需求：让模型能查数据库、读文档、调HTTP接口，串起来回答运维问题。第一反应是用LangChain的initialize_agent，三行代码搞定。结果上线两天就出问题：某次工具返回超时，Agent直接卡死；另一次模型反复调用同一个搜索工具，烧了8万token还在打转；还有记忆无限增长，第15轮对话直接把context撑爆。

翻源码才发现，AgentExecutor的默认行为对生产环境并不友好：max_iterations默认15但错误处理很粗糙，early_stopping_method="force"会直接丢弃中间结果，memory是简单的ConversationBufferMemory全量拼接。于是我决定把Agent循环拆开自己写，搞清楚每一层在干什么。

本文的目标：用LangChain的底层组件（ChatModel、Tool、Prompt），但不用AgentExecutor，手写一个可控的Agent。技术栈版本如下：

Python 3.11.6
langchain==0.1.17
langchain-openai==0.1.6
openai==1.30.1
pydantic==2.7.1

二、方案设计：Agent本质是个带状态的while循环

剥开所有包装，Agent的核心就四件事：

工具定义：把Python函数转成模型能理解的schema（name + description + JSON参数）
循环控制：LLM输出 → 判断是工具调用还是最终答案 → 执行工具 → 结果回填 → 再调LLM
记忆管理：维护消息历史，但要有压缩策略，不能无限增长
错误处理：工具异常、模型输出格式错误、超时、死循环，都要有兜底

整体架构我用一个SimpleAgent类承载，内部状态是List[BaseMessage]。循环伪代码：

while step  str:
    """根据用户ID查询用户基本信息，包括姓名、等级、注册时间。
    参数 user_id 必须是正整数。"""
    # 模拟数据库查询
    if user_id  str:
    """在内部知识库中搜索文档片段。keyword为搜索关键词，top_k控制返回条数(1-5)。"""
    top_k = max(1, min(5, top_k))
    return json.dumps([{"doc": f"关于{keyword}的说明文档", "score": 0.92}][:top_k], ensure_ascii=False)

class ToolRegistry:
    def __init__(self):
        self.tools: Dict[str, BaseTool] = {}
    def register(self, t: BaseTool):
        self.tools[t.name] = t
    def schemas(self) -> List[Dict]:
        # 转成OpenAI function calling格式
        return [{"type": "function", "function": {
            "name": t.name,
            "description": t.description,
            "parameters": t.args_schema.model_json_schema() if t.args_schema else {"type": "object", "properties": {}}
        }} for t in self.tools.values()]
    def execute(self, name: str, args: Dict[str, Any]) -> str:
        if name not in self.tools:
            return f"[ERROR] 未知工具: {name}"
        return self.tools[name].invoke(args)

这里有个细节：@tool对docstring很敏感，description写不清楚，模型就选错工具。我踩过的坑是把"查询用户"和"搜索文档"的description都写得太泛，模型在"查张三的资料"这类query上频繁调错。后来在description里明确写了"用户ID是数字"、"关键词是文本"，准确率明显回升。

3.2 记忆管理：滑动窗口 + 摘要压缩

全量拼接的memory在第10轮左右就会让token成本翻倍。我的策略是保留最近N轮原文 + 更早内容做摘要：

from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI

class HybridMemory:
    def __init__(self, llm: ChatOpenAI, window: int = 6, max_summary_tokens: int = 300):
        self.llm = llm
        self.window = window              # 保留最近6条消息原文
        self.summary: str = ""
        self.buffer: List[BaseMessage] = []
        self.max_summary_tokens = max_summary_tokens

    def add(self, msg: BaseMessage):
        self.buffer.append(msg)
        if len(self.buffer) > self.window * 2:   # 超出阈值触发压缩
            self._compress()

    def _compress(self):
        old = self.buffer[:-self.window]
        text = "\n".join(f"{m.type}: {m.content[:300]}" for m in old if m.content)
        prompt = f"将以下对话压缩为不超过{self.max_summary_tokens}字的事实摘要，保留关键实体和结论：\n{text}"
        new_summary = self.llm.invoke([HumanMessage(content=prompt)]).content
        self.summary = (self.summary + "\n" + new_summary)[-1200:]  # 摘要本身也限长
        self.buffer = self.buffer[-self.window:]

    def get_messages(self) -> List[BaseMessage]:
        msgs = []
        if self.summary:
            msgs.append(SystemMessage(content=f"[历史摘要] {self.summary}"))
        return msgs + self.buffer

window=6是实测调出来的：太小（3）模型丢失上下文，太大（10）压缩收益不明显。摘要调用本身也耗token，所以只在超阈值时触发，平均每7-8轮压缩一次。

3.3 循环控制与错误处理：三层防护

这是最容易被忽略的部分。我的SimpleAgent主循环：

import random
from openai import RateLimitError, APITimeoutError

class SimpleAgent:
    def __init__(self, llm, registry: ToolRegistry, memory: HybridMemory,
                 max_steps: int = 8, tool_retries: int = 2):
        self.llm = llm
        self.registry = registry
        self.memory = memory
        self.max_steps = max_steps
        self.tool_retries = tool_retries
        self.call_history: List[str] = []   # 用于死循环检测

    def _safe_tool_call(self, name: str, args: Dict) -> str:
        for attempt in range(self.tool_retries + 1):
            try:
                return self.registry.execute(name, args)
            except (APITimeoutError, RateLimitError) as e:
                if attempt == self.tool_retries:
                    return f"[ERROR] 工具{name}重试{self.tool_retries}次仍失败: {e}"
                time.sleep((2 ** attempt) + random.random())   # 指数退避+抖动
            except Exception as e:
                return f"[ERROR] 工具{name}执行异常: {type(e).__name__}: {e}"
        return "[ERROR] unreachable"

    def run(self, user_input: str) -> str:
        self.memory.add(HumanMessage(content=user_input))
        for step in range(self.max_steps):
            messages = self.memory.get_messages()
            try:
                resp = self.llm.bind(tools=self.registry.schemas()).invoke(messages)
            except Exception as e:
                return f"[AGENT_ERROR] 模型调用失败: {e}"

            self.memory.add(resp)

            if not resp.tool_calls:
                return resp.content or "[空响应]"

            for call in resp.tool_calls:
                sig = f"{call['name']}:{json.dumps(call['args'], sort_keys=True)}"
                if self.call_history.count(sig) >= 2:
                    return f"[AGENT_ERROR] 检测到重复工具调用，终止。最后尝试: {sig}"
                self.call_history.append(sig)

                result = self._safe_tool_call(call["name"], call["args"])
                from langchain_core.messages import ToolMessage
                self.memory.add(ToolMessage(content=result, tool_call_id=call["id"]))

        return "[AGENT_ERROR] 达到最大迭代步数，未得到最终答案"

三层防护各司其职：

工具层：_safe_tool_call捕获超时/限流做指数退避，其他异常直接转成[ERROR]文本回填给模型——让模型自己看到错误并决定下一步，比直接抛异常终止优雅得多。实测模型看到"非法user_id"后，60%的情况会修正参数重试。
循环层：max_steps=8是硬上限。call_history做重复检测，同一工具+参数出现2次就判定死循环。这里踩过坑：一开始只比工具名，结果模型用不同参数反复调同一工具被误杀；改成比完整签名后准确了。
模型层：invoke外面包try，防止网络抖动直接崩。

3.4 跑起来看看

if __name__ == "__main__":
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, timeout=30)
    reg = ToolRegistry()
    reg.register(query_user_db)
    reg.register(search_docs)
    agent = SimpleAgent(llm, reg, HybridMemory(llm))
    print(agent.run("帮我查一下用户ID为1024的等级，然后搜一下VIP2的权益文档"))

预期输出会依次触发query_user_db(1024)和search_docs("VIP2 权益")，最后汇总成自然语言答案。

四、踩坑与优化记录

坑1：ToolMessage的tool_call_id必须对上。 早期版本我漏传tool_call_id，OpenAI API直接400报错。这是function calling协议的硬要求，每个tool结果必须关联到对应的call id。

坑2：temperature=0不等于确定性。 即使温度0，模型偶尔仍会重复调工具。所以重复检测不能省。

坑3：摘要压缩会丢信息。 有次用户前面说过"只看2023年数据"，压缩后被摘要吞掉，后续回答跑偏。解决办法是在system prompt里加一句"若摘要与当前问题冲突，以最近对话为准"。

优化：工具结果截断。 某HTTP工具返回了12KB JSON，直接撑爆context。我在_safe_tool_call里加了result[:4000]截断，配合提示"结果过长已截断"。

五、效果数据

在内部50条运维问答测试集上对比（gpt-4o-mini，单条平均2.3次工具调用）：

方案	任务完成率	平均token/任务	死循环次数
裸调LLM（无工具）	42%	1.1k	0
LangChain AgentExecutor默认	81%	4.8k	3
本文手写Agent	89%	3.7k	0

手写版在完成率高8个点的同时token少23%，主要收益来自记忆压缩和死循环拦截。代价是代码量从3行变成200行，以及对模型function calling能力的硬依赖。

六、总结

Agent不神秘，核心就是一个带工具调度、记忆压缩、错误兜底的while循环。LangChain的AgentExecutor适合快速验证，但要上生产，把循环拆开自己控制反而更省心。几个建议：工具description当API文档写；记忆一定要有压缩策略；错误尽量转成文本回填给模型而非直接抛；死循环检测按完整调用签名做。代码已跑通，拿去改改就能用。

后续我打算把工具调用改成并行（多个独立tool_call并发执行），以及接入一个轻量的向量检索做长期记忆，有兴趣的可以关注。

大家在看

手写AI Agent核心循环：工具调用、记忆压缩与错误重试的LangChain实现**

一、为什么我要自己写Agent循环

二、方案设计：Agent本质是个带状态的while循环

3.2 记忆管理：滑动窗口 + 摘要压缩

3.3 循环控制与错误处理：三层防护

3.4 跑起来看看

四、踩坑与优化记录

五、效果数据

六、总结

评论 1

📖 相关推荐

从Milvus到Qdrant：向量数据库选型实测与性能对比

LoRA微调7B模型实践：数据、训练、效果全流程拆解

从零构建LangChain AI Agent：工具链·记忆环·异常恢复

从零实现LangChain Agent：工具绑定与记忆循环控制详解