Agent 上下文窗口管理:当 System Prompt 太长时该怎么办

Agent 上下文窗口管理:当 System Prompt 太长时该怎么办

Agent 上下文窗口管理:当 System Prompt 太长时该怎么办

一、深度引言与场景痛点

在写这篇文章之前,我刚翻了一下手头某个 Agent 项目的 System Prompt——2937 个 token,包含了角色设定、工具说明、输出格式规范、安全策略、知识库索引指南、示例对话……读完之后我突然理解了 LLM 的处境:你给它的 System Prompt 越长,它在"记住自己是谁"这件事上花的心智就越多,留给"真正思考问题"的注意力和 token 预算就越少

这个问题的后果不是"效果稍微差一点",而是断崖式下降。在最近一次 A/B 测试中,我们发现:当 System Prompt 从 500 token 增长到 3000 token 后,Agent 的幻觉率上升了 23%,工具调用错误率上升了 17%,尤其是在多轮对话的第三轮之后,Agent 开始"忘记"中间的检索结果,反复要求用户提供已经有过的信息。

为什么会这样?因为 LLM 的注意力机制本质上是一种加权分配机制——它无法"均匀关注"上下文中的每一个 token,而是会给某些部分更高的注意力。System Prompt 越长,它就越是"霸占"注意力资源,让对话历史、检索结果这些真正动态变化的信息处于"注意力边缘"。在多轮对话中,随着新 token 的持续涌入,旧的 System Prompt 虽然还在上下文里,但 LLM 对它的"关注度"会逐步衰减。

本文不讨论"如何写更简洁的 Prompt"这种写作技巧,而是聚焦于工程层面的上下文窗口管理策略——如何使用滑动窗口、摘要压缩、分层结构和动态指令注入来高效利用有限的 token 预算。

二、底层机制与原理深度剖析

LLM 的上下文窗口就像一张固定大小的餐桌——你能摆上去的菜是有限的。如果前菜(System Prompt)摆了太多,主菜(用户问题和检索结果)就没地方放了。

flowchart TD A[上下文窗口总预算: 128K tokens] --> B{Token 预算分配} B --> C[System Prompt: 500-2000 tokens] B --> D[对话历史: 动态增长] B --> E[检索结果: 按需注入] B --> F[工具调用记录: 累积] B --> G[预留余量: 10%] C --> H{System Prompt 膨胀} H -->|超限| I[策略1: 分层加载] H -->|超限| J[策略2: 摘要压缩] H -->|超限| K[策略3: 动态注入] subgraph I_Detail [分层加载] I1[核心层: 角色 + 安全规则 常驻] --> I2[工具层: 按需激活] I2 --> I3[知识层: 检索时注入] end subgraph J_Detail [摘要压缩] J1[滚动摘要: 旧对话压缩为摘要] --> J2[System Prompt 自身摘要] end subgraph K_Detail [动态注入] K1[识别当前意图] --> K2[仅注入相关指令] K2 --> K3[无关规则暂时屏蔽] end

理解上下文窗口管理的关键在于区分**"哪些是必须常驻的"和"哪些可以随用随取的"**:

  • 常驻内容:角色定义、安全策略——这些丢失后 Agent 的基本行为会失控。但它们应该尽量精简,控制在 500 token 以内。
  • 半常驻内容:工具 API 描述——只有当 Agent 真正需要调用工具时才需要这些信息。可以在 Agent 决定调用工具后再注入。
  • 临时内容:检索结果、示例对话——只在当前请求上下文中有效,上一轮的检索结果可以被压缩成摘要。

三、生产级代码实现

