openJiuwen agent-core 上下文引擎 CurrentRoundCompressor：当前轮工作压缩器的配置、原理与实战

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习

导读

本文讲解 openJiuwen agent-core 上下文引擎（Context Engine）中的当前轮压缩处理器CurrentRoundCompressor：当上下文 Token 数达到上下文容量的一定比例时，它调用 LLM 把"本轮内已完成的工作"（最后一次真实用户消息之后的推理、工具调用、工具结果等）压缩成一条<memory_block_current>摘要消息，而用户请求本身保持原样不动。读完本文，你将掌握该处理器的全部配置参数、触发判定逻辑、压缩产物结构、错误重试机制，并能直接写出可运行的最小接入示例。

一、它解决什么问题：长任务中的"本轮失控"

在工具型 Agent 的长任务中，上下文膨胀往往发生在当前这一轮：用户发出一条请求后，Agent 连续进行多轮"思考 → 工具调用 → 工具结果"的循环，例如反复读取文件、执行命令、查看测试输出。这些中间过程消息数量巨大，但其中大量信息（如大段文件内容、冗长的命令输出）在被消费过后已不再需要以原始形态占据上下文。

CurrentRoundCompressor的定位正是针对这一场景：

• 只压缩"本轮"产生的工作消息：即最后一次真实用户消息（last real user message）之后产生的所有推理、工具调用与工具结果；
• 用户请求永不压缩：最后一次真实用户消息及其之前的内容构成"保留前缀（preserved prefix）"，作为任务的意图锚点原样保留；
• 压缩产物是可恢复的检查点：摘要消息不是简单的"对话总结"，而是能让下一个模型实例继续执行当前任务的"执行状态快照"（execution state snapshot）。

从源码结构看，该处理器位于 forked/compressor/current_round_compressor.py，与DialogueCompressor（对话历史压缩）、RoundLevelCompressor（整轮级全上下文压缩）等同属压缩处理器家族，但分工不同：CurrentRoundCompressor只处理最新用户消息之后的活跃工作区。

二、配置类 CurrentRoundCompressorConfig：7 个参数全解

配置类CurrentRoundCompressorConfig继承自PrefixCompactProcessorConfig（定义见 forked/compressor/base.py），两者结合共提供以下核心参数：

参数约束与行为要点（源码 current_round_compressor.py 中的 pydantic 校验）：

• trigger_context_ratio必须严格大于 0、小于 1（gt=0.0, lt=1.0）；min_target_context_ratio允许 0、小于 1（ge=0.0, lt=1.0）；keep_recent_messages非负。
• model与model_client必须同时配置，否则处理器永远不会触发压缩——在 base.py 中，只有两者都非None时才会构造Model与CompressionExecutor，trigger_get_context_window首先就会因执行器缺失而直接返回False。
• 工具边界保护：keep_recent_messages指定的保留边界会被自动向后扩展，保证"工具调用（AssistantMessage.tool_calls）与它的工具结果（ToolMessage）"永不分离。这一逻辑由adjust_keep_recent_for_tool_boundaries实现（base.py）：它从尾部保护区内收集所有工具结果 ID，再向前扫描对应的工具调用消息，把起始边界移动到最早的关联调用之前。
• 基类还提供enable_kv_cache_affinity（默认False，由 ReActAgent 场景的 ContextProcessorRail 推导；直接使用处理器时需显式开启）与compression_recall_config（回忆归档配置，从ContextEngineConfig继承，旧版enable_recall等字段会被_reject_legacy_recall_config校验器拒绝并提示迁移）。

三、工作原理：三段式 Span 划分与触发判定

3.1 把消息流切成三段

CurrentRoundCompressor的核心是_build_span方法（current_round_compressor.py），它把整条上下文切成PrefixCompactSpan三段：

