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(contentcontent, tierContentTier.CORE, namename) 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(contentdescription, tierContentTier.TOOL, namename) 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[文档 {i1}]\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_tokens128000, max_completion_tokens4096) # 设置核心层尽量精简 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, fPython 中的 async/await 是怎么工作的轮次 {i}) manager.add_message(assistant, fasync/await 是协程的基础轮次 {i} 的回答...) # 构建上下文当前请求只需要 search_docs 工具 context manager.build_context( user_query用代码演示一下 asyncio.gather 的用法, retrieval_results[ asyncio.gather 用于并发执行多个协程。它会等待所有协程完成返回结果列表。, 使用 gather 时应注意异常处理。如果 return_exceptionsTrue异常会作为结果返回。, ], 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、检索、生成三步拆开并行跑。