使用 Elasticsearch 管理智能体记忆

在新兴的上下文工程学科中，在正确的时间为 AI 智能体提供正确的信息至关重要。上下文工程最重要的一个方面就是管理 AI 的记忆。AI 系统就像人类一样，依赖于短期记忆和长期记忆来回忆信息。如果我们希望大型语言模型 (LLM) 智能体能够进行逻辑对话、记住用户偏好，或基于先前的结果或响应进行构建，我们需要为它们配备有效的记忆机制。

毕竟上下文中的所有内容都会影响 AI 的响应。“垃圾进，垃圾出”说的就是这个道理。

本文将介绍短期记忆和长期记忆对 AI 智能体的意义，具体包括：

• 短期记忆和长期记忆的区别。

• 它们与使用向量数据库（如 Elasticsearch）的检索增强生成 (RAG) 技术有何关联，以及为什么需要细致的记忆管理。

• 忽略记忆（包括上下文溢出和上下文污染）有何风险。

• 最佳实践，如上下文修剪、总结和仅检索相关内容，以保持代理的记忆既有用又安全。

• 最后，我们将探讨如何使用 Elasticsearch 在多智能体系统中共享和传播记忆，使智能体能够协作而不会产生混乱。

AI 智能体中的短期记忆与长期记忆

AI 智能体的短期记忆通常指的是即时的对话上下文或状态——其本质上是活跃会话中的当前聊天历史记录或最近消息。这包括用户的最新查询和最近的来回交流。这与一个人在进行对话时脑海中记住的信息非常相似。

AI 框架通常会将这种瞬时记忆作为智能体状态的一部分来维护（例如使用检查点进程来存储对话状态，LangGraph 的此示例就介绍了这一点）。短期记忆存在于会话范围内；也就是说，它存在于单个会话或任务中，会话结束后即重置或清除，除非明确保存在其他地方。ChatGPT 中提供的临时聊天 就是会话范围内短期记忆的一个例子。

而长期记忆指的是跨越对话或会话持续存在的信息。这是智能体日积月累保留的知识，包括早前学习的事实，或我们告知其永久记住的用户偏好或任何数据。

长期记忆通常通过从外部源（如文件或向量数据库）存储和获取来配置，这些外部源位于即时上下文窗口之外。与短期聊天历史记录不同，长期记忆并非自动包含在每个提示中。相反，基于特定场景，智能体必须在调用相关工具时回忆或检索该信息。在实践中，长期记忆可能包括用户的个人资料信息、智能体先前生成的答案或分析，或者智能体可以查询的知识库。

例如，如果您有一个旅行规划智能体，短期记忆将包含当前行程查询的详细信息（日期、目的地、预算）以及该聊天中的任何后续问题；而长期记忆可以存储用户的一般旅行偏好、过去的行程和在之前会话中分享的其他事实。当用户在后面再次访问时，智能体可以从这个长期存储中提取资源（例如，用户喜欢海滩和山脉，平均预算为 10 万卢比，有愿望清单，更喜欢体验历史和文化而非适合儿童的景点），这样就不会每次都把用户当作一张白纸。

短期记忆（聊天历史记录）提供即时的上下文和连续性，而长期记忆则提供更广泛的背景，供智能体在需要时调用。大多数先进的 AI 智能体框架都能做到这两点：它们会跟踪最近的对话以保持上下文，并提供在长期信息库中查找或存储信息的机制。管理短期记忆可确保其保持在上下文窗口内，而管理长期记忆则能帮助智能体基于以往的互动和角色来构建答案。

上下文工程中的内存和 RAG

在实践中，我们如何让 AI 智能体拥有有用的长期记忆？

语义记忆是长期记忆的一个重要方法，通常通过检索增强生成 (RAG) 来配置。这需要将 LLM 与外部知识存储或支持向量的数据存储（如 Elasticsearch）耦合。当 LLM 需要提示或其内置训练之外的信息时，它会对 Elasticsearch 执行语义检索，并将最相关的结果作为上下文注入到提示中。通过这种方式，模型的有效上下文不仅包括最近的对话（短期记忆），还包括即时获取的相关长期事实。LLM 随后根据自身推理和检索到的信息来给出答案，它有效地将短期记忆和长期记忆结合起来，从而做出更准确和更感知上下文的响应。

Elasticsearch 可用于为 AI 智能体配置长期记忆。下面是一个高级示例，演示如何从 Elasticsearch 中检索上下文以配置长期记忆。

按照这种方式，智能体通过搜索相关数据来“记忆”，而不是将所有内容存储在有限的提示中，从而导致不同的风险。

