一、为什么我要手写一个Agent

先说结论：LangChain的Agent抽象层在快速原型阶段确实好用，但当你需要精确控制prompt结构、调试LLM返回格式异常、或者优化token消耗时，那层封装反而成了障碍。我最近在一个内部知识库问答项目里用AgentExecutor，遇到三个典型问题：一是max_iterations触发后直接抛异常而不是优雅返回中间结果；二是ConversationBufferMemory在长对话下token暴涨到8000+；三是工具参数解析失败时LangChain的重试逻辑不透明，日志里根本看不出LLM到底返回了什么。

所以我决定把Agent拆开，用最原始的方式手写一遍。AutoGPT的思路其实很朴素——就是让LLM在一个循环里不断"思考→行动→观察"，直到给出最终答案。LangChain的ReAct实现本质上也是这个套路。理解了这一层，你就能按自己的需求定制任何环节。

本文的Agent基于ReAct（Reasoning + Acting）范式，核心循环如下：

用户输入 → LLM推理(Thought) → 选择工具(Action) → 执行工具 → 获得观察结果(Observation) → 拼回上下文 → 再次推理 → ... → 最终答案(Final Answer)

下面逐层拆解实现。

二、环境与版本

  • Python 3.11.6
  • openai 1.30.0（使用新版client.chat.completions.create接口）
  • pydantic 2.7.1（用于工具参数schema校验）
  • 模型：gpt-4o-mini（temperature=0，top_p=1）

选gpt-4o-mini是因为Agent循环里LLM调用次数多，成本敏感。实测在ReAct格式遵循上，4o-mini比3.5-turbo稳定不少，格式错误率从12%降到4%左右。

三、方案设计

整体分四个模块：

  1. ToolRegistry：工具注册中心，每个工具包含name、description、参数JSON Schema、执行函数。description的质量直接决定LLM选工具的准确率，后面会细说。
  2. Memory：对话记忆管理，采用滑动窗口+摘要压缩的混合策略，控制总token在2000以内。
  3. OutputParser：解析LLM返回的ReAct格式文本，提取Thought/Action/Action Input/Final Answer。
  4. AgentLoop：主循环控制器，管理最大迭代次数、错误重试、超时降级。

Prompt采用经典的ReAct模板，但做了两点改造：一是在system prompt里强制要求"每次只能输出一个Action"；二是把工具列表用JSON Schema形式注入，比自然语言描述更精确。

四、核心实现

4.1 工具定义与注册

先看工具的数据结构。我用pydantic做参数校验，这样LLM传错参数时能在执行前拦截：

import json
import time
from typing import Callable, Any
from pydantic import BaseModel, ValidationError, create_model

class Tool:
    def __init__(self, name: str, description: str, 
                 args_schema: type[BaseModel], func: Callable):
        self.name = name
        self.description = description
        self.args_schema = args_schema
        self.func = func

    def run(self, raw_args: str) -> str:
        """执行工具，带参数校验和异常捕获"""
        try:
            args_dict = json.loads(raw_args)
        except json.JSONDecodeError as e:
            return f"[ToolError] 参数不是合法JSON: {e}. 请重新生成Action Input."

        try:
            validated = self.args_schema(**args_dict)
        except ValidationError as e:
            return f"[ToolError] 参数校验失败: {e.errors()}. 请检查参数类型."

        try:
            start = time.time()
            result = self.func(**validated.model_dump())
            elapsed = time.time() - start
            # 截断过长结果，防止撑爆上下文
            result_str = str(result)
            if len(result_str) > 1500:
                result_str = result_str[:1500] + "...[结果已截断]"
            return f"[ToolSuccess {elapsed:.2f}s] {result_str}"
        except Exception as e:
            return f"[ToolError] 执行异常: {type(e).__name__}: {e}"


class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, Tool] = {}

    def register(self, tool: Tool):
        self._tools[tool.name] = tool

    def get(self, name: str) -> Tool | None:
        return self._tools.get(name)

    def describe(self) -> str:
        """生成注入prompt的工具描述"""
        lines = []
        for t in self._tools.values():
            schema = t.args_schema.model_json_schema()
            lines.append(
                f"- {t.name}: {t.description}\n"
                f"  参数schema: {json.dumps(schema.get('properties', {}), ensure_ascii=False)}"
            )
        return "\n".join(lines)

