AI Agent 的记忆瓶颈：上下文漂移根因与自驱动工具链解法

原文首次发表于 tamiz.pro。

大语言模型（LLM）已经彻底改变了软件开发，但当我们将它们堆叠成多步骤的智能体时，一个根本性的架构缺陷浮出水面：上下文窗口。与维护着不变的需求和状态记忆的人类工程师不同，AI 智能体常常遭受"上下文漂移"——随着对话历史的增长，它们会丢失对指令的追踪或产生事实性幻觉。

这就是记忆瓶颈。它不仅仅是令牌限制的问题；而是智能体如何在时间维度上管理状态的系统性失败。在本次深度解析中，我们将剖析为什么标准的 ReAct 循环在记忆压力下会失效，以及自主驱动工具——一种能够自主管理工具、记忆和执行而无需持续人工干预的架构——如何解决这一问题。

1. 智能体失效的解剖学

要理解解决方案，我们必须首先诊断疾病。AI 智能体通常在以下循环中运作：

观察：接收用户输入和对话历史。

思考：LLM 分析上下文以决定下一步动作。

行动：智能体调用工具（例如 search_web、execute_code）。

观察：工具的输出被附加到上下文窗口。

随着智能体执行更多步骤，上下文窗口逐渐填满。现代 LLM（如 GPT-4o 或 Claude 3.5 Sonnet）拥有巨大的窗口（128k-200k 个令牌），但更大的窗口并不等同于更好的记忆。它们等同于注意力稀释。

注意力接收器问题

LLM 是概率引擎。随着上下文增长，概率质量在无关令牌上分布得更稀疏。这导致两种常见的失效模式：

中间迷失：在长对话开始时提供的关键指令，随着新工具输出的到来而被遗忘或降权。

工具调用崩溃：模型开始重复之前的工具调用，因为它"忘记"了结果或决策逻辑，导致无限循环或冗余工作。

一个典型例子是数据分析智能体。在获取了五个数据集并执行了三次聚合操作后，模型可能会忘记第一步提供的"收入"初始定义，导致最终结论错误。智能体拥有数据，但它丢失了状态。

2. 从响应式到自主式：自主驱动转型

"自主驱动工具"这个术语借鉴了自动驾驶汽车的类比。在响应式智能体中，"驾驶员"（LLM）查看当前帧（上下文）并决定向左转（调用工具 A）还是向右转（调用工具 B）。如果帧很杂乱（记忆瓶颈），驾驶员就会崩溃。

在自主驱动架构中，系统包含自己的传感器和导航系统，这些系统独立于驾驶员的即时感知运行。这转化为：

• 外部化记忆：一种持久化存储层（向量数据库、SQL 或 JSON 文件），能够超越上下文窗口存活。
• 自主编排：一个控制平面，管理工具执行、错误恢复和状态转换，不完全依赖 LLM 的工作记忆。
• 工具即基础设施：工具不仅仅是待调用的函数；它们是具有生命周期状态的托管资源。

3. 自主驱动工具的核心支柱

我们如何构建这个？需要三个技术组件。

A. 语义记忆（RAG 用于状态）

我们不再将整个对话历史塞入上下文窗口，而是将关键状态提取到语义记忆存储中。

• 工作记忆：活动上下文（最近 N 轮对话）。
• 情景记忆：向量数据库，存储过去的交互、工具结果和决策，按嵌入向量索引。

当智能体需要回忆 50 步前做出的决策时，它不会读取历史。它查询向量数据库以获取相关嵌入向量，并仅将摘要注入回上下文。

B. 记忆管理器中间件

这是"自主驱动"的大脑。它位于 LLM 和工具之间。其职责包括：

• 摘要：自动汇总对话的较旧部分，并用简洁摘要替换。
• 相关性过滤：决定哪些过去的工具输出与当前查询仍然相关。
• 上下文打包：通过组合当前输入、工作记忆和检索到的情景记忆来构建最优提示。

C. 工具状态机

在自主驱动系统中，工具不是无状态的。例如，deploy_to_prod 工具应维护状态（例如 pending、deploying、success、failed）。智能体可以查询工具执行的状态，而无需重新运行它或在其上下文中记住完整日志。