将 RAG 与 Elasticsearch 或任何向量存储结合使用可带来诸多好处：

首先，它将模型的知识扩展到了训练截止点之外。智能体可以检索 LLM 可能不知晓的最新信息或特定领域的数据。这对于询问近期事件或专业话题至关重要。

其次，按需获取上下文有助于减少幻觉，尤其是当 LLM 未针对您的细分用例进行专有或高度专业化的数据训练时，这很有可能会导致出现幻觉。正如 OpenAI 最近的一篇论文（《Why Language Models Hallucinate》）强调的那样， LLM 不是像以往所激励的那样通过评估来猜测或编造新信息，而是以 Elasticsearch 中的事实参考为基础。当然， LLM 依赖于向量存储中数据的可靠性来真正防止错误信息，并根据核心相关性措施检索相关数据。

第三，RAG 支持智能体处理的知识库远远大于提示所能容纳的任何内容。RAG 不是将整个文档（例如长篇研究论文或政策文件）推送到上下文窗口，从而导致信息过载或无关信息上下文污染，而是依赖于分块。大型文档会被分解成语义上有意义的小块，系统只检索与查询最相关的几个数据块。这样一来，模型要显得知识渊博不需要长达数百万个词元的上下文来支撑；它只需要访问更大语料库中的正确数据块。

值得注意的是，随着 LLM 上下文窗口的扩大（一些模型现在支持数十万甚至数百万个词元），关于 RAG 是否“已死”的争论也随之出现。为什么不将所有数据推送到提示中？如果您有同样的疑惑，请参阅我的同事 Jeffrey Rengifo 和 Eduard Martin 撰写的精彩文章《Longer context ≠ better: Why RAG still matters》。这避免了“垃圾进，垃圾出”的问题：LLM 始终专注于少数重要内容，而不是应付噪音。

也就是说，将 Elasticsearch 或任何向量存储集成到 AI 智能体架构中可以提供长期记忆。智能体将知识存储在外部，并在需要时将其作为记忆上下文提取出来。这可以作为一种架构来实现，在每次用户查询后，智能体都会在 Elasticsearch 上搜索相关信息，然后在调用 LLM 之前将排序靠前的结果附加到提示中。如果响应包含有用的新信息，它也会保存回长期存储中（从而形成学习的反馈循环）。通过使用这种基于检索的记忆，智能体可以随时了解最新信息，无需将所有知识塞入每个提示，尽管上下文窗口支持一百万个词元。这种技术是上下文工程的基石，结合了信息检索和生成式 AI 的优势。

下面是一个在会话期间使用 LangGraph 的检查点系统管理记忆中对话状态的示例。（请参阅我们的支持上下文工程应用程序。）

# Initialize chat memory (Note: This is in-memory only, not persistent)
memory = MemorySaver()

# Create a LangGraph agent
langgraph_agent = create_react_agent(model=llm, tools=tools, checkpointer=memory)

...
...
# Only process and display checkpoints if verbose mode is enabled
if args.verbose:
    # List all checkpoints that match a given configuration
    checkpoints = memory.list({"configurable": {"thread_id": "1"}})
    # Process the checkpoints
    process_checkpoints(checkpoints)

以下是它存储检查点的方式：

Checkpoint:
Timestamp: 2025-12-30T09:19:41.691087+00:00
Checkpoint ID: 1f0e560a-c2fa-69ec-8001-14ee5373f9cf
User: Hi I'm Som, how are you? (Message ID: ad0a8415-5392-4a58-85ad-84154875bbf2)
Agent: Hi Som! I'm doing well, thank you! How about you? (Message ID: 
56d31efb-14e3-4148-806e-24a839799ece)
Agent:  (Message ID: lc_run--019b6e8e-553f-7b52-8796-a8b1fbb206a4-0)

Checkpoint:
Timestamp: 2025-12-30T09:19:40.350507+00:00
Checkpoint ID: 1f0e560a-b631-6a08-8000-7796d108109a
User: Hi I'm Som, how are you? (Message ID: ad0a8415-5392-4a58-85ad-84154875bbf2)
Agent: Hi Som! I'm doing well, thank you! How about you? (Message ID: 
56d31efb-14e3-4148-806e-24a839799ece)

Checkpoint:
Timestamp: 2025-12-30T09:19:40.349027+00:00
Checkpoint ID: 1f0e560a-b62e-6010-bfff-cbebe1d865f6

对于长期记忆，以下是在 Elasticsearch 上执行语义搜索的方法，以便在将检查点汇总并索引到 Elasticsearch 后使用向量嵌入检索以前的相关对话。

