LangChain+AutoGPT架构手写Agent：工具调用与记忆循环实现

pip install langchain==0.1.0 openai==1.12.0

用户问题 → 系统Prompt + 工具描述 + 历史记忆
        → GPT-4 输出 JSON（action 或 final_answer）
        → 如果是 action，查工具表并执行
        → 捕获异常，将错误信息追加到记忆
        → 回到循环，直到 final_answer 或达到 max_iterations

from langchain.tools import BaseTool
from langchain.utilities import DuckDuckGoSearchAPIWrapper
from pydantic import BaseModel, Field
import math

# 计算器工具的参数定义
class CalculatorInput(BaseModel):
    expression: str = Field(description="数学表达式，如 '2**10' 或 'sqrt(16)'")

class CalculatorTool(BaseTool):
    name = "calculator"
    description = "用于执行数学计算，输入一个数学表达式字符串"
    args_schema = CalculatorInput

    def _run(self, expression: str) -> str:
        # 安全起见，只允许白名单函数
        allowed_names = {"abs", "round", "min", "max", "sqrt": math.sqrt}
        try:
            # 使用eval但限制命名空间
            result = eval(expression, {"__builtins__": {}}, allowed_names)
            return f"计算结果: {result}"
        except Exception as e:
            return f"计算错误: {str(e)}"

# 搜索工具
class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")

class SearchTool(BaseTool):
    name = "web_search"
    description = "搜索互联网获取最新信息"
    args_schema = SearchInput

    def _run(self, query: str) -> str:
        wrapper = DuckDuckGoSearchAPIWrapper(max_results=3)
        return wrapper.run(query)

# 注册到工具表
tools = [CalculatorTool(), SearchTool()]
tool_map = {tool.name: tool for tool in tools}

import json
from collections import deque
from langchain.chat_models import ChatOpenAI

class SimpleAgent:
    def __init__(self, tools, model_name="gpt-4-0125-preview", max_iter=10, memory_size=5):
        self.llm = ChatOpenAI(model_name=model_name, temperature=0.3, max_tokens=500)
        self.tool_map = {tool.name: tool for tool in tools}
        self.max_iter = max_iter
        self.memory = deque(maxlen=memory_size * 2)  # 存用户和AI消息

    def build_prompt(self, user_query):
        tool_desc = "\n".join([f"- {name}: {tool.description}" for name, tool in self.tool_map.items()])
        system = f"""你是一个AI助手，可以调用以下工具：
{tool_desc}

输出格式必须是JSON：
{{"action": "tool_name", "params": {{"arg": "value"}}}}
或
{{"action": "final_answer", "params": {{"result": "你的最终回答"}}}}

注意：一次只能调用一个工具，收到工具结果后再决定下一步。"""
        return system

    def run(self, user_query):
        self.memory.append({"role": "user", "content": user_query})
        for step in range(self.max_iter):
            # 构建消息列表
            messages = [{"role": "system", "content": self.build_prompt(user_query)}]
            messages.extend(self.memory)

            # 调用模型
            response = self.llm.invoke(messages)
            assistant_msg = response.content
            self.memory.append({"role": "assistant", "content": assistant_msg})

            # 解析JSON
            try:
                parsed = json.loads(assistant_msg)
            except json.JSONDecodeError:
                # 模型可能输出了带前后缀的JSON，尝试提取
                start = assistant_msg.find('{')
                end = assistant_msg.rfind('}') + 1
                parsed = json.loads(assistant_msg[start:end])

            # 执行动作
            if parsed["action"] == "final_answer":
                return parsed["params"]["result"]

            tool = self.tool_map.get(parsed["action"])
            if not tool:
                error_msg = f"错误：未知工具 {parsed['action']}"
                self.memory.append({"role": "user", "content": error_msg})
                continue

            # 执行工具并捕获异常
            try:
                result = tool.run(parsed["params"])
                observation = f"工具 {parsed['action']} 返回: {result}"
            except Exception as e:
                observation = f"工具执行异常: {str(e)}"

            self.memory.append({"role": "user", "content": observation})

        return "达到最大迭代次数，任务失败。"