LangChain手写Agent循环控制与记忆管理，AutoGPT架构降本60%

# requirements.txt
langchain==0.2.3
langchain-openai==0.1.8
openai==1.30.1
python-dotenv==1.0.1
tiktoken==0.6.0

┌─────────────┐
│ LLM (GPT-4o)│←── 每次迭代只发送“工具描述+当前任务+最近3步”
└─────────────┘
       ↓ 输出JSON格式action
┌─────────────┐
│ ActionParser│→ 异常则触发错误恢复机制
└─────────────┘
       ↓
┌─────────────┐     ┌──────────────┐
│ ToolRegistry│ →  │  ToolExecutor │ 含超时熔断
└─────────────┘     └──────────────┘
       ↓ 返回observation
┌─────────────┐
│ MemoryStore │ → 滑动窗口存储(step, action, observation)
└─────────────┘

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Optional

class SqlQueryInput(BaseModel):
    query: str = Field(description="SQL查询语句，必须是SELECT开头")
    db_name: str = Field(default="analytics", description="目标数据库名")

class SqlQueryTool(BaseTool):
    name = "sql_query"
    description = "执行SELECT查询并返回结果。输入必须包含query和db_name。"
    args_schema = SqlQueryInput

    def _run(self, query: str, db_name: str = "analytics") -> str:
        # 此处省略数据库连接代码
        if not query.strip().upper().startswith("SELECT"):
            return "Error: 只允许SELECT语句"
        result = execute_sql(db_name, query)
        return str(result[:500])  # 限制返回长度防止上下文爆炸

from collections import deque
import tiktoken

class SlidingWindowMemory:
    def __init__(self, max_steps=3, max_tokens=1500):
        self.steps = deque(maxlen=max_steps)
        self.encoder = tiktoken.encoding_for_model("gpt-4o")
        self.max_tokens = max_tokens

    def add(self, step: dict):
        self.steps.append(step)
        self._truncate()

    def _truncate(self):
        # 计算总token数，超过阈值则丢弃最老步骤
        total = 0
        for s in self.steps:
            total += len(self.encoder.encode(json.dumps(s["action"]))) + \
                     len(self.encoder.encode(s["observation"]))
        while total > self.max_tokens and len(self.steps) > 1:
            removed = self.steps.popleft()
            total -= len(self.encoder.encode(json.dumps(removed)))

import json
from typing import Optional

class AgentLoop:
    def __init__(self, llm, tools, memory, max_iterations=8):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.memory = memory
        self.max_iterations = max_iterations
        self.repeat_threshold = 3  # 连续相同动作阈值

    def run(self, task: str) -> str:
        history = []
        last_action = None
        repeat_count = 0

        for step in range(self.max_iterations):
            # 构建提示词
            prompt = self._build_prompt(task, history)
            response = self.llm.invoke(prompt)
            parsed = self._parse_response(response.content)

            if parsed is None:
                # 错误恢复机制：重试3次，每次提示更清晰
                for retry in range(3):
                    repair_prompt = f"上次输出格式错误，请输出JSON：{response.content}"
                    response = self.llm.invoke(repair_prompt)
                    parsed = self._parse_response(response.content)
                    if parsed: break
                if parsed is None:
                    return "Agent失败：无法解析LLM输出"

            if parsed["action"] == "final_answer":
                return parsed["answer"]

            # 重复动作检测
            if parsed["action"] == last_action and \
               parsed["action_input"] == last_params:
                repeat_count += 1
                if repeat_count >= self.repeat_threshold:
                    # 强制切换策略：注入新指令
                    history.append({
                        "action": "system_hint",
                        "observation": "重复操作，请尝试不同方法或给出最终答案"
                    })
                    repeat_count = 0
            else:
                repeat_count = 0

            # 执行工具
            tool = self.tools.get(parsed["action"])
            if not tool:
                # 动态注册工具失败处理
                return f"工具 {parsed['action']} 不存在，请检查工具列表"

            try:
                observation = tool.invoke(parsed["action_input"])
            except Exception as e:
                observation = f"Tool execution error: {str(e)[:200]}"

            # 记忆存储
            self.memory.add({
                "step": step,
                "action": parsed,
                "observation": observation
            })
            history.append({
                "action": parsed,
                "observation": observation
            })
            last_action = parsed["action"]
            last_params = parsed["action_input"]

        return f"达到最大迭代次数 {self.max_iterations}，任务未完成"