Functions: 
retrieve_from_elasticsearch() 

# Enhanced Elasticsearch retrieval with rank_window and verbose display
def retrieve_from_elasticsearch(query: str, k: int = 5, rank_window: int = None) -> tuple[List[Dict[str, Any]], str]:
    """
    Retrieve context from Elasticsearch with score-based ranking
    
    Args:
        query: Search query
        k: Number of results to return
        rank_window: Number of candidates to retrieve before ranking (default: args.rank_window)
        
    Returns:
        Tuple of (retrieved_documents, formatted_context_string)
    """
    if not es_client or not es_index_name:
        return [], "Elasticsearch is not available. Cannot search long-term memory."
    
    if rank_window is None:
        rank_window = args.rank_window
    
    try:
        # Check if index exists and has documents
        if not es_client.indices.exists(index=es_index_name):
            return [], "No previous conversations stored in long-term memory yet."
        
        # Get document count
        try:
            doc_count = es_client.count(index=es_index_name)["count"]
            if doc_count == 0:
                return [], "Long-term memory is empty. No previous conversations to search."
        except Exception as e:
            return [], f"Error checking memory: {str(e)}"
        
        # Generate embedding for the query
        try:
            query_embedding = embeddings.embed_query(query)
        except Exception as e:
            return [], f"Error generating embedding: {str(e)}"
        
        # Perform semantic search using kNN with rank_window
        try:
            search_body = {
                "knn": {
                    "field": "vector",
                    "query_vector": query_embedding,
                    "k": k,
                    "num_candidates": rank_window  # Retrieve more candidates, then rank top k
                },
                "_source": ["text", "content", "message_type", "timestamp", "thread_id"],
                "size": k
            }
            
            response = es_client.search(index=es_index_name, body=search_body)
            
            if not response.get("hits") or len(response["hits"]["hits"]) == 0:
                return [], "No relevant previous conversations found in long-term memory."
            
            # Extract documents with scores
            retrieved_docs = []
            for hit in response["hits"]["hits"]:
                source = hit["_source"]
                score = hit["_score"]
                retrieved_docs.append({
                    "content": source.get("content", source.get("text", "")),
                    "message_type": source.get("message_type", "unknown"),
                    "timestamp": source.get("timestamp", "unknown"),
                    "thread_id": source.get("thread_id", "unknown"),
                    "score": score
                })
            
            # Format context string
            context_parts = []
            for i, doc in enumerate(retrieved_docs, 1):
                context_parts.append(doc["content"])
            
            context_string = "\n\n".join(context_parts)
            
            # Verbose display
            if args.verbose:
                rich.print(f"\n[bold yellow]🔍 RETRIEVAL ANALYSIS[/bold yellow]")
                rich.print("="*80)
                rich.print(f"[blue]Query:[/blue] {query}")
                rich.print(f"[blue]Retrieved:[/blue] {len(retrieved_docs)} documents (from {rank_window} candidates)")
                rich.print(f"[blue]Total context length:[/blue] {len(context_string)} characters\n")
                
                for i, doc in enumerate(retrieved_docs, 1):
                    rich.print(f"[cyan]📄 Document {i} | Score: {doc['score']:.4f} | Type: {doc['message_type']}[/cyan]")
                    rich.print(f"[cyan]   Timestamp: {doc['timestamp']} | Thread: {doc['thread_id']}[/cyan]")
                    content_preview = doc['content'][:200] + "..." if len(doc['content']) > 200 else doc['content']
                    rich.print(f"[cyan]   Content: {content_preview}[/cyan]")
                    rich.print("-" * 80)
            
            return retrieved_docs, context_string
            
        except Exception as e:
            return [], f"Error searching memory: {str(e)}"
            
    except Exception as e:
        return [], f"Error accessing long-term memory: {str(e)}"

既然我们已经探讨了如何利用 LangGraph 的检查点在 Elasticsearch 中索引和提取短期记忆和长期记忆，接下来让我们花点时间了解为什么索引和转储完整对话可能存在风险。

不管理上下文内存的风险

我们花了大量篇幅讨论了上下文工程以及短期和长期记忆，接下来让我们了解如果不好好管理智能体的记忆和上下文会发生什么。

遗憾的是，当 AI 的上下文变得极其庞大或包含不良信息时，很多事情都可能会出错。随着上下文窗口变大，新的失效模式也会随之出现，例如：

• 上下文污染

• 上下文干扰

• 上下文混淆

• 上下文冲突

• 上下文泄露和知识冲突

• 幻觉和错误信息

