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

Python 3.11.6
langchain-core==0.2.38
langchain-openai==0.1.25
openai==1.45.0
pydantic==2.9.2

用户输入 → 构造prompt(系统指令+工具描述+历史记忆) → LLM输出
  → 解析输出：是Final Answer还是Action?
     ├─ Final Answer → 返回
     └─ Action → 执行工具 → 结果写入记忆 → 回到构造prompt

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
import random

class WeatherInput(BaseModel):
    city: str = Field(description="中国城市名，如'北京'")

class WeatherTool(BaseTool):
    name: str = "get_weather"
    description: str = "根据城市名查询当前摄氏度温度，仅支持中国城市"
    args_schema: type[BaseModel] = WeatherInput

    def _run(self, city: str) -> str:
        # 模拟外部API，10%概率抛异常用于测试错误处理
        if random.random()  str:
        if not all(c in "0123456789+-*/(). " for c in expression):
            raise ValueError("表达式含非法字符")
        return str(eval(expression))  # 演示用，生产环境务必用ast解析

import re, json, time
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FTimeout

class Scratchpad:
    def __init__(self, max_turns: int = 10):
        self.max_turns = max_turns
        self.items: list[dict] = []

    def add(self, role: str, content: str):
        self.items.append({"role": role, "content": content})
        self._trim()

    def _trim(self):
        # 保留首条用户输入 + 最近 max_turns-1 条
        if len(self.items) > self.max_turns:
            self.items = [self.items[0]] + self.items[-(self.max_turns - 1):]

    def render(self) -> str:
        return "\n".join(f"{i['role']}: {i['content']}" for i in self.items)

REACT_PROMPT = """你是一个Agent。可用工具：
{tools}

严格按以下格式输出，每次只做一步：
Thought: 你的思考
Action: 工具名，必须是[{tool_names}]之一
Action Input: JSON对象，如{{"city": "北京"}}
或当你有最终答案时：
Thought: 我已得到答案
Final Answer: 最终回答

历史：
{scratchpad}
用户问题：{question}
"""

def parse_action(text: str):
    if "Final Answer:" in text:
        return "final", text.split("Final Answer:")[-1].strip()
    m = re.search(r"Action:\s*(\w+)\s*\nAction Input:\s*(\{.*?\})", text, re.S)
    if not m:
        return "invalid", None
    try:
        return "action", (m.group(1), json.loads(m.group(2)))
    except json.JSONDecodeError:
        return "invalid", None

def run_agent(question: str, tools: dict, llm, max_steps: int = 8):
    pad = Scratchpad(max_turns=10)
    pad.add("user", question)
    action_history = []
    tool_desc = "\n".join(f"- {t.name}: {t.description}" for t in tools.values())

    for step in range(max_steps):
        prompt = REACT_PROMPT.format(
            tools=tool_desc, tool_names=", ".join(tools),
            scratchpad=pad.render(), question=question)
        raw = llm.invoke(prompt).content
        kind, payload = parse_action(raw)

        if kind == "final":
            return payload, step + 1
        if kind == "invalid":
            pad.add("system", "格式错误，请严格按Thought/Action/Action Input输出")
            continue

        tool_name, args = payload
        sig = (tool_name, json.dumps(args, sort_keys=True))
        if action_history.count(sig) >= 2:
            return f"[中止] 检测到重复动作 {sig}，疑似死循环", step + 1
        action_history.append(sig)

        tool = tools.get(tool_name)
        if tool is None:
            pad.add("observation", f"错误：工具{tool_name}不存在")
            continue

        try:
            with ThreadPoolExecutor(max_workers=1) as ex:
                obs = ex.submit(tool.run, args).result(timeout=10)
        except FTimeout:
            obs = f"错误：工具{tool_name}执行超时(10s)"
        except Exception as e:
            obs = f"错误：{type(e).__name__}: {str(e)[:120]}。请换一种方式或直接给出答案。"
        pad.add("observation", obs)

    return "[中止] 达到最大轮数", max_steps