LangChain与AutoGPT思想融合：手写可观测AI Agent核心循环

Python 3.11.8
langchain==0.2.7
langchain-openai==0.1.8
openai==1.30.2
pydantic==2.7.1
python-dotenv==1.0.1

max_iterations = 10  # 防止无限循环
retry_limits = 3     # 单步JSON解析失败重试上限
context_window = 6   # 滑动窗口消息条数

from langchain_core.tools import tool
from pydantic import BaseModel, Field

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

@tool("calculator", args_schema=CalcInput)
def calculator(expression: str) -> str:
    """执行简单的四则运算，注意：不支持括号嵌套"""
    try:
        # 安全评估：仅允许数字和运算符
        allowed = set("0123456789+-*/.")
        if not set(expression).issubset(allowed):
            return "错误：包含非法字符"
        result = eval(expression)  # 注意：生产环境请用ast模块
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算失败: {str(e)}"

class QueryInput(BaseModel):
    table: str = Field(description="表名，仅支持 users 或 orders")
    key: str = Field(description="主键值")

@tool("db_query", args_schema=QueryInput)
def db_query(table: str, key: str) -> str:
    """模拟数据库查询，返回固定测试数据"""
    mock_data = {
        "users": {"1": "张三, 年龄30", "2": "李四, 年龄25"},
        "orders": {"1": "订单A, 金额100元", "2": "订单B, 金额200元"}
    }
    result = mock_data.get(table, {}).get(key, "未找到记录")
    return f"查询结果: {result}"

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
import json, re
from collections import deque

class SimpleAgent:
    def __init__(self, tools: dict, model="gpt-4o-mini", max_iter=10):
        self.tools = tools  # {"calculator": calculator, "db_query": db_query}
        self.llm = ChatOpenAI(
            model=model,
            temperature=0.2,
            max_tokens=500,
            response_format={"type": "json_object"}
        )
        self.max_iter = max_iter
        self.system_prompt = """你是一个智能助手。你必须按以下步骤操作：
        1. 思考：用中文解释你打算怎么做
        2. 行动：调用工具，格式为 {"tool": "工具名", "args": {"参数名": "值"}}
        3. 观察：等待工具返回结果
        4. 重复直到得到最终答案，最后输出 {"final_answer": "你的结论"}
        严格要求：每次输出必须是一个JSON对象，不要包含其他文字。"""

    def run(self, user_input: str) -> str:
        messages = deque(maxlen=6)  # 滑动窗口记忆
        messages.append(SystemMessage(content=self.system_prompt))
        messages.append(HumanMessage(content=user_input))

        for step in range(self.max_iter):
            print(f"\n>>> 步骤 {step+1}/{self.max_iter}")
            # 1. 调用LLM
            response = self.llm.invoke(list(messages))
            assistant_msg = response.content
            print(f"LLM输出: {assistant_msg}")

            # 2. 解析JSON（带重试）
            parsed = self._parse_json_with_retry(assistant_msg, step)
            if parsed is None:
                # 重试失败，直接返回错误信息
                return "抱歉，我无法理解你的请求，请重新表述。"

            # 3. 判断是否结束
            if "final_answer" in parsed:
                return parsed["final_answer"]

            # 4. 执行工具调用
            tool_name = parsed.get("tool")
            args = parsed.get("args", {})
            if tool_name not in self.tools:
                error_msg = f"错误：工具 {tool_name} 不存在"
                messages.append(HumanMessage(content=error_msg))
                continue

            try:
                tool_result = self.tools[tool_name].invoke(args)
                print(f"工具 {tool_name} 返回: {tool_result}")
                # 将观察结果加入上下文
                messages.append(HumanMessage(content=f"观察结果: {tool_result}"))
            except Exception as e:
                error_msg = f"工具执行异常: {str(e)}"
                messages.append(HumanMessage(content=error_msg))

        return "达到最大迭代次数，无法完成任务。"

    def _parse_json_with_retry(self, text: str, step: int) -> dict | None:
        for attempt in range(3):
            try:
                # 移除可能的Markdown代码块标记
                json_str = re.sub(r"```json|```", "", text).strip()
                return json.loads(json_str)
            except json.JSONDecodeError:
                if attempt < 2:
                    print(f"JSON解析失败，重试{attempt+1}次...")
                    # 重试时要求LLM只输出JSON
                    retry_prompt = "你的上次输出不是合法JSON，请重新输出，只包含JSON对象。"
                    response = self.llm.invoke([
                        HumanMessage(content=retry_prompt),
                        HumanMessage(content=text)
                    ])
                    text = response.content
                else:
                    print("JSON解析连续失败3次，放弃")
                    return None
        return None

# 实例化并运行
agent = SimpleAgent(tools={"calculator": calculator, "db_query": db_query})
print(agent.run("查询用户ID为1的信息，然后计算他的年龄乘以2的结果"))