LangChain与AutoGPT启发下从零手写AI Agent循环控制与记忆

python==3.10.12
langchain==0.1.0
langchain-openai==0.0.2.post1
openai==1.12.0
pydantic==2.6.1

from collections import deque
from typing import Dict, List, Any
from langchain_core.tools import BaseTool

class CircularMemory:
    """循环队列记忆，只保留最近N轮交互"""
    def __init__(self, capacity: int = 6):
        self.capacity = capacity
        self.buffer: deque = deque(maxlen=capacity)
        self.summary: str = ""

    def add(self, thought: str, action: str, observation: str) -> None:
        # 关键：把observation截断到500字符，防止数值溢出
        truncated_obs = observation[:500]
        self.buffer.append({
            "thought": thought[-300:],  # 同样截断thought
            "action": action,
            "observation": truncated_obs
        })

    def compress(self, llm) -> str:
        """每3轮调用一次，压缩历史"""
        if len(self.buffer)  str:
        return "\n".join(
            f"Thought: {item['thought']}\nAction: {item['action']}\nObs: {item['observation']}"
            for item in self.buffer
        )

class CalculatorTool(BaseTool):
    name: str = "calculator"
    description: str = "计算数学表达式，输入必须是四则运算表达式，如'(2+3)*4'"

    def _run(self, query: str) -> str:
        # 安全求值：只允许数字、运算符和括号
        allowed_chars = set("0123456789+-*/(). ")
        if not all(c in allowed_chars for c in query):
            return "Error: 非法字符"
        try:
            # 使用eval但限制命名空间，这是唯一敢用eval的场景
            result = eval(query, {"__builtins__": None}, {})
            return str(result)
        except ZeroDivisionError:
            return "Error: 除零错误"
        except Exception as e:
            return f"Error: {str(e)[:100]}"

class AgentLoop:
    def __init__(self, llm, tools: Dict[str, BaseTool], max_iterations: int = 5):
        self.llm = llm
        self.tools = tools
        self.max_iterations = max_iterations
        self.memory = CircularMemory(capacity=6)

    def run(self, task: str) -> str:
        prompt = f"任务: {task}\n可用工具: {list(self.tools.keys())}\n"
        for i in range(self.max_iterations):
            # 每3轮压缩一次记忆
            if i > 0 and i % 3 == 0:
                memory_str = self.memory.compress(self.llm)
            else:
                memory_str = self.memory._format()

            full_prompt = prompt + memory_str + "\n请输出JSON格式(thought, action, action_input):"
            response = self.llm.invoke(full_prompt).content

            # 解析JSON，这是另一个易碎点
            try:
                parsed = json.loads(response.replace("```json", "").replace("```", "").strip())
                thought = parsed["thought"]
                action = parsed["action"]
                action_input = parsed["action_input"]
            except (json.JSONDecodeError, KeyError) as e:
                # 模型输出不合规时，强制回退到"最终答案"
                thought = "模型输出格式错误"
                action = "final_answer"
                action_input = f"解析失败: {e}"

            if action == "final_answer":
                return action_input

            # 执行工具，错误不杀死循环
            if action not in self.tools:
                observation = f"Error: 未知工具 {action}"
            else:
                try:
                    observation = self.tools[action].run(action_input)
                except Exception as e:
                    observation = f"Tool execution failed: {str(e)[:200]}"

            self.memory.add(thought, action, observation)
            print(f"[轮次{i+1}] thought={thought[:50]} action={action} obs={observation[:50]}")

        return "达到最大迭代轮次，未找到答案"

import re
json_match = re.search(r'\{.*\}', response, re.DOTALL)
if json_match:
    parsed = json.loads(json_match.group())