LangChain与AutoGPT双核驱动：手写Agent记忆与循环控制机制

pip install langchain==0.1.0 langchain-openai==0.0.2.post1 openai==1.6.1 faiss-cpu==1.7.4

from langchain.tools import BaseTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词，必须包含产品名+竞品名，如'小米SU7 对比 比亚迪汉'")
    max_results: int = Field(default=5, description="返回结果数量，默认5，最大10")

class SearchTool(BaseTool):
    """重写后的搜索工具：描述里塞满了决策信息"""
    name = "web_search"
    description = """当用户需要实时信息、最新价格、竞品动态时使用。
    输入必须是JSON格式: {"query": "关键词", "max_results": 5}
    输出返回标题+URL+摘要的列表。
    注意：如果搜索无结果，返回空列表，不要重复调用超过3次。
    失败场景：网络超时时，间隔10秒重试一次。
    """
    args_schema = SearchInput

    def _run(self, query: str, max_results: int = 5) -> list:
        # 实际搜索逻辑
        return search_api(query, max_results)

from langchain.memory import ConversationBufferWindowMemory
from langchain.agents import AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI

memory = ConversationBufferWindowMemory(
    k=6,  # 保留最近6轮对话
    memory_key="chat_history",
    return_messages=True,
    human_prefix="用户",  # 中文前缀，避免英文干扰
    ai_prefix="助手"
)

import sqlite3, faiss, numpy as np
from langchain_openai import OpenAIEmbeddings

class LongTermMemory:
    def __init__(self, db_path="memory.db"):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")
        self.index = faiss.IndexFlatL2(1536)  # ada-002的维度
        self._init_db()

    def save(self, task: str, solution: str, success: bool):
        vec = self.embeddings.embed_query(task)
        self.index.add(np.array([vec], dtype=np.float32))
        self.conn.execute(
            "INSERT INTO memory (task, solution, success) VALUES (?,?,?)",
            (task, solution, int(success))
        )
        self.conn.commit()

    def recall(self, task: str, threshold: float = 0.82):
        vec = self.embeddings.embed_query(task)
        distances, indices = self.index.search(np.array([vec], dtype=np.float32), k=3)
        # 返回相似度>threshold的历史方案
        return [(self.conn.execute("SELECT solution FROM memory WHERE rowid=?", (i,)).fetchone(), d) 
                for i, d in zip(indices[0], distances[0]) if d = self.max_retries:
                return False  # 触发降级
        else:
            self.fail_count = 0  # 成功一次就重置计数
        # 第三层：全局超时
        if time.time() - self.start_time > self.global_timeout:
            return False
        return True

    def run(self, *args, **kwargs):
        self.start_time = time.time()
        return super().run(*args, **kwargs)