4. 架构：自主驱动智能体循环

让我们用类 Python 伪代码来看一个自主驱动智能体的概念实现。我们将使用将控制器（编排器）与执行器（LLM）分离的模式。

class SelfDrivingAgent:
    def __init__(self, llm, memory_store, tool_registry):
        self.llm = llm
        self.memory = memory_store  # Vector DB or SQL
        self.tools = tool_registry
        self.context_buffer = []

    def run(self, user_query):
        # 1. Retrieve Relevant History
        # Instead of sending full history, we query memory for relevant past events
        relevant_context = self.memory.query(user_query, top_k=3)

        # 2. Build Prompt
        prompt = self._construct_prompt(user_query, relevant_context)

        # 3. LLM Decision
        decision = self.llm.generate(prompt)

        # 4. Tool Execution & State Tracking
        if decision.action == "call_tool":
            tool_result = self.tools.execute(decision.tool_name, decision.params)

            # 5. Memory Commitment
            # Store the outcome semantically, not just textually
            self.memory.commit(
                event_type="tool_execution",
                tool=decision.tool_name,
                result_summary=summary(tool_result),
                embedding=generate_embedding(f"{decision.tool_name}: {tool_result}")
            )

            # 6. Recursive Step or Final Answer
            return self._handle_result(decision, tool_result)

        return decision.final_answer

与标准智能体的关键区别：

• 查询，而非倾倒：我们根据相关性查询记忆，而不是倾倒整个历史。
• 结构化提交：工具结果在存储前被摘要和嵌入，保持上下文窗口精简。
• 状态感知：智能体通过记忆存储了解先前工具执行的状态。

5. 使用 LangGraph 和 Memory 的实践实现

对于生产级自主驱动智能体，像 LangGraph（由 LangChain 提供）这样的框架提供了管理有状态多智能体工作流的基础设施。LangGraph 允许你用中央状态对象定义节点（工具/LLM）和边（转换）。

以下是你如何在 LangGraph 中实现记忆增强的工具调用：

from langgraph.graph import StateGraph, END
from typing import TypedDict
import uuid

class AgentState(TypedDict):
    messages: list  # Current conversation
    memory: list    # Retrieved relevant history
    tool_results: dict  # Cached tool results

# Define a memory retrieval node
def retrieve_memory(state: AgentState) -> AgentState:
    query = state['messages'][-1].content
    # Query vector store
    relevant_docs = vector_store.similarity_search(query, k=2)
    state['memory'] = relevant_docs
    return state

# Define a tool-calling node
def call_tool(state: AgentState) -> AgentState:
    # LLM decides to call a tool
    # ... tool execution logic ...
    # Store result in tool_results for future reference
    state['tool_results'][tool_name] = result
    return state

# Build the graph with memory-aware transitions
workflow = StateGraph(AgentState)
workflow.add_node("retrieve_memory", retrieve_memory)
workflow.add_node("agent", "llm_node")
workflow.add_node("tool", call_tool)

workflow.set_entry_point("retrieve_memory")
workflow.add_edge("retrieve_memory", "agent")
workflow.add_conditional_edges(
    "agent",
    lambda x: "tool" if x.get("needs_tool") else END,
    {"tool": "tool", "END": END}
)
workflow.add_edge("tool", "agent")

app = workflow.compile()

为什么这能解决瓶颈：

• 记忆是显式的：状态中的 memory 键从外部源填充，而不仅仅是累积的历史。
• 工具结果被缓存：tool_results 字典充当短期缓存，防止 LLM 需要记住先前轮次中的原始输出。
• 控制流是确定性的：图结构确保记忆检索在 LLM 做出决策之前发生，减少了上下文漂移的机会。

6. 高级：自反思工具

最稳健的自主驱动智能体结合了自反思。工具执行后，智能体应评估：

• 结果成功了吗？
• 我是否误解了工具的用途？
• 我需要为将来参考存储这个结果吗？

这个元认知步骤可以作为图中的一个独立节点实现，批判工具的输出并相应地更新记忆存储。