让我们来分析一下这些问题以及因上下文管理不善而产生的其他风险：

上下文污染

上下文污染指的是错误或有害信息混入上下文中，并“污染”模型的后续输出。一个常见的例子是模型产生的幻觉被当作事实并插入到对话历史记录中。然后，该模型可能会在以后的响应中以该错误为基础，从而使错误更加严重。在迭代智能体循环中，一旦虚假信息进入共享上下文（例如智能体工作笔记的摘要），它可能会被反复强化。

DeepMind 的研究人员在 Gemini 2.5 报告（TL;DR，请点击此处查看）中观察到，一个长期运行的 Pokémon 游戏智能体出现了这种情况：如果智能体产生一个错误的游戏状态幻觉，并且该状态被记录到它的上下文（它对目标的记忆）中，那么智能体就会围绕一个不可能完成的目标形成毫无意义的策略，从而陷入困境。换句话说，受污染的记忆会让智能体无限期地走上错误的道路。

上下文污染可能是无意的（误操作），也可能是恶意的，例如通过提示注入攻击，用户或第三方偷偷输入隐藏指令或错误事实，智能体随后记住并遵循这些指令或事实。

建议的应对措施：

根据来自 Wiz、Zerlo 和 Anthropic 的见解，针对上下文污染的对策主要是防止不良或误导性信息进入 LLM 的提示、上下文窗口或检索管道。关键步骤包括：

• 不断检查上下文：监测对话或检索到的文本中是否有任何可疑或有害内容，而不仅仅是监测起始提示。

• 使用可信来源：根据可信度对文档进行评分或标记，以便系统优先选择可靠的信息，并忽略低分数据。

• 发现异常数据：使用工具检测异常、不合适或被篡改的内容，并在模型使用前将其删除。

• 过滤输入和输出：添加护栏，防止有害或误导性文本进入系统或被模型重复。

• 用干净的数据不断更新模型：定期用经过验证的信息刷新系统，以纠正任何漏网的不良数据。

• 人机协同：安排人员审查重要的输出或将其与已知的可信来源进行比较。

简单的用户习惯也很有帮助，比如重置冗长的聊天记录、只分享相关信息、将复杂的任务分解成更小的步骤，以及在模型外保留干净的备注。

这些措施可共同构建多层防御，保护 LLM 免受上下文污染，并保持输出的准确性和可信度。

如果不采取这里提到的对策，智能体可能会记住一些指令，比如忽略以前的指南或攻击者插入的琐碎事实，从而导致得到有害的输出。

上下文干扰

上下文干扰是指当上下文变得过长时，模型过度关注上下文而忽视在训练过程中学到的内容。在极端情况下，这类似于灾难性遗忘；也就是说，模型会有效地“遗忘”它的底层知识，变得过分依赖摆在它面前的信息。先前的研究表明，当提示过长时，LLM 往往会失去注意力。

以 Gemini 2.5 智能体为例，它支持百万词元级别的窗口，但当其上下文增长超过一定程度时（在实验中约为 100000 个词元），它会开始专注于重复其过去的操作，而不是提出新的解决方案。从某种意义上说，该智能体成了其广泛历史的囚徒。它不停地查看以前的动作记录（上下文）并模仿它们，而不是利用其底层的训练知识来制定新颖的策略。

这只会适得其反。我们希望模型能够利用相关的上下文来帮助推理，而不是让上下文凌驾于其思考能力之上。值得注意的是，即使是那些拥有巨大窗口的模型也会表现出这种上下文腐烂：随着词元的增加，它们的性能会不均匀地下降。它们仿佛有注意力预算。就像人类工作记忆有限一样， LLM 对词元的关注能力也是有限的，随着预算捉襟见肘，其精准度和专注度会下降。

作为缓解措施，您可以通过分块、工程化正确信息、定期总结上下文以及利用评分来评估和监控响应的准确性来防止上下文干扰。

这些方法可使模型始终基于相关的上下文和其底层训练，从而降低干扰的风险，并提高整体推理质量。

上下文混淆

上下文混淆是指模型使用上下文中的多余内容生成低质量响应的情况。一个典型的例子是为智能体提供它可能会用到的大量工具或 API 定义。如果这些工具有很多与当前任务无关，模型可能仍然会试图不恰当地使用它们，仅仅因为它们出现在上下文中。实验发现，提供过多非必需的工具或文档可能会降低性能。智能体会开始出错，例如调用错误的函数或引用不相关的文本。

