LangChain与AutoGPT双轮驱动：手写Agent循环控制与记忆管理全拆解

Python 3.10.12
langchain==0.1.0
langchain-openai==0.0.2
openai==1.10.0

用户输入 → 记忆检索 → LLM推理（决定行动） → 执行工具 → 结果回填 → 循环判断（是否需继续）

from pydantic import BaseModel, Field
from langchain.tools import BaseTool
from typing import Type, Optional

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

class CalculatorTool(BaseTool):
    name = "calculator"
    description = "计算数学表达式，支持加减乘除和括号"
    args_schema: Type[BaseModel] = CalculatorInput

    def _run(self, expression: str) -> str:
        try:
            # 安全计算：限制在四则运算
            result = eval(expression, {"__builtins__": {}}, {})
            return f"计算结果: {result}"
        except Exception as e:
            return f"计算失败: {str(e)}"

    async def _arun(self, expression: str) -> str:
        return self._run(expression)

class SearchTool(BaseTool):
    name = "web_search"
    description = "搜索互联网获取最新信息，输入为搜索关键词"

    def _run(self, query: str) -> str:
        # 模拟搜索，实际项目中替换为真实API
        return f"关于'{query}'的搜索结果：根据公开资料，..." 

from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage, AIMessage
import json

class SimpleAgent:
    def __init__(self, tools, memory_window=5, max_iters=8):
        self.tools = {t.name: t for t in tools}
        self.memory = []  # 短期记忆：消息列表
        self.memory_window = memory_window  # 滑动窗口大小
        self.max_iters = max_iters  # 最大迭代次数
        self.llm = ChatOpenAI(model="gpt-4-1106-preview", temperature=0.2)
        self.system_prompt = self._build_system_prompt()

    def _build_system_prompt(self):
        tool_desc = "\n".join(
            f"{name}: {tool.description}，参数格式: {tool.args_schema.schema()['properties']}"
            for name, tool in self.tools.items()
        )
        return f"""你是智能助手，可以调用以下工具：
{tool_desc}

严格按以下流程工作：
1. 如果需要调用工具，输出JSON格式：{{"action": "工具名", "action_input": {{"参数名": "参数值"}}}}
2. 如果不需要工具或已获得最终答案，输出：{{"action": "Final Answer", "action_input": "你的回答"}}
3. 工具调用失败时，根据错误信息调整参数重试，最多重试2次。"""

    def _trim_memory(self):
        """滑动窗口裁剪：保留最近N轮对话"""
        if len(self.memory) > self.memory_window * 2:  # 每轮含user+assistant两条
            self.memory = self.memory[-(self.memory_window * 2):]

    def run(self, user_input: str) -> str:
        self.memory.append(HumanMessage(content=user_input))
        iterations = 0

        while iterations < self.max_iters:
            iterations += 1
            # 构建消息序列
            messages = [SystemMessage(content=self.system_prompt)] + self.memory

            # 调用LLM
            response = self.llm.invoke(messages)
            content = response.content.strip()
            self.memory.append(AIMessage(content=content))
            self._trim_memory()

            # 解析JSON动作
            try:
                action_data = json.loads(content)
                action = action_data.get("action", "")
                action_input = action_data.get("action_input", {})
            except json.JSONDecodeError:
                # 容错：LLM可能输出纯文本，尝试提取JSON
                import re
                json_match = re.search(r'\{.*\}', content, re.DOTALL)
                if json_match:
                    action_data = json.loads(json_match.group())
                    action = action_data["action"]
                    action_input = action_data["action_input"]
                else:
                    return "无法解析LLM输出，请换一种说法重试"

            # 执行工具或返回最终答案
            if action == "Final Answer":
                return action_input
            elif action in self.tools:
                tool = self.tools[action]
                try:
                    # 工具执行，失败时回填错误信息
                    result = tool.run(action_input)
                    self.memory.append(HumanMessage(content=f"工具{action}执行结果: {result}"))
                except Exception as e:
                    error_msg = f"工具{action}执行失败: {str(e)}，请检查参数或换一种方式"
                    self.memory.append(HumanMessage(content=error_msg))
            else:
                # 未知动作，让LLM重新推理
                self.memory.append(HumanMessage(content=f"未知动作'{action}'，可用工具: {list(self.tools.keys())}"))

        return "达到最大迭代次数，任务未完成"

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import MemoryVectorStore

class LongTermMemory:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.store = MemoryVectorStore.from_documents([], self.embeddings)

    def save(self, task, result):
        self.store.add_texts([f"任务: {task}\n结果: {result}"])

    def recall(self, query, k=3):
        docs = self.store.similarity_search(query, k=k)
        return "\n".join(d.page_content for d in docs)

relevant_history = self.long_memory.recall(user_input)
if relevant_history:
    self.memory.append(HumanMessage(content=f"历史相关记录:\n{relevant_history}"))