def self_reflect(state: AgentState) -> AgentState:
    tool_output = state['tool_results']
    reflection_prompt = f"Evaluate the success of this tool call: {tool_output}. Summarize key findings for future memory."
    reflection = llm.generate(reflection_prompt)
    state['memory'].append({
        "type": "reflection",
        "content": reflection,
        "timestamp": datetime.now()
    })
    return state

这段反思成为语义记忆的一部分，使智能体能够从过去的工具交互中"学习"。

7. 生产环境最佳实践

分层记忆：热记忆（Redis）用于最近的交互，冷记忆（PostgreSQL + 向量嵌入）用于长期存储。

摘要策略：对较旧的消息实施激进摘要。在达到特定 token 阈值后，每第 10 条消息替换为摘要。

工具版本控制：随着工具集的演进，确保智能体对工具 schema 的记忆也随之更新。过时的工具定义会导致执行失败。

故障恢复：自驱动系统必须优雅地处理工具故障。如果某个工具失败，智能体应使用修改后的参数重试，或升级到人工干预，而不是无限循环。

记忆瓶颈是 AI 智能体在复杂多步骤任务中失败的主要原因。通过从被动式、历史倾倒模式转向自驱动工具架构——记忆外部化、工具状态被管理、编排自主化——我们可以构建不仅智能、而且可靠的智能体。

这种方法与高级工程师的工作方式如出一辙：他们不需要记住自己写过的每一行代码；他们使用文档（记忆）、标准化流程（工具）和清晰的架构模式（编排）来解决问题，无论复杂程度如何。

了解更多关于构建生产级 AI 智能体的见解，请参阅 Tamiz's Insights on AI system architecture。

常见问题

问：RAG 与智能体中语义记忆的区别是什么？ 答：RAG（检索增强生成）通常检索外部知识（文档、网页）来回答问题。自驱动智能体中的语义记忆检索内部状态（过去的工具调用、决策、结果）以在多步骤任务中保持连续性。

问：我实际上需要多少上下文？ 答：目标是最低可行上下文。对于 200k token 的模型，你可能认为不需要优化。然而注意力稀释是真实存在的。将活跃上下文保持在 10k token 以下，并将其余内容卸载到记忆中的效果通常优于向每次输入喂养完整历史。

问：我可以将此方法与任何 LLM 一起使用吗？ 答：可以。自驱动架构与框架无关。无论你使用的是 OpenAI、Anthropic，还是像 Llama 3 这样的开源模型，外部记忆和自主编排的模式同样适用。

让我们再回答几个问题，然后进入实际实现部分。

问：外部记忆会变慢吗？ 答：像 Chroma、Milvus 或 Weaviate 这样的现代向量数据库，在 10 万向量以下的查询中以个位数毫秒级返回结果。延迟惩罚与你 LLM 每轮生成所花费的秒数相比可以忽略不计。如果你遇到了缓慢问题，通常是索引问题，而不是检索问题。

问：如何防止智能体无限循环？ 答：实施三项保障措施：（1）每个任务的最大步骤预算；（2）工具调用的去重检查，这样同一操作不会重复执行；（3）反思步骤，智能体评估其最后一个操作是否向目标推进了。如果未检测到进展，编排器会触发带有更新上下文的重新规划。

问：成本如何？ 答：记忆外部化将成本从重复的上下文膨胀转移到一次性嵌入和索引。对于典型的智能体会话，你在 LLM 调用上的花费远高于记忆操作。关键优化是选择性召回——只获取与当前子目标相关的记忆，而不是每次都将整个知识库倾倒到提示中。

构建自驱动智能体

理论够了。让我们开始动手。

我们将使用 Python 构建一个最小但完整的实现，包含三层：工具注册表、记忆层和编排循环。

第一层：工具注册表

工具是智能体的双手。每项能力都必须声明性注册，以便编排器能够对其进行推理。

# tools.py
from dataclasses import dataclass
from typing import Any, Callable

@dataclass
class ToolSpec:
    name: str
    description: "str"
    parameters: dict  # JSON Schema
    fn: Callable[..., Any]

    def to_openai_format(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            },
        }