在一个案例中，一个小型的 Llama 3.1 8B 模型在有 46 个工具可供考虑时未能完成任务，在只有 19 个工具可供考虑时却成功完成了任务。额外的工具造成了混乱，尽管上下文的长度没有超出限制。根本问题在于，提示中的任何信息都会被模型关注。如果它不知道忽略某些内容，那么这些内容可能会以不希望的方式影响其输出。不相关的内容可能会“窃取”模型的一些注意力并将其引入歧途（例如，不相关的文档可能会导致智能体答非所问）。上下文混淆通常表现为模型做出低质量的反应，将不相关的上下文整合在一起。参考研究论文：《Less is More: Optimizing Function Calling for LLM Execution on Edge Devices》。

这提醒我们，上下文不一定越多越好，尤其是在没有进行相关性管护的情况下。

上下文冲突

上下文冲突是指上下文的某些部分相互矛盾，导致内部不一致，从而破坏模型的推理。如果智能体积累了多条相互冲突的信息，上下文冲突就会发生。

例如，想象一个智能体从两个来源获取数据：一个说 A 航班下午 5 点起飞，另一个说 A 航班下午 6 点起飞。如果两个事实都出现在上下文中，可怜的模型无法知道哪个是正确的；它可能会混淆或生成不正确或不相似的答案。

上下文冲突也经常发生在多轮对话中，因为模型前期的回答尝试会与后来完善的信息一起留在上下文中。

Microsoft 和 Salesforce 的一项研究显示，如果将复杂查询分解为多个聊天机器人回合（逐步添加细节），与在单个提示中提供所有细节相比，前者的最终准确性会显著下降。为什么？因为前期的回合包含来自模型的部分或不正确的中间答案，而这些答案会保留在上下文中。当模型后来尝试用所有信息来回答问题时，它的记忆中仍然包含那些错误的尝试，这些尝试与更正后的信息相冲突，并导致模型偏离正轨。从根本上说，对话的上下文与对话本身发生了冲突。模型可能会无意中使用已经过时的上下文（来自前期的对话轮），这些上下文在添加新信息后不再适用。

在智能体系统中，上下文冲突尤其危险，因为智能体可能会结合来自不同工具或子智能体的输出。如果这些输出不一致，则汇总的上下文就不一致。这样一来，智能体在试图调和矛盾时就会陷入困境或产生无意义的结果。防止上下文冲突需要确保上下文的新鲜度和一致性，例如清除或更新任何过时的信息，不混用未经一致性审查的来源。

上下文泄露和知识冲突

在多个智能体或用户共享记忆存储的系统中，存在信息在上下文间流失的风险。

例如，如果两个独立用户的数据嵌入存在于同一向量数据库中且没有适当的访问控制，响应用户 A 查询的智能体可能会意外检索用户 B 的部分记忆。这种跨上下文泄露可能会暴露私人信息，或在响应中造成混乱。

根据“OWASP 定义的LLM 应用十大风险”，多租户向量数据库必须防范此类泄露：

根据《LLM 08:2025 向量和嵌入弱点，常见的风险之一是上下文泄露：

在多租户环境中，多类用户或应用共享同一个向量数据库，用户或查询之间存在上下文泄露的风险。当来自多个来源的数据相互矛盾时，可能会出现数据联合知识冲突错误。当 LLM 无法使用来自检索增强的新数据取代其在训练中学到的旧知识时，这种情况也可能会发生。

另一方面，LLM 可能难以用记忆中的新信息覆盖其内置的知识。如果模型是根据某个事实进行训练的，而检索到的上下文却表明了相反的情况，那么模型可能会对应该相信哪个事实感到困惑。如果没有适当的设计，智能体可能会混淆上下文或未能用新证据更新旧知识，从而导致得出过时或不正确的答案。

幻觉和错误信息

即使没有长上下文的干扰，幻觉（LLM 编造听起来合理但实际上是虚假的信息）也已是一个老生常谈的问题，而糟糕的记忆管理会加剧这个问题。

如果智能体的记忆缺少一个关键事实，模型可能会用猜测来填补空白，如果这个猜测随后进入上下文（使其受污染），错误就会持续存在。

OWASP LLM 安全报告（LLM09:2025 错误信息）强调错误信息是一个核心漏洞：LLM 可以产生自信但虚构的答案，而用户可能会过度信任它们。一个拥有不良或过时长期记忆的智能体可能会自信地引用去年真实但现在错误的内容，除非其记忆保持更新。

过度依赖 AI 输出（无论是用户还是智能体本身在循环中过度依赖）会使情况变得更糟。如果没有人定期检查记忆中的信息，智能体可能会积累虚假信息。这就是 RAG 经常被用来减少幻觉的原因：检索权威来源，模型就不必编造事实。但如果您的检索拉取了错误的文档（比如包含错误信息的文档），或者早期幻觉没有被去除，系统可能会在所有操作中传播这些错误信息。

