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

用户输入 → 组装消息(系统提示+记忆+历史) → LLM推理
  → 有tool_calls? 
      是 → 执行工具(带异常捕获) → 结果作为ToolMessage回灌 → 轮次+1 → 回到LLM
      否 → 输出最终答案，写入记忆
  → 超过max_iterations 或 检测到重复调用 → 强制收敛

from langchain_core.tools import tool
from pydantic import BaseModel, Field
import ast, operator, os

class CalcInput(BaseModel):
    expression: str = Field(description="要计算的数学表达式，仅支持+ - * / 和括号，例如 '(3+5)*2'")

@tool("calculator", args_schema=CalcInput)
def calculator(expression: str) -> str:
    """计算数学表达式，返回字符串结果。"""
    allowed = {ast.Add: operator.add, ast.Sub: operator.sub,
               ast.Mult: operator.mul, ast.Div: operator.truediv}
    def _eval(node):
        if isinstance(node, ast.Num):
            return node.n
        if isinstance(node, ast.BinOp) and type(node.op) in allowed:
            return allowed[type(node.op)](_eval(node.left), _eval(node.right))
        raise ValueError(f"不支持的表达式节点: {ast.dump(node)}")
    tree = ast.parse(expression, mode="eval")
    return str(_eval(tree.body))

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

@tool("get_weather", args_schema=WeatherInput)
def get_weather(city: str) -> str:
    """查询指定城市天气（示例为mock数据）。"""
    mock = {"北京": "晴 26℃", "上海": "多云 24℃", "深圳": "雷阵雨 29℃"}
    if city not in mock:
        raise ValueError(f"暂不支持的城市: {city}")
    return f"{city}：{mock[city]}"

class FileInput(BaseModel):
    path: str = Field(description="文件相对路径")
    content: str = Field(default="", description="写入内容；为空表示读取")

@tool("file_io", args_schema=FileInput)
def file_io(path: str, content: str = "") -> str:
    """读写工作目录下的文本文件。content为空时读取。"""
    safe_root = os.path.abspath("./agent_workspace")
    full = os.path.abspath(os.path.join(safe_root, path))
    if not full.startswith(safe_root):
        raise PermissionError("路径越界，拒绝访问")
    if content:
        os.makedirs(os.path.dirname(full), exist_ok=True)
        with open(full, "w", encoding="utf-8") as f:
            f.write(content)
        return f"已写入 {len(content)} 字符到 {path}"
    if not os.path.exists(full):
        raise FileNotFoundError(f"文件不存在: {path}")
    with open(full, encoding="utf-8") as f:
        return f.read()[:2000]

TOOLS = [calculator, get_weather, file_io]

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
import tiktoken

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

def count_tokens(messages) -> int:
    total = 0
    for m in messages:
        total += len(enc.encode(getattr(m, "content", "") or ""))
        total += 4  # 消息结构开销
    return total

class MemoryManager:
    def __init__(self, llm, max_tokens=2000, keep_recent=6):
        self.llm = llm
        self.max_tokens = max_tokens
        self.keep_recent = keep_recent
        self.summary = ""
        self.history = []  # 只存 Human/AI 文本消息，ToolMessage 不进长期记忆

    def add(self, msg):
        self.history.append(msg)
        if count_tokens(self.history) > self.max_tokens:
            self._compress()

    def _compress(self):
        old, recent = self.history[:-self.keep_recent], self.history[-self.keep_recent:]
        text = "\n".join(f"{m.type}: {m.content}" for m in old)
        prompt = f"已有摘要：{self.summary}\n新对话：\n{text}\n请合并为不超过150字的中文摘要，保留关键实体和数字。"
        self.summary = self.llm.invoke(prompt).content
        self.history = recent

    def build_context(self) -> str:
        parts = []
        if self.summary:
            parts.append(f"【历史摘要】{self.summary}")
        parts.extend(f"{m.type}: {m.content}" for m in self.history)
        return "\n".join(parts)

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage
import json, hashlib

SYSTEM_PROMPT = """你是一个工具型助手。规则：
1. 能用工具就用工具，不要编造数据；
2. 每次只调用必要的工具，拿到结果后立即给出最终答案；
3. 工具报错时，根据错误信息调整参数或换方案，不要重复同样的调用。"""

class SimpleAgent:
    def __init__(self, tools, max_iterations=8, model="gpt-4o-mini", temperature=0):
        self.llm = ChatOpenAI(model=model, temperature=temperature).bind_tools(tools)
        self.tool_map = {t.name: t for t in tools}
        self.memory = MemoryManager(ChatOpenAI(model=model, temperature=0))
        self.max_iterations = max_iterations

    def run(self, user_input: str) -> str:
        self.memory.add(HumanMessage(content=user_input))
        messages = [SystemMessage(content=SYSTEM_PROMPT),
                    HumanMessage(content=self.memory.build_context())]
        call_signatures = []

        for step in range(1, self.max_iterations + 1):
            try:
                ai_msg = self.llm.invoke(messages)
            except Exception as e:
                return f"[LLM调用失败] {type(e).__name__}: {e}"

            if not ai_msg.tool_calls:
                self.memory.add(AIMessage(content=ai_msg.content))
                return ai_msg.content or "（模型未返回内容）"

            messages.append(ai_msg)
            for tc in ai_msg.tool_calls:
                sig = hashlib.md5(
                    f"{tc['name']}:{json.dumps(tc['args'], sort_keys=True, ensure_ascii=False)}".encode()
                ).hexdigest()
                call_signatures.append(sig)
                if call_signatures.count(sig) >= 3:
                    return f"[提前终止] 检测到重复调用 {tc['name']}，参数: {tc['args']}"

                observation = self._execute_tool(tc)
                messages.append(ToolMessage(content=observation, tool_call_id=tc["id"]))

        return f"[达到最大轮次 {self.max_iterations}] 任务未收敛，最后模型输出: {ai_msg.content}"

    def _execute_tool(self, tool_call) -> str:
        name, args = tool_call["name"], tool_call["args"]
        tool = self.tool_map.get(name)
        if tool is None:
            return f"[错误] 未知工具: {name}，可用工具: {list(self.tool_map)}"
        try:
            result = tool.invoke(args)
            return str(result)[:1500]  # 截断，防止上下文爆炸
        except Exception as e:
            return f"[工具异常] {type(e).__name__}: {e}。请检查参数或换一种方式。"

agent = SimpleAgent(TOOLS, max_iterations=8)
print(agent.run("帮我查一下北京和上海天气，然后算一下温差（北京26，上海24），最后把结果写到 report.txt"))