注意run方法里把JSON解析错误和参数校验错误都包装成字符串返回给LLM，而不是抛异常。这是关键设计——Agent的错误应该成为Observation的一部分，让LLM自己纠正，而不是让程序崩溃。

注册两个示例工具：

from pydantic import Field

class SearchArgs(BaseModel):
    query: str = Field(description="搜索关键词，中文或英文")

class CalcArgs(BaseModel):
    expression: str = Field(description="数学表达式，如 (3+5)*2")

def web_search(query: str) -> str:
    # 实际项目替换为真实搜索API
    mock_db = {"天气": "北京今天晴，25度", "langchain": "LangChain是一个LLM应用框架"}
    for k, v in mock_db.items():
        if k in query.lower():
            return v
    return f"未找到'{query}'的相关结果"

def calculator(expression: str) -> str:
    # 生产环境务必用安全解析，此处仅演示
    allowed = set("0123456789+-*/(). ")
    if not set(expression) = self.max_recent_turns * 2:
            self._compress()

    def _compress(self):
        """把最早的两轮压成摘要占位，实际项目可调LLM摘要"""
        overflow = self.recent[:2]
        self.recent = self.recent[2:]
        text = " | ".join(f"{m['role']}: {m['content'][:80]}" for m in overflow)
        self.summary = (self.summary + " || " + text)[-500:]  # 摘要本身也限长

    def build_context(self) -> list[dict]:
        ctx = []
        if self.summary:
            ctx.append({"role": "system", "content": f"[历史对话摘要] {self.summary}"})
        ctx.extend(self.recent)
        return ctx

    def approx_tokens(self) -> int:
        total = sum(len(m["content"]) for m in self.recent) + len(self.summary)
        return total // 2  # 中文粗估：2字符≈1token

这套策略下，20轮对话的上下文稳定在1800 token左右，比裸Buffer方案节省约70%。

4.3 输出解析与主循环

ReAct格式解析是错误高发区。LLM经常多输出一行解释、少写Action Input:、或者把JSON写成单引号。我的解析器做了容错处理，配合主循环的重试机制：

import re

REACT_PROMPT = """你是一个AI助手，可以使用以下工具：
{tool_desc}

请严格按以下格式回复，每次只输出一个步骤：
Thought: 你的思考过程
Action: 工具名，必须是 [{tool_names}] 之一
Action Input: 工具参数的JSON对象

当你得出最终答案时，使用：
Thought: 我已获得足够信息
Final Answer: 你的最终回答

用户问题: {question}
{history}
"""

def parse_react_output(text: str) -> dict:
    """解析LLM输出，容错处理常见格式偏差"""
    result = {"thought": "", "action": None, "action_input": None, "final": None}

    m = re.search(r"Thought:\s*(.*?)(?=\n(?:Action|Final)|\Z)", text, re.S)
    if m:
        result["thought"] = m.group(1).strip()

    m = re.search(r"Final Answer:\s*(.*)", text, re.S)
    if m:
        result["final"] = m.group(1).strip()
        return result

    m = re.search(r"Action:\s*(\w+)", text)
    if m:
        result["action"] = m.group(1).strip()

    m = re.search(r"Action Input:\s*(.*)", text, re.S)
    if m:
        raw = m.group(1).strip()
        # 容错：去除markdown代码块包裹
        raw = re.sub(r"^```(?:json)?\s*|\s*```$", "", raw, flags=re.S).strip()
        result["action_input"] = raw

    return result