import asyncio from dataclasses import dataclass, field from typing import List, Dict, Optional, Any, Callable from enum import Enum import tiktoken class ContentTier(Enum): """内容分层""" CORE = "core" # 常驻:角色 + 安全规则 TOOL = "tool" # 半常驻:工具定义 CONTEXTUAL = "contextual" # 临时:检索结果 KNOWLEDGE = "knowledge" # 按需:领域知识 @dataclass class PromptBlock: """Prompt 内容块""" content: str tier: ContentTier name: str token_count: int = 0 def __post_init__(self): if not self.token_count: enc = tiktoken.get_encoding("cl100k_base") self.token_count = len(enc.encode(self.content)) class ContextWindowManager: """上下文窗口管理器 核心能力: 1. Token 预算分配 2. 动态内容注入 3. 对话历史摘要压缩 4. 溢出保护 """ def __init__( self, model_max_tokens: int = 128000, reserve_ratio: float = 0.10, max_completion_tokens: int = 4096, ): self.model_max_tokens = model_max_tokens self.reserve_ratio = reserve_ratio self.max_completion_tokens = max_completion_tokens # 预算分配 self.available_tokens = int( model_max_tokens * (1 - reserve_ratio) - max_completion_tokens ) # 内容层 self.core_blocks: List[PromptBlock] = [] self.tool_blocks: List[PromptBlock] = [] self.conversation_history: List[Dict] = [] self.history_summary: str = "" @property def core_token_usage(self) -> int: return sum(b.token_count for b in self.core_blocks) @property def tool_token_usage(self) -> int: return sum(b.token_count for b in self.tool_blocks) def add_core(self, name: str, content: str): """添加核心层内容""" block = PromptBlock(content=content, tier=ContentTier.CORE, name=name) self.core_blocks.append(block) if self.core_token_usage > self.available_tokens * 0.3: raise ValueError( f"核心层 token 超限: {self.core_token_usage} > " f"{int(self.available_tokens * 0.3)},请精简 System Prompt" ) def add_tool(self, name: str, description: str): """添加工具定义(懒加载,不立即计入预算)""" block = PromptBlock(content=description, tier=ContentTier.TOOL, name=name) self.tool_blocks.append(block) def add_message(self, role: str, content: str): """添加对话消息""" enc = tiktoken.get_encoding("cl100k_base") token_count = len(enc.encode(content)) self.conversation_history.append({ "role": role, "content": content, "token_count": token_count, }) # 检查是否超出预算 total = self._estimate_total_tokens() if total > self.available_tokens: self._compress_history() def _estimate_total_tokens(self) -> int: """估算当前总 token 用量""" history_tokens = sum(m["token_count"] for m in self.conversation_history) summary_tokens = len(tiktoken.get_encoding("cl100k_base").encode(self.history_summary)) return ( self.core_token_usage + history_tokens + summary_tokens + 500 # 格式开销估计 ) def _compress_history(self): """摘要压缩对话历史""" if len(self.conversation_history) < 4: # 历史不太长,不做压缩 return # 保留最近 3 轮对话,更早的压缩为摘要 keep_recent = 3 old_messages = self.conversation_history[:-keep_recent] recent_messages = self.conversation_history[-keep_recent:] # 构建摘要上下文 summary_context = "以下是之前对话的摘要,请在后续对话中参考:\n" for msg in old_messages: summary_context += f"[{msg['role']}]: {msg['content'][:200]}...\n" # 在生产环境中,这里调用 LLM 生成高质量摘要 # 此处简化为拼接截断 self.history_summary = summary_context[:800] # 限制摘要长度 self.conversation_history = recent_messages # 将摘要作为虚拟系统消息插入 self.conversation_history.insert(0, { "role": "system", "content": self.history_summary, "token_count": len(tiktoken.get_encoding("cl100k_base").encode(self.history_summary)), }) def build_context( self, user_query: str, retrieval_results: Optional[List[str]] = None, activated_tools: Optional[List[str]] = None, ) -> List[Dict[str, str]]: """构建最终发送给 LLM 的上下文 这是核心方法:根据当前请求动态组装上下文 """ messages = [] # 1. 始终包含核心层 core_content = "\n".join(b.content for b in self.core_blocks) messages.append({"role": "system", "content": core_content}) # 2. 按需注入工具定义(只注入当前请求相关的) if activated_tools: tool_content = "以下工具当前可用:\n" for tool_name in activated_tools: for block in self.tool_blocks: if block.name == tool_name: tool_content += f"\n--- {tool_name} ---\n{block.content}" messages.append({"role": "system", "content": tool_content}) # 3. 注入检索结果(带长度保护) if retrieval_results: max_retrieval_tokens = int(self.available_tokens * 0.4) retrieval_content = "以下是与问题相关的知识库内容:\n" current_tokens = 0 enc = tiktoken.get_encoding("cl100k_base") for i, result in enumerate(retrieval_results): result_tokens = len(enc.encode(result)) if current_tokens + result_tokens > max_retrieval_tokens: retrieval_content += f"\n(共 {len(retrieval_results)} 条结果,此处仅展示前 {i} 条)" break retrieval_content += f"\n[文档 {i+1}]\n{result}" current_tokens += result_tokens messages.append({"role": "system", "content": retrieval_content}) # 4. 对话历史 for msg in self.conversation_history: messages.append({"role": msg["role"], "content": msg["content"]}) # 5. 当前用户问题 messages.append({"role": "user", "content": user_query}) return messages def get_budget_report(self) -> Dict[str, Any]: """获取预算使用报告""" total = self._estimate_total_tokens() return { "available": self.available_tokens, "used": total, "usage_pct": round(total / self.available_tokens * 100, 1), "core_blocks": len(self.core_blocks), "core_tokens": self.core_token_usage, "history_messages": len(self.conversation_history), "has_summary": bool(self.history_summary), } # ================== 使用示例 ================== async def demo(): manager = ContextWindowManager(model_max_tokens=128000, max_completion_tokens=4096) # 设置核心层(尽量精简) manager.add_core("role", "你是一个技术助手,擅长 Python 和 AI 开发。回答应准确、简洁。") manager.add_core("safety", "不得生成有害内容。不确定时请明确告知。") # 注册工具(懒加载) manager.add_tool( "search_docs", "搜索内部文档。参数: query(string, 必需) - 搜索关键词。返回: 相关文档列表。", ) manager.add_tool( "run_python", "执行 Python 代码。参数: code(string, 必需) - Python 代码。返回: 执行结果。", ) # 模拟多轮对话 for i in range(10): manager.add_message("user", f"Python 中的 async/await 是怎么工作的?轮次 {i}") manager.add_message("assistant", f"async/await 是协程的基础,轮次 {i} 的回答...") # 构建上下文(当前请求只需要 search_docs 工具) context = manager.build_context( user_query="用代码演示一下 asyncio.gather 的用法", retrieval_results=[ "asyncio.gather 用于并发执行多个协程。它会等待所有协程完成,返回结果列表。", "使用 gather 时应注意异常处理。如果 return_exceptions=True,异常会作为结果返回。", ], activated_tools=["search_docs"], ) print(manager.get_budget_report()) print(f"构建的上下文消息数: {len(context)}") if __name__ == "__main__": asyncio.run(demo())

