从零手写AI Agent：基于LangChain与AutoGPT范式的工具、记忆与循环控制实现

python 3.11.9
langchain-core==0.3.28
langchain-openai==0.2.14
openai==1.59.0
pydantic==2.10.4
tiktoken==0.8.0

┌─────────────────────────────────────┐
│           Agent Loop (主循环)         │
│  while step  list[dict]:
        """把注册的工具转成 OpenAI tools 格式"""
        schemas = []
        for spec in self._tools.values():
            # 从函数签名提取参数模型
            sig = inspect.signature(spec.func)
            fields = {}
            for pname, p in sig.parameters.items():
                ann = p.annotation if p.annotation is not inspect.Parameter.empty else str
                fields[pname] = (ann, Field(..., description=f"参数 {pname}"))
            ArgsModel = create_model(f"{spec.name}_args", **fields)
            schemas.append({
                "type": "function",
                "function": {
                    "name": spec.name,
                    "description": spec.description,
                    "parameters": ArgsModel.model_json_schema(),
                }
            })
        return schemas

    def execute(self, name: str, args: dict) -> str:
        spec = self._tools.get(name)
        if not spec:
            return f"[ERROR] 未知工具: {name}"
        last_err = None
        for attempt in range(spec.max_retries + 1):
            try:
                result = spec.func(**args)
                return str(result)[:2000]  # 截断，防止 observation 撑爆上下文
            except Exception as e:
                last_err = e
                time.sleep(0.3 * (attempt + 1))  # 简单退避
        return f"[ERROR] 工具 {name} 执行失败: {type(last_err).__name__}: {last_err}"

# memory.py
import tiktoken
from langchain_openai import ChatOpenAI

ENC = tiktoken.encoding_for_model("gpt-4o-mini")

class MemoryManager:
    def __init__(self, llm: ChatOpenAI, window: int = 6,
                 summary_threshold: int = 3000):
        self.llm = llm
        self.window = window
        self.threshold = summary_threshold
        self.summary: str = ""
        self.history: list[dict] = []  # 原始消息

    def _count(self, msgs: list[dict]) -> int:
        return sum(len(ENC.encode(str(m.get("content", "")))) for m in msgs)

    def add(self, msg: dict):
        self.history.append(msg)
        if len(self.history) > self.window * 2:
            self._maybe_compress()

    def _maybe_compress(self):
        early = self.history[:-self.window]
        if self._count(early)  list[dict]:
        msgs = [{"role": "system", "content": system_prompt}]
        if self.summary:
            msgs.append({"role": "system", "content": f"[历史摘要] {self.summary}"})
        return msgs + self.history

# agent.py
import json, hashlib
from langchain_openai import ChatOpenAI
from tools import ToolRegistry
from memory import MemoryManager

SYSTEM_PROMPT = """你是一个工具调用型 Agent。规则：
1. 需要外部信息时调用工具，不要编造。
2. 每次只调用一个工具，拿到结果后再决定下一步。
3. 信息足够时直接给出最终答案，格式为 "FINAL: "。
4. 不要重复调用相同参数的工具。"""

class SimpleAgent:
    def __init__(self, llm: ChatOpenAI, tools: ToolRegistry,
                 max_steps: int = 8):
        self.llm = llm.bind_tools(tools.openai_schemas())
        self.tools = tools
        self.memory = MemoryManager(llm)
        self.max_steps = max_steps

    def run(self, user_input: str) -> str:
        self.memory.add({"role": "user", "content": user_input})
        last_fingerprint, err_streak, repeated = None, 0, 0

        for step in range(1, self.max_steps + 1):
            msgs = self.memory.build_messages(SYSTEM_PROMPT)
            resp = self.llm.invoke(msgs)

            # 无工具调用 → 最终答案
            if not resp.tool_calls:
                self.memory.add({"role": "assistant", "content": resp.content})
                return resp.content

            for tc in resp.tool_calls:
                name, args = tc["name"], tc["args"]
                fp = hashlib.md5(
                    f"{name}:{json.dumps(args, sort_keys=True)}".encode()
                ).hexdigest()
                if fp == last_fingerprint:
                    repeated += 1
                    if repeated >= 1:
                        print(f"[Loop] 检测到重复动作 {name}，强制终止")
                        return self._force_answer(msgs)
                last_fingerprint, repeated = fp, 0

                obs = self.tools.execute(name, args)
                if obs.startswith("[ERROR]"):
                    err_streak += 1
                else:
                    err_streak = 0

                self.memory.add({"role": "assistant", "content": "",
                                 "tool_calls": [tc]})
                self.memory.add({"role": "tool", "tool_call_id": tc["id"],
                                 "content": obs})
                print(f"[Step {step}] {name}({args}) -> {obs[:80]}...")

                if err_streak >= 3:
                    print("[Loop] 连续3次工具失败，终止")
                    return self._force_answer(msgs)

        print(f"[Loop] 达到 max_steps={self.max_steps}")
        return self._force_answer(msgs)

    def _force_answer(self, msgs: list[dict]) -> str:
        resp = self.llm.invoke(msgs + [
            {"role": "user",
             "content": "基于以上已有信息，直接给出最终答案，不要调用工具。"}
        ])
        return resp.content