class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, ToolSpec] = {}

    def register(self, tool: ToolSpec):
        self._tools[tool.name] = tool

    def get(self, name: str) -> ToolSpec:
        if name not in self._tools:
            raise KeyError(f"Tool '{name}' not found")
        return self._tools[name]

    def list(self) -> list[ToolSpec]:
        return list(self._tools.values())

# 示例工具
def read_file(path: str) -> str:
    with open(path) as f:
        return f.read()

def write_file(path: str, content: str) -> str:
    with open(path, "w") as f:
        f.write(content)
    return f"Wrote {len(content)} chars to {path}"

def search_web(query: str, max_results: int = 5) -> list[dict]:
    # 生产环境中与搜索 API 集成
    return [{"title": query, "snippet": f"Result for {query}"}] * max_results

registry = ToolRegistry()
registry.register(ToolSpec(
    name="read_file",
    description="Read the contents of a file from disk",
    parameters={
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "Absolute or relative file path"},
        },
        "required": ["path"],
    },
    fn=read_file,
))
registry.register(ToolSpec(
    name="write_file",
    description="Write content to a file on disk",
    parameters={
        "type": "object",
        "properties": {
            "path": {"type": "string"},
            "content": {"type": "string"},
        },
        "required": ["path", "content"],
    },
    fn=write_file,
))
registry.register(ToolSpec(
    name="search_web",
    description="Search the web for information",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string"},
            "max_results": {"type": "integer", "default": 5},
        },
        "required": ["query"],
    },
    fn=search_web,
))

第二层：情景记忆

这正是我们解决记忆瓶颈的地方。每个观察、工具结果和决策都成为带有语义嵌入的结构化记忆。

# memory.py
import hashlib
import json
import numpy as np
from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Optional

@dataclass
class Memory:
    id: str
    type: str  # "observation" | "decision" | "tool_result" | "reflection"
    content: str
    context: Optional[str]
    timestamp: str
    embedding: Optional[list[float]] = None
    importance: float = 1.0

    def to_dict(self) -> dict:
        return asdict(self)

    @classmethod
    def from_dict(cls, d: dict) -> "Memory":
        d = d.copy()
        return cls(**d)

class VectorMemoryStore:
    """使用余弦相似度的简单内存向量存储。"""

    def __init__(self, embed_fn=None):
        self.memories: list[Memory] = []
        self.embed_fn = embed_fn or self._noop_embed

    def _noop_embed(self, text: str) -> list[float]:
        """确定性占位嵌入。用真实模型替换。"""
        h = int(hashlib.md5(text.encode()).hexdigest(), 16)
        return [(h >> (i * 8)) & 0xFF for i in range(16)]

    def add(self, memory: Memory):
        if self.embed_fn and not memory.embedding:
            memory.embedding = self.embed_fn(memory.content)
        self.memories.append(memory)

    def recall(self, query: str, k: int = 5) -> list[Memory]:
        query_emb = self.embed_fn(query)
        scored = []
        for m in self.memories:
            if not m.embedding:
                continue
            sim = self._cosine(query_emb, m.embedding) * m.importance
            scored.append((sim, m))
        scored.sort(reverse=True, key=lambda x: x[0])
        return [m for _, m in scored[:k]]

    def _cosine(self, a: list[float], b: list[float]) -> float:
        dot = sum(x * y for x, y in zip(a, b))
        na = (sum(x * x for x in a)) ** 0.5
        nb = (sum(x * x for x in b)) ** 0.5
        return dot / (na * nb) if na and nb else 0.0

    def clear(self):
        self.memories = []

第三层：编排循环

这是核心所在——自主决策发生的地方。编排器运行一个循环：观察 → 规划 → 执行 → 反思 → 存储。

# orchestrator.py
import json
from typing import Optional
from tools import ToolRegistry
from memory import VectorMemoryStore, Memory

class AgentOrchestrator:
    MAX_STEPS = 20
    PROGRESS_THRESHOLD = 0.1  # minimum semantic similarity to prior state