总之，记忆管理不善可能会导致错误和误导性的输出，这可能会造成损害，尤其是在高风险情况下（例如在金融或医疗领域提供错误建议）。智能体需要机制来验证或纠正其记忆内容，而不是无条件地信任上下文中的任何内容。

总的来说，给 AI 智能体无限长的记忆或将所有可能的东西都转储到其上下文中并非成功的秘诀。

LLM 应用程序中内存管理的最佳实践

为了避免上述陷阱，开发人员和研究人员为 AI 系统的上下文和记忆管理设计了许多最佳实践。这些做法旨在使 AI 的工作上下文保持精简、相关且更新的状态。以下是一些关键策略，以及它们如何发挥作用的示例。

RAG：使用针对性上下文

RAG 的大部分内容在前面的章节中已有介绍，所以这里仅作简要的实用提醒：

• 使用有针对性的检索，而不是批量加载：只检索最相关的片段，而不是将整个文档或完整的对话历史记录推送到提示中。

• 将 RAG 视为即时记忆调用：仅在需要时才获取上下文，而不是将所有内容跨轮次传递。

• 优先使用感知相关性的检索策略：top-k 语义搜索、倒数排序融合或工具装载过滤等方法有助于减少噪音并提高基础。

• 扩大上下文窗口并不会消除对 RAG 的需求：两个高度相关的段落几乎总是比 20 页松散相关的内容更有效。

也就是说，RAG 并不是要增加更多的上下文，而是要增加合适的上下文。

工具装载

工具装载是指只给模型提供它在执行任务时实际需要的工具。这个词源于游戏：您要选择一种适合当下情况的装备。工具太多会减慢您的速度；错误的工具会导致失败。根据研究论文《Less is more》，LLM 也是如此。当工具数量超过 30 个左右时，描述会开始重叠，模型会变得混乱。当工具数量超过约 100 个时，失败几乎是必然的。这不是一个上下文窗口问题，而是上下文混淆的问题。

RAG-MCP 是一个简单有效的解决办法。它不是将每个工具都放入提示中，而是将工具描述存储在向量数据库中，每个请求只检索最相关的工具。在实际操作中，这样可以使装载保持小型化和专注化，大幅缩短提示，并且可以将工具选择的准确性至多提高 3 倍。

小模型甚至会更快出现这样的问题。研究表明，8B 模型在装载数十种工具时会失败，但在精简装载后会变得成功。动态选择工具（有时先由 LLM 推理它认为需要的工具）可将性能提高 44% ，同时还能降低功耗和缩短延迟。关键点是，大多数智能体只需要少量工具，但随着系统的发展，设计决策需要首先考虑工具装载和 RAG-MCP。

上下文修剪：限制聊天历史记录的长度

如果对话持续了很多轮，累积的聊天历史记录可能会变得太大而无法容纳，导致上下文溢出或对模型造成过多干扰。

修剪是指随着对话的增加，以编程方式删除或缩短对话中不太重要的部分。一种简单的形式是，当您达到一定限制时，删除对话中最早的轮次，只保留最新的 N 条消息。更复杂的修剪可能涉及删除无关的题外话或以前不再需要的指令。修剪的目标是保持上下文窗口不受旧新闻干扰。

例如，如果智能体在 10 轮对话前解决了一个子问题，并且我们已经翻篇，我们可以从上下文中删除该部分历史记录（假设我们已不需要该部分）。许多基于聊天功能的实现方式都是如此：它们会维护一个滚动显示最新消息的窗口。

修剪可以很简单，比如在对话的最早部分被总结或被认为无关紧要后“忘记”这些部分。这样一来，我们就能降低上下文溢出错误的风险，也能减少上下文干扰，使模型不会被旧的或偏离主题的内容干扰。这种方法非常类似于人类可能记不住一个小时谈话中的每一个字，但会记住重点。

如果您对上下文修剪感到困惑，正如作者 Drew Breunig 在此强调的那样，使用 Provence 模型 (`naver/provence-reranker-debertav3-v1`) 可能会有所帮助。Provence 模型是一个轻量级 (1.75 GB)、高效且准确的问答上下文修剪器。它可以将大型文档裁剪为仅与给定查询最相关的文本。您可以按特定时间间隔调用它。

以下是我们在代码中调用 `provence-reranker` 模型来修剪上下文的做法：