1. preserved_prefix（保留前缀）：[0, last_user_index + 1)，即最后一次真实用户消息及其之前的全部内容——永远不压缩；
2. messages_to_compress（待压缩区）：(last_user_index, split_index)，即本轮内用户消息之后、保留尾之前的活跃工作消息；
3. protected_tail（保护尾部）：末尾effective_keep条最近消息（经过工具边界修正）。

其中"真实用户消息"的判定在 base.py：只认UserMessage且内容不以内部前缀开头——内部前缀集合INTERNAL_USER_PREFIXES包括<system-reminder>、<memory_block_current>、<memory_block_dialogue>、<memory_block_round>、<memory_block_session>、<recovered_context>、[STATE_REINJECTION]等（见 support/util.py），这些"用户角色"消息实际承载的是内部状态，不能当作真实用户输入。

3.2 两段式触发判定

在上下文窗口物化（get_context_window）阶段，处理器按trigger → on两段式工作（接口定义见 forked/base.py）：

1. trigger_get_context_window（base.py）：
   • 计算整窗 Token 数（优先使用模型上报的 usage，否则用 token_counter 统计消息 + 工具）；
   • 若未达到context_max × trigger_context_ratio的绝对阈值 → 不触发；
   • 构建 span，若没有可压缩消息 → 不触发；
   • 计算待压缩区 Token 数，若低于context_max × min_target_context_ratio→ 跳过（记录target_below_min）；
   • 全部通过才返回True。
2. on_get_context_window（base.py）：真正执行压缩，成功后把context_window.context_messages与context内的消息替换为压缩后序列，并返回携带compact_summary的ContextEvent。

3.3 压缩产物：<memory_block_current>结构

压缩成功后，原消息被替换为一条UserMessage，内容包裹在<memory_block_current>...</memory_block_current>中（标记常量定义于 current_round_compressor.py，消息包装逻辑见 base.py）：

<memory_block_current> <meaning> This is a compressed summary of work already performed after the latest user request. ... </meaning> <conflict_policy> Newer raw messages, fresh tool results, and the latest user instructions override this summary. </conflict_policy> <summary> ...（LLM 生成的执行状态快照）... </summary> </memory_block_current>

处理器类属性还定义了压缩摘要的语义契约（current_round_compressor.py）：摘要不是新的用户请求，而是"已完成的推理、工具调用、代码变更、测试结果与后续步骤"的恢复依据；较新的原始消息、新鲜的工具结果与最新的用户指令优先于摘要。reinject_builder_names = ["plan_mode", "plan", "task_status", "todo", "skills", "read_file"]表明压缩后还会尝试把计划模式、计划、任务状态、待办、技能与文件读取等关键状态以"状态重注入消息"的形式补回上下文（_build_compacted_messages，base.py）。

3.4 压缩提示词：面向"可恢复性"而非"省 Token"

处理器使用内置提示词CURRENT_COMPACT_PROMPT（定义见 prompts/prompts.py）。该提示词的核心设计值得关注：

• 输出硬约束：纯文本、禁止任何工具调用、输出必须恰好包含<coverage_check>与<state_snapshot>两个 XML 块；
• 角色定位："执行状态压缩助手（Execution State Compression Assistant）"，目标是让另一个模型实例能以最小的重复劳动继续当前工作，而不是"总结对话"；
• 不追求最省 Token，而是追求执行连续性（maximize execution continuity within the available token budget）；
• <state_snapshot>内部有 11 个固定小节：Active Task、Constraints and Preferences、Completed Work in This Active Segment、Current Work and Active State、Immediate Resume Point、Pending Tasks and Next Useful Step、Key Facts/Decisions/Evidence/Fixes、Files/Code Areas/Artifacts、Blockers/Risks/Verification、Critical Context、Relevant Files & Structure；
• 安全要求：删除敏感信息（API Key、Token、口令等）。

压缩响应还会经过_extract_state_snapshot_or_raw解析（base.py）：若响应内容包含<state_snapshot>...</state_snapshot>块则提取其内容作为摘要，否则使用完整原始文本。

四、完整可运行示例（继承原文档并补充注释）