def __init__(
        self,
        llm_client,
        model: str,
        registry: ToolRegistry,
        memory: VectorMemoryStore,
        system_prompt: str = "",
    ):
        self.llm = llm_client
        self.model = model
        self.registry = registry
        self.memory = memory
        self.system_prompt = system_prompt or self._default_system_prompt()
        self.step_count = 0
        self.task_history: list[dict] = []

def _default_system_prompt(self) -> str:
        return """You are an autonomous AI agent. Your goal is to accomplish tasks by reasoning,
planning, and using tools. Think carefully before acting. Learn from observations and
build on past experiences stored in your memory. When unsure, search before guessing.
Keep your responses concise and action-oriented."""

def run(self, goal: str, context: str = "") -> dict:
        """Execute a goal autonomously. Returns execution trace."""
        self.step_count = 0
        self.task_history = []

# Store the initial goal as a memory
        self.memory.add(Memory(
            id=self._mkid("goal"),
            type="observation",
            content=goal,
            context=context,
            timestamp=datetime.now().isoformat(),
            importance=2.0,
        ))

messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": f"Goal: {goal}\n{f'Context: {context}' if context else ''}"},
        ]

trace = {"goal": goal, "steps": [], "final_output": None}

while self.step_count < self.MAX_STEPS:
            self.step_count += 1
            step = self._execute_step(messages, trace)
            trace["steps"].append(step)

if step["type"] == "success":
                trace["final_output"] = step["content"]
                break

if step["type"] == "blocked":
                trace["final_output"] = step.get("reason", "Agent could not complete the task.")
                break

return trace

def _execute_step(self, messages: list, trace: dict) -> dict:
        """Single orchestration step: recall → decide → act → reflect."""
        # 1. Recall relevant memories
        relevant = self.memory.recall(messages[-1]["content"], k=3)
        memory_context = ""
        if relevant:
            recalled = "\n".join(f"[{m.type}] {m.content}" for m in relevant)
            memory_context = f"\nRelevant past experience:\n{recalled}"
            # Add recalled memories as system context for this step
            messages.append({
                "role": "system",
                "content": f"Recalled context:{memory_context}",
            })

# 2. Get LLM decision
        response = self.llm.chat(self.model, messages)
        thought = response.get("content", "")
        tool_calls = response.get("tool_calls", [])

# 3. Execute tool calls if any
        if tool_calls:
            results = []
            for tc in tool_calls:
                tool_name = tc["function"]["name"]
                args = json.loads(tc["function"]["arguments"])
                try:
                    tool = self.registry.get(tool_name)
                    result = tool.fn(**args)
                    status = "success"
                except Exception as e:
                    result = f"Error: {e}"
                    status = "error"

results.append({"tool": tool_name, "result": result, "status": status})

# Store tool interaction as memory
                self.memory.add(Memory(
                    id=self._mkid(f"{tool_name}-{args}"),
                    type="tool_result",
                    content=str(result),
                    context=f"Calle

以下是如何端到端运行完整系统的方法：

# main.py
import json
from tools import ToolRegistry, registry
from memory import VectorMemoryStore
from orchestrator import AgentOrchestrator

# Minimal mock LLM client — swap with your actual provider
class MockLLMClient:
    """Replace this with OpenAI, Anthropic, or any chat-compatible client."""

def __init__(self):
        self.call_count = 0

def chat(self, model: str, messages: list) -> dict:
        """A deterministic mock that simulates agent reasoning."""
        self.call_count += 1
        last_msg = messages[-1]["content"] if messages else ""

# Simulate multi-step tool use for demonstration
        if "research" in last_msg.lower() or self.call_count <= 2:
            return {
                "content": "I need to search the web first, then analyze the results.",
                "tool_calls": [
                    {
                        "id": f"call_{self.call_count}",
                        "type": "function",
                        "function": {
                            "name": "search_web",
                            "arguments": json.dumps({"query": last_msg}),
                        },
                    }
                ],
            }

if "search_web" in last_msg or "result" in last_msg.lower():
            return {
                "content": "Based on my research, here is a comprehensive answer to the original question.",
                "tool_calls": [],
            }

return {
            "content": "I cannot complete this task without additional information or tools.",
            "tool_cal