# Context pruning with Provence
def prune_with_provence(query: str, context: str, threshold: Optional[float] = None) -> str:
    """
    Prune context using Provence reranker model
    
    Args:
        query: User's query/question
        context: Original context to prune
        threshold: Relevance threshold (0-1) for Provence reranker.
                   If None, uses args.pruning_threshold.
                   0.1 = conservative (recommended, no performance drop)
                   0.3-0.5 = moderate to aggressive pruning
    
    Returns:
        Pruned context with only relevant sentences
    """
    if provence_model is None:
        return context
    
    if threshold is None:
        threshold = args.pruning_threshold
    
    try:
        # Use Provence's process method
        provence_output = provence_model.process(
            question=query,
            context=context,
            threshold=threshold,
            always_select_title=False,
            enable_warnings=False
        )
        
        # Extract pruned context from output
        pruned_context = provence_output.get('pruned_context', context)
        reranking_score = provence_output.get('reranking_score', 0.0)
        
        # Log statistics
        original_length = len(context)
        pruned_length = len(pruned_context)
        reduction_pct = ((original_length - pruned_length) / original_length * 100) if original_length > 0 else 0
        
        if args.verbose:
            rich.print(f"[cyan]📊 Pruning stats: {pruned_length}/{original_length} chars ({reduction_pct:.1f}% reduction, threshold={threshold:.2f}, rerank_score={reranking_score:.3f})[/cyan]")
        
        return pruned_context if pruned_context else context
        
    except Exception as e:
        rich.print(f"[yellow]⚠️ Error in Provence pruning: {str(e)}[/yellow]")
        rich.print(f"[yellow]⚠️ Falling back to original context[/yellow]")
        return context

我们使用 Provence 重排序模型 (`naver/provence-reranker-debertav3-v1`) 来对句子的相关性进行评分。基于阈值的过滤功能可将句子保留在相关性阈值之上。此外，我们引入了一种回退机制，如果修剪失败，我们将返回原始上下文。最后，统计日志在详细模式下跟踪减少百分比。

上下文总结：将旧信息浓缩而非完全放弃

总结是对修剪的补充。当历史记录或知识库变得过于庞大时，您可以使用 LLM 生成重要内容的简短总结，并在未来使用该总结代替完整内容，就像我们在上面的代码中所做的那样。

例如，如果 AI 助手进行了 50 轮对话，系统不是在第 51 轮将全部 50 轮对话发送给模型（很可能容纳不下），而是选择第 1 至 40 轮，让模型用一段话对其总结，然后在下一个提示中只提供该总结和最后的 10 轮对话。这样一来，模型无需每个细节也能知道讨论的内容。早期的聊天机器人用户是手动总结的，他们问聊天机器人：“你能总结一下我们到目前为止谈过的内容吗？”然后带着总结继续进行新的会话。现在总结可以自动进行。总结不仅可以节省上下文窗口的空间，还可以通过去除多余的细节并只保留重要的事实来减少上下文混淆/干扰。

以下展示了我们如何使用 OpenAI 模型（您可以使用任何大型语言模型）来压缩上下文，同时保留所有相关信息，并消除冗余和重复。

# Context summarization
def summarize_context(query: str, context: str) -> str:
    """
    Summarize context using LLM to reduce duplication and focus on relevant information
    
    Args:
        query: User's query/question
        context: Context to summarize
        
    Returns:
        Summarized context
    """
    try:
        summary_prompt = f"""You are an expert at summarizing conversation context.

Your task: Analyze the provided conversation context and produce a condensed summary that fully answers or supports the user's specific question.

The summary must:
1. Preserve every fact, detail, and information that directly relates to the question
2. Eliminate redundancy and duplicate information
3. Maintain chronological flow when relevant
4. Focus on information that helps answer: "{query}"

Context to summarize:
{context}

Provide a concise summary that preserves all relevant information:"""

        summary = llm.invoke(summary_prompt).content
        
        if args.verbose:
            original_length = len(context)
            summary_length = len(summary)
            reduction_pct = ((original_length - summary_length) / original_length * 100) if original_length > 0 else 0
            rich.print(f"[cyan]📝 Summarization stats: {summary_length}/{original_length} chars ({reduction_pct:.1f}% reduction)[/cyan]")
        
        return summary
        
    except Exception as e:
        rich.print(f"[yellow]⚠️ Error in context summarization: {str(e)}[/yellow]")
        rich.print(f"[yellow]⚠️ Falling back to original context[/yellow]")
        return context

重要的是，当上下文得到总结后，模型被琐碎细节或过往错误牵制的可能性会降低（假设总结是准确的）。