class SimpleAgent:
    def __init__(self, registry: ToolRegistry, memory: AgentMemory,
                 max_iterations: int = 6, max_retries: int = 2):
        self.registry = registry
        self.memory = memory
        self.max_iterations = max_iterations
        self.max_retries = max_retries
        from openai import OpenAI
        self.client = OpenAI()  # 读取 OPENAI_API_KEY 环境变量

    def _call_llm(self, messages: list[dict]) -> str:
        resp = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            temperature=0,
            max_tokens=512,
        )
        return resp.choices[0].message.content or ""

    def run(self, question: str) -> str:
        self.memory.add("user", question)
        scratchpad = ""  # 本轮工具调用轨迹，不写入长期记忆

        for i in range(self.max_iterations):
            prompt = REACT_PROMPT.format(
                tool_desc=self.registry.describe(),
                tool_names=", ".join(self.registry._tools.keys()),
                question=question,
                history=scratchpad,
            )
            messages = [{"role": "system", "content": prompt}] + self.memory.build_context()

            # LLM调用带重试
            raw_output = None
            for attempt in range(self.max_retries + 1):
                try:
                    raw_output = self._call_llm(messages)
                    break
                except Exception as e:
                    if attempt == self.max_retries:
                        return f"[AgentError] LLM调用失败: {e}"
                    time.sleep(1.5 ** attempt)  # 指数退避

            parsed = parse_react_output(raw_output)

            # 情况1：得到最终答案
            if parsed["final"]:
                self.memory.add("assistant", parsed["final"])
                return parsed["final"]

            # 情况2：格式异常，把错误反馈给LLM重试
            if not parsed["action"] or not parsed["action_input"]:
                scratchpad += f"\n[格式错误] 你的输出不符合ReAct格式，请严格使用 Action:/Action Input: 标签。原始输出片段: {raw_output[:120]}\n"
                continue

            # 情况3：未知工具
            tool = self.registry.get(parsed["action"])
            if tool is None:
                scratchpad += f"\n[工具不存在] '{parsed['action']}' 不在可用列表中，请从 [{', '.join(self.registry._tools.keys())}] 中选择。\n"
                continue

            # 正常执行工具
            observation = tool.run(parsed["action_input"])
            scratchpad += (
                f"\nThought: {parsed['thought']}"
                f"\nAction: {parsed['action']}"
                f"\nAction Input: {parsed['action_input']}"
                f"\nObservation: {observation}\n"
            )

        # 达到最大迭代：返回已收集的中间结果而非抛异常
        fallback = f"[已达到最大迭代次数{self.max_iterations}] 基于当前信息的部分结论:\n{scratchpad[-800:]}"
        self.memory.add("assistant", fallback)
        return fallback

五、踩坑与优化

坑1：工具描述太模糊导致选错工具。 最初calculator的描述只写了"计算器"，LLM在问"3的平方加5"时居然去调了web_search。改成"执行数学计算，处理加减乘除和括号运算"后，选工具准确率从78%提升到96%。工具description的写法建议：动词开头+明确适用场景+明确不适用场景。

坑2：max_iterations耗尽后抛异常。 LangChain默认行为是抛ValueError，用户看到的是500错误。我改成返回中间轨迹的最后800字符，用户体验好很多，而且调试时能直接看到卡在哪一步。

坑3：工具返回结果过长。 有一次搜索返回了4000字的网页正文，直接把下一轮prompt撑爆。加了1500字符截断后解决。如果你的工具确实需要返回长文本，建议在工具内部先做摘要再返回。

坑4：temperature不要设太高。 Agent场景下temperature>0.3会导致ReAct格式稳定性明显下降。我实测temperature=0.7时格式错误率是0的3倍。这个场景要的是确定性，不是创造性。

优化点：scratchpad与长期记忆分离。 工具调用轨迹只存在于当前轮的scratchpad里，不写入memory。否则多轮对话时历史工具调用会污染上下文，导致LLM混淆当前问题。这是一个容易被忽略的设计细节。

六、效果数据

在内部50条测试用例（含单工具调用、多工具串联、无需工具直接回答三类）上的实测数据：

指标 数值
任务完成率 88%（44/50）
工具调用成功率 94.7%
平均迭代轮次 2.3轮
单轮平均延迟 3.2s（gpt-4o-mini）
上下文token峰值 1850（20轮对话压测）
格式解析失败率 4.2%

失败的6条里，4条是多步推理中LLM提前给出Final Answer（信息不全），2条是复杂表达式的calculator参数格式错误。前者可以通过在prompt里加"确认信息充分后再给Final Answer"缓解，后者需要更强的参数schema约束。

七、总结

手写一遍Agent之后，我对LangChain那些封装的理解反而更深了。核心就四件事：工具schema要精确、记忆要做压缩、解析要容错、循环要有兜底。这四块任何一块偷懒，生产环境都会教你做人。

代码总量不到300行，但每一行都是踩坑踩出来的。如果你也在做Agent开发，建议先用这个最小实现跑通业务逻辑，再决定要不要引入框架。很多时候，你需要的是可控性，而不是抽象层。

后续可以扩展的方向：并行工具调用（一次返回多个Action）、工具结果的向量化检索回填、以及基于token预算的动态记忆淘汰策略。这些等我下一个迭代再来填坑。