以下示例完整继承自关联文档，并补充了关键注释。它演示了：如何构建模型配置与处理器配置、如何通过forked.activate()注册处理器、如何用名称引用处理器并创建上下文。

>>> import os >>> import asyncio >>> from openjiuwen.core.context_engine import ContextEngine, ContextEngineConfig >>> from openjiuwen.core.context_engine.processor import forked >>> from openjiuwen.core.context_engine.processor.forked.compressor.current_round_compressor import ( ... CurrentRoundCompressor, ... CurrentRoundCompressorConfig, ... ) >>> from openjiuwen.core.foundation.llm import ( ... UserMessage, ... AssistantMessage, ... ToolMessage, ... ModelRequestConfig, ... ModelClientConfig, ... ) >>> >>> # 1. 通过环境变量准备模型服务参数（按需替换为真实值） >>> API_BASE = os.getenv("API_BASE", "your api base") >>> API_KEY = os.getenv("API_KEY", "your api key") >>> MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini") >>> MODEL_PROVIDER = os.getenv("MODEL_PROVIDER", "OpenAI") >>> >>> async def main(): ... # 2. 配置压缩用的模型：model 与 model_client 必须同时配置，否则永不触发 ... model_config = ModelRequestConfig(model=MODEL_NAME) ... model_client_config = ModelClientConfig( ... client_provider=MODEL_PROVIDER, ... api_base=API_BASE, ... api_key=API_KEY, ... ) ... # 3. 配置当前轮压缩器：上下文 80% 时触发，末尾保留 2 条最近消息 ... compressor_config = CurrentRoundCompressorConfig( ... trigger_context_ratio=0.8, ... keep_recent_messages=2, ... model=model_config, ... model_client=model_client_config, ... ) ... forked.activate() # 注册处理器，使其可以被按名称引用 ... engine_config = ContextEngineConfig(default_window_message_num=100) ... engine = ContextEngine(engine_config) ... ctx = await engine.create_context( ... "demo_ctx", ... None, ... history_messages=[], ... processors=[("CurrentRoundCompressor", compressor_config)], ... ) ... # 4. 注入一条用户消息 + 一轮"工具调用 → 工具结果"的活跃工作 ... await ctx.add_messages([ ... UserMessage(content="Help me fix this error"), ... AssistantMessage(content="", tool_calls=[{"id": "1", "name": "read_file", "type": "function", "arguments": "{}"}]), ... ToolMessage(content="file content ...", tool_call_id="1"), ... AssistantMessage(content="Located the problem, fixing it now."), ... ]) ... # 5. 尚未达到 trigger_context_ratio（0.8），因此不触发压缩，返回原始消息数 ... return len(ctx.get_messages()) >>> >>> asyncio.run(main()) 4

示例输出说明：上面输出4是压缩未触发时的原始消息数。当上下文达到触发阈值后：

• 用户请求保持完整（1 条）；
• 除末尾keep_recent_messages（示例中为 2）之外的本轮消息被替换为 1 条<memory_block_current>摘要；
• 因此get_messages()变为"1 条用户消息 + 1 条摘要 + keep_recent_messages 条最近消息"。

五、源码级纵深：错误分类、重试与观测

5.1 压缩调用的错误分类与重试

压缩模型调用由CompressionExecutor封装（compression_executor.py），异常通过classify_compression_error归入 8 类CompressionErrorKind：context_overflow、rate_limit、authentication、timeout、server_unstable、api_request_error、unknown等（依据错误文本关键词与 HTTP 状态码识别，如 413 判为上下文溢出、429 判为限流、401/403 判为鉴权失败、500/502/503/504 判为服务不稳定）。

_invoke_compression_with_retries（base.py）实现了两套恢复策略：