不过，总结必须谨慎进行。糟糕的总结可能会遗漏关键细节，甚至引入错误。它本质上是对模型的另一个提示（“总结这个”），因此它可能会产生幻觉或失去细微差别。最佳做法是逐步总结，或许可以保留一些典型事实，不对它们进行总结。

尽管如此，它已经被证明非常有用。在 Gemini 智能体场景中，每隔约 10 万个词元对上下文进行总结是抵消模型重复倾向的一种方法。总结就像对话或数据的压缩记忆。作为开发人员，我们可以通过让智能体针对对话历史记录或长文档定期调用总结函数（可能是一个较小的 LLM 或一个专用例程）来实现这一点。得到的总结会在提示中替代原始内容。这种策略已被广泛使用，以便将上下文限制在一定范围内，并提炼信息。

上下文隔离：尽可能隔离上下文

这在复杂的智能体系统或多步骤工作流程中更为重要。上下文隔离的理念是将一个大任务拆分成较小的孤立任务，每个任务都有自己的上下文，这样就不会积累一个包含所有内容的庞大上下文。每个子智能体或子任务使用特定的上下文处理问题的某一部分，然后由更高级的智能体、主管或协调员整合结果。

Anthropic 的研究策略使用多个子智能体，每个子智能体研究问题的不同方面，各自拥有自己的上下文窗口，而主智能体负责阅读这些子智能体提炼的结果。这种并行的模块化方法意味着没有一个上下文窗口会变得过于臃肿。这也减少了无关信息混淆的可能性，每个线程都紧扣主题（避免上下文混淆），而且在回答具体子问题时不会携带不必要的包袱。从某种意义上说，这就像是在运行独立的思维线程，它们只分享各自的结果，而不是整个思维过程。

在多智能体系统中，这种方法至关重要。如果智能体 A 正在处理任务 A 而智能体 B 正在处理任务 B，那么除非确实需要，否则任何一个智能体都没有理由使用另一个智能体的完整上下文。智能体可以只交换必要的信息。例如，智能体 A 可以通过一个主管智能体将其发现的综合总结传递给智能体 B，同时每个子智能体都维护自己的专用上下文线程。这种设置不需要人机协同干预；它依赖于一个具有启用工具的主管智能体，并进行最小化和受控的上下文共享。

在设计系统时，尽量减少智能体或工具在运行时的必要上下文重叠，可以大大提高系统的清晰度和性能。您可以把它想象成 AI 微服务，每个组件处理自己的上下文，您以受控的方式在它们之间传递消息，而不是在一个单一的上下文中传递消息。这些最佳实践通常会结合使用。此外，这还能让您灵活地修剪琐碎的历史记录，总结重要的旧消息或对话，将详细日志卸载到 Elasticsearch 以获得长期上下文，并在需要时使用检索功能调回任何相关内容。

正如此处提到的那样，我们的指导原则是上下文是一种有限且宝贵的资源。我们希望提示中的每个词元都能发挥作用，换句话说，它应该对输出的质量有所帮助。如果记忆中的某些东西没有发挥其应有的作用（或者更糟糕的是，它们会主动造成混乱），那么我们就应该对其进行修剪、总结或将其去除。

作为开发者，我们现在可以像编写代码一样编程上下文，决定包含哪些信息，如何格式化它，以及何时删除或更新它。通过遵循这些实践，我们可以为 LLM 智能体提供所需的上下文，使其能够执行任务，而不会陷入前面描述的失败模式。其结果是，智能体能够记住应该记住的内容，忘记不需要的内容，并及时检索所需的内容。

结论

记忆不是您添加到智能体中的东西；它是您设计的一部分。短期记忆是智能体的工作记忆板，而长期记忆是其持久的知识存储。RAG 是两者之间的桥梁，将被动的数据存储（如 Elasticsearch）转变为一个主动的回忆机制，能够为输出提供依据并保持智能体的最新状态。

但记忆是把双刃剑。当您让上下文不受控制地增长时，您会导致上下文污染、干扰、混乱和冲突，在共享系统中甚至会导致数据泄露。这就是为什么最重要的记忆工作不是“多存储”，而是“更好地管护”：有选择地检索，积极修剪，仔细总结，避免混合无关的上下文（除非任务真的需要）。

在实践中，好的上下文工程看起来就像良好的系统设计：上下文更小和更充分、组件之间的交互受控、原始状态和您实际希望模型看到的提炼状态清晰分离。如果方法得当，您最终得到的不是一个什么都记得的智能体，而是一个在正确的时间，出于正确的原因，记住正确事情的智能体。