LangChain与AutoGPT双引擎：手写AI Agent的记忆与循环控制机制

python 3.10.11
langchain 0.1.0
openai 1.6.1
tiktoken 0.5.1

while not done and step  类型

def tool(name: str, description: str, args_schema: Dict[str, str]):
    def decorator(func):
        return Tool(name, description, func, args_schema)
    return decorator

# 示例工具
@tool("search_web", "搜索网页，返回前k条结果", {"query": "string", "k": "int"})
def search_web(query: str, k: int = 3):
    # 模拟搜索API
    results = [f"{query} 结果{i}" for i in range(k)]
    return "\n".join(results)

@tool("extract_content", "从URL提取正文文本", {"url": "string"})
def extract_content(url: str):
    # 模拟HTML解析，实际可用BeautifulSoup
    return f"{url} 的正文内容，长度约500字"

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
import json

class MemoryManager:
    def __init__(self, max_short_term=6):
        self.short_term = []  # 最近对话（原始文本）
        self.max_short_term = max_short_term
        self.long_term = Chroma(collection_name="agent_mem",
                                embedding_function=OpenAIEmbeddings(model="text-embedding-ada-002"))

    def add(self, role: str, content: str):
        self.short_term.append({"role": role, "content": content})
        # 保持短期记忆长度
        if len(self.short_term) > self.max_short_term:
            # 将最早的两条记忆压缩进长期记忆
            old = self.short_term.pop(0)
            self.long_term.add_texts([old["content"]], metadatas=[{"role": old["role"]}])
            self.short_term.pop(0)  # 同时丢弃对应的assistant回复

    def get_context(self, query: str, k: int = 3) -> str:
        # 合并短期与长期检索结果
        long_results = self.long_term.similarity_search(query, k=k)
        long_text = "\n".join([doc.page_content for doc in long_results])
        short_text = "\n".join([f"{m['role']}: {m['content']}" for m in self.short_term])
        return f"长期记忆:\n{long_text}\n\n短期记忆:\n{short_text}"

class Agent:
    def __init__(self, tools: list, llm, memory: MemoryManager, max_steps=10):
        self.tools = {t.name: t for t in tools}
        self.llm = llm
        self.memory = memory
        self.max_steps = max_steps
        self.error_count = 0

    def run(self, goal: str) -> str:
        step = 0
        done = False
        while step = 3:
                    return f"失败: 连续3次解析错误，最后错误: {e}"
                self.memory.add("assistant", f"解析失败: {e}，请重新输出JSON")
                continue

            # 执行工具（带重试）
            tool_func = self.tools[tool_name].func
            try:
                for attempt in range(3):  # 重试2次
                    try:
                        result = tool_func(**args)
                        break
                    except Exception as e:
                        if attempt == 2:
                            raise
                        time.sleep(2 ** attempt)  # 指数退避
                self.error_count = 0  # 成功则重置错误计数
            except Exception as e:
                self.error_count += 1
                self.memory.add("assistant", f"工具执行失败: {e}")
                continue

            # 记录观察
            self.memory.add("assistant", f"执行{tool_name}，结果: {result}")

            # 判断是否完成
            check_prompt = f"目标: {goal}\n最新结果: {result}\n是否已达成目标? 回答yes或no"
            done = self.llm.invoke(check_prompt).content.strip().lower() == "yes"

        if not done:
            return "已达最大步骤数，任务未完成"
        return result