• 上下文溢出（CONTEXT_OVERFLOW）预算收缩重试：当压缩请求自身超出模型上下文时，依次尝试(0.85, 0.65, 0.5)三档预算比例（_CONTEXT_OVERFLOW_RETRY_BUDGET_RATIOS），把更多待压缩消息移入保护尾部、缩小送入压缩模型的窗口后重试；预算耗尽则放弃本次压缩，保留原上下文；
• 瞬时错误（限流/超时/服务不稳定/未知）退避重试：最多重试 2 次，退避延迟按0.05s × 2^(n-1)指数增长（_TRANSIENT_COMPRESSION_MAX_RETRIES = 2、_TRANSIENT_COMPRESSION_RETRY_BASE_DELAY_SECONDS = 0.05）。

压缩失败不会破坏主 Agent 上下文——on_get_context_window返回None时原窗口保持不变，这正是"压缩是尽力而为的优化"这一设计原则的体现。

5.2 压缩收益校验与回滚

即使压缩成功，_has_compression_benefit（base.py）还会比较压缩前后 Token 数：只有new_tokens < original_tokens时才真正替换上下文；否则放弃新结果，若已写入回忆归档还会触发归档回滚（_delete_compression_archive）。这防止了"压缩后反而更大"的负收益情况。

5.3 可观测性：压缩落盘（Compression Dump）

开启enable_compression_dump=True后，每次真实压缩调用会把完整工件落盘（_dump_compression_artifact，base.py），包含：发给压缩模型的prompt与request、原始消息、三段 span、压缩摘要、压缩后消息、usage 统计等，可用于离线分析压缩效果。落盘实现采用延迟导入避免循环依赖（见 compression_dump.py）。同时，trigger_get_context_window与on_get_context_window全程通过_write_context_debug记录threshold_check、span_built、compression_retry等调试事件，便于定位"为什么没压缩"。

5.4 与 ReActAgent 的模型切换协同

基类提供rebind_model（base.py）：当主 Agent 在运行时切换模型时，可把压缩执行器重绑到当前活跃模型上，避免压缩器一直使用过时配置。该机制从源码看是为 ReActAgent 动态换模型场景设计的扩展点。

六、测试验证与接入建议

6.1 测试覆盖

单元测试见 tests/unit_tests/core/context_engine/test_current_round_compressor.py，覆盖了关键行为：

• 用 mock 模型响应验证"大消息超过阈值后触发压缩"；
• 通过ContextEngine.create_context以processors=[("CurrentRoundCompressor", compressor_config)]方式注册处理器；
• 使用长度型 token counter 模拟 Token 计数，验证触发判定逻辑；
• 验证压缩目标是"最后一条 UserMessage 之后的 Assistant/Tool 消息"这一核心语义。

6.2 接入与调参建议

• 首次接入：先保持trigger_context_ratio=0.8与keep_recent_messages=0默认值，用示例中的最小代码验证管道连通；
• 降低触发门槛：若任务中工具结果普遍较长、希望更早压缩，可将trigger_context_ratio下调（例如0.6）；注意该值必须大于 0 小于 1；
• 保护最近信息：若希望最近几条助手回复/工具结果不被压缩，设置keep_recent_messages=2~4，工具边界会自动保护成对消息；
• 防止无意义压缩：min_target_context_ratio默认0.1，可压内容太少时会自动跳过；
• 压缩前自查：确认model与model_client均正确配置——这是最容易被忽略的"永不触发"原因；
• 生产排障：开启enable_compression_dump=True并设置compression_dump_dir，结合调试日志分析每次压缩的收益与失败原因；
• 注意容量来源：触发阈值基于"上下文容量（context_max）"计算，该容量由resolve_context_max根据模型与上下文配置解析（base.py），实际部署时应确保上下文容量配置与所选模型一致。

关联阅读：ContextProcessor 基类说明、同族的 DialogueCompressor 与 RoundLevelCompressor 可构成"当前轮 + 历史对话 + 整轮"的完整压缩策略组合。

• 人工智能
• AI Agent
• Agent 框架
• 大模型
• 工具调用
• RAG
• 提示工程
• 强化学习