四、边界分析与架构权衡

1. 分层架构的成本

分层虽然解决了 System Prompt 过长的问题,但引入了注入延迟——每次请求前需要根据意图判断应该注入哪些内容。这个判断逻辑如果也依赖一次 LLM 调用(做意图分类),那等于是"为了省 token 多耗了一次 token"。建议使用简单的关键词匹配或分类器来做工具激活判断,而不是每次都问 LLM"你需要哪个工具"。

2. 摘要压缩的信息损失

历史摘要压缩是一个不可逆的信息损失过程——旧的对话细节一旦被压缩为摘要,就再也回不来了。如果 Agent 在压缩后的发言中说"我刚才提到的那三个参数",这个"三个参数"在摘要里可能就只剩"参数"两个字了。压缩策略需要根据场景合理设置阈值:

  • 技术支持 Agent:保留更长的原始历史(技术细节容易在摘要中丢失)
  • 闲聊 Agent:积极压缩(情绪和上下文概括就够了)
  • 写作 Agent:保留原文(写作风格需要在细粒度上保持一致)

3. 工具定义的"胖客户端"问题

如果你的 Agent 有 50+ 个工具,即使不全部注入,光是需要"判断注入哪些"的决策逻辑就已经很复杂了。解决思路是工具分组 + 渐进式暴露——先把工具按业务领域分组,LLM 先看到的是"工具组"而非具体工具,当它选择某个组后,再展开该组内的具体工具定义。

4. 模型行为的"长度偏好"

值得注意的是,不同模型对 System Prompt 长度的敏感度不同。GPT-4 对长 System Prompt 的容忍度较高,而一些开源模型(如 Llama 3 的早期版本)在 System Prompt 超过 800 token 后就会出现明显的指令遵循率下降。在做模型切换时,需要重新评估你的窗口管理策略。

五、总结

Token 预算管理不是"省着用"那么简单——它更像是一个优先级编排问题。核心公式是:

可用 Token = 模型上限 × (1 - 预留比例) - max_tokens

在这个预算内,你要为每种内容分配配额:核心层 ≤ 30%(1500 token 以内)、检索结果 ≤ 40%、对话历史 = 剩下的。当预算吃紧时,优先压缩对话历史(通过摘要),其次是截断检索结果(保留高分前几条),永远不要压缩核心层——因为那是 Agent 的"人格底线"。

记住一个原则:不要让 System Prompt 替你做检索应该做的事。如果你发现 System Prompt 里出现了"如果用户问 XXX,你就回答 YYY",那说明你需要的是一个更好的检索系统,而不是一个更长的 Prompt。

下一篇聊聊异步 RAG 流水线——怎么用 asyncio 把 embedding、检索、生成三步拆开并行跑。