openJiuwen context_engine 上下文引擎完全指南：ContextEngine、ModelContext 与处理器链的配置与实战

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

openjiuwen.core.context_engine是 openJiuwen 中负责Agent 运行期模型输入上下文构建、存取与容量管理的核心模块：它以ContextEngine为入口，按「会话 + 上下文 ID」管理多个ModelContext，在消息写入与窗口物化两个生命周期通过可插拔的处理器链完成消息卸载（Offload）与对话压缩（Compression），并提供 Token 统计、窗口裁剪、压缩原文召回（Compression Recall）与 Session Memory 物化等能力。读完本文，你将掌握上下文引擎的核心概念与配置项、ModelContext的增删查改接口、内置处理器链的完整参数体系，以及如何在 Agent 与 Workflow 场景中落地长上下文管理方案。

模块概览：上下文引擎解决了什么问题

上下文引擎（ContextEngine）是 openJiuwen 中用于模型输入上下文构建与复用的统一组件。它属于Agent 层面的成员，由 Agent 在初始化时创建并持有，围绕Session与ModelContext工作，主要承担三件事：

• 历史对话消息管理：短期记忆维护与窗口裁剪（按消息数或对话轮数）；
• Token 统计与窗口控制：构造每次送入模型推理的上下文窗口，并统计各类消息的 Token 消耗；
• 多上下文并存：支持一个会话内同时跑多个子任务/子对话，按context_id隔离。

其公开类与职责见下表（详细 API 见 模块索引）：

处理器体系另见 processor 模块索引。

三个核心概念：ContextEngine / ModelContext / ContextWindow

• ContextEngine：入口类openjiuwen.core.context_engine.ContextEngine，作为 Agent 成员在初始化时创建，负责按session_id + context_id管理多个ModelContext实例，提供create_context / get_context / clear_context / save_contexts能力。
• ModelContext：代表某个会话中的一个具体「对话上下文」，维护消息列表、负责构造窗口（get_context_window()）。
• ContextWindow：get_context_window()的返回结果，是一次「即将送入模型」的窗口快照，封装三类内容：system_messages（系统指令）、context_messages（对话历史）、tools（工具定义）。

ContextEngineConfig 完整配置说明

ContextEngineConfig（config.md）是上下文引擎的完整配置类，核心字段如下：

几点关键语义（可从 源码 印证）：

• 窗口截断规则：default_window_message_num与default_window_round_num均为None时，get_context_window()不做任何窗口截断，返回完整对话历史（仅受max_context_message_num缓冲区上限与处理器卸载/压缩影响）。
• 窗口来源优先级：显式的context_window_tokens与model_context_window_tokens优先于远端 OpenRouter 元数据。
• 模型窗口元数据兜底：源码中还提供model_context_window_tokens_override字段，其优先级低于全局context_window_tokens，用于在模型名未命中映射时兜底。

CompressionRecallConfig：压缩原文归档与召回

启用后，压缩器在压缩前会把被替换的原始消息归档，Harness 同时注册recall_compressed_context工具，模型可按需找回被压缩掉的内容：

配置样例

from openjiuwen.core.context_engine import ( CompressionRecallConfig, ContextEngineConfig, ) config = ContextEngineConfig( default_window_round_num=10, context_window_tokens=128_000, compression_recall_config=CompressionRecallConfig( enabled=True, chunk_size_tokens=3000, chunk_overlap_tokens=300, ), )

注意：KV Cache 亲和不属于ContextEngineConfig，需要通过 Agent 侧配置开启：ReActAgentConfig.configure_kv_cache_affinity(enable_kv_cache_affinity=True)。

ContextEngine 入口 API

ContextEngine(config: ContextEngineConfig = None)（context_engine.md）是对话上下文管理的入口，负责创建、缓存和持久化ModelContext实例，并支持通过Processor进行消息压缩、offload 等处理。源码位于 context_engine.py，核心方法如下：

create_context：创建或获取 ModelContext

async create_context( context_id: str = "default_context_id", session: Session = None, *, processors: List[Tuple[str, BaseModel]] = None, history_messages: List[BaseMessage] = None, token_counter: TokenCounter = None, ) -> ModelContext:

• context_id：会话内当前上下文的唯一标识，用于区分同一会话内的多个上下文；
• session：会话对象；为None时使用"default_session_id"；
• processors：处理器配置列表，每项为(processor_type, config)，如[("MessageOffloader", MessageOffloaderConfig(...))]；
• history_messages：上下文的初始消息列表；
• token_counter：Token 统计策略；显式传入时始终使用该策略；为None时仅当ContextEngineConfig.enable_tiktoken_counter=True才使用TiktokenCounter，否则使用内置字符数估算。

若池中已存在同(session_id, context_id)的上下文，则加载 session 状态后返回；否则新建并缓存。

get_context / clear_context / save_contexts

get_context(context_id: str = "default_context_id", session_id: str = "default_session_id") -> Optional[ModelContext]

从内部缓存获取已存在的ModelContext，不存在返回None。

clear_context(context_id: str = None, session_id: str = None)

从内部缓存移除上下文：两参数均为None时清空全部；仅提供session_id时删除该 session 下所有 context；同时提供两者时删除指定 context；找不到目标时记录警告日志（提供context_id时必须同时提供session_id）。

async save_contexts(session: Session, context_ids: list[str])

批量保存多个上下文的运行时状态（消息、滑动窗口位置、token 计数与统计信息）；context_ids为None时持久化该 session 下全部 context。

register_processor：注册自定义处理器

@ContextEngine.register_processor()是类方法装饰器，用于将新的上下文处理器注册到引擎的处理器映射表，引擎运行时通过processor_type()查找并实例化：

from openjiuwen.core.context_engine import ContextEngine from openjiuwen.core.context_engine.processor.base import ContextProcessor from pydantic import BaseModel, Field class MyProcessorConfig(BaseModel): threshold: int = Field(default=5, gt=0) @ContextEngine.register_processor() class MyProcessor(ContextProcessor): def __init__(self, config: MyProcessorConfig): super().__init__(config) async def trigger_add_messages(self, context, messages_to_add, **kwargs): return False async def on_add_messages(self, context, messages_to_add, **kwargs): return None, messages_to_add def load_state(self, state): pass def save_state(self): return {}

注册后即可在create_context时传入("MyProcessor", MyProcessorConfig(...))使用。

ModelContext 接口详解

ModelContext（base.md）定义消息增删、窗口构建、统计等接口，支持BaseMessage类型的内容格式。其核心方法均为抽象方法，使用时通过engine.create_context()获得实现类实例。

消息增删查改

• add_message / add_messages：abstract add_message(message: BaseMessage | List[BaseMessage]) -> List[BaseMessage]，向模型上下文中添加一条或多条消息，返回追加后（Processor 处理后的）消息列表。
• get_messages：get_messages(size: Optional[int] = None, with_history: bool = True) -> List[BaseMessage]，获取消息（不执行移除）；size=None表示获取全部，with_history=False仅返回最新消息。
• set_message：set_message(messages: List[BaseMessage], with_history: bool = True)，替换消息列表；with_history=True替换「上下文消息 + 历史消息」拼接后的集合，False仅替换历史消息。
• pop_message：pop_message(size: int = 1, with_history: bool = True) -> List[BaseMessage]，移除并返回最新size条消息；with_history=True会同时从底层持久化历史中移除。
• clear_message：clear_message(with_history: bool = True)，移除当前对话轮次中添加的所有消息；with_history=True会同时清空底层持久化历史。

get_context_window：构造模型推理窗口

abstract get_context_window( system_messages: List[BaseMessage] = None, tools: List[ToolInfo] = None, window_size: Optional[int] = None, dialogue_round: Optional[int] = None, **kwargs ) -> "ContextWindow"

• system_messages：前置到消息窗口的系统级消息；
• tools：纳入窗口的工具定义信息；
• window_size：要包含的历史消息最大数量，省略则包含全部历史；
• dialogue_round：要保留的最新对话轮次数。一个轮次的定义：从用户消息开始，至下一条不含工具调用的助手消息结束（即工具调用轮次中的最终回复）；未完成轮次（仅有用户消息、无后续助手消息）仍计为一个轮次。该参数与window_size同时指定时优先级更高；显式设置时值必须大于0。

统计与标识

• statistic()：返回ContextStats，包含消息与 token 数量聚合统计。
• session_id() / context_id()：返回会话级与上下文级全局唯一标识。
• token_counter()：返回TokenCounter实例。
• reloader_tool()：返回上下文原始消息重载工具。当配置项enable_reload打开时，可将工具注册到 Agent，帮助模型在需要详细原文时重载被处理器处理后的消息原文。工具名默认为reload_original_context_messages；offload_handle指向卸载内容的 UUID 或文件路径，offload_type为存储后端类型（当前支持"in_memory"内存存储）。

ContextStats 与 ContextWindow

ContextStats表示任意上下文或上下文窗口的总消息量、总 token 以及各类消息的数量与 token 使用情况，字段包括：total_messages、total_tokens、system_messages、user_messages、assistant_messages、tool_messages、tools，以及对应的*_message_tokens、tool_tokens。

ContextWindow表示即将发送给 LLM 的上下文窗口，字段为system_messages、context_messages、tools、statistic；get_messages()将系统指令与对话历史拼接返回，get_tools()返回本轮可调用的工具列表。

完整可运行示例

import asyncio from openjiuwen.core.context_engine import ContextEngine, ContextEngineConfig from openjiuwen.core.foundation.llm import UserMessage async def main(): engine = ContextEngine(ContextEngineConfig()) ctx = await engine.create_context(session=None, history_messages=[]) added = await ctx.add_messages(UserMessage(content="你好")) return len(added), ctx.get_messages()[0].content asyncio.run(main()) # (1, '你好')

上下文处理器（Processor）体系

处理器可在两个生命周期介入，每点由一对协程构成（先trigger_*判断是否介入，再on_*执行介入）：

基类ContextProcessor（base.md）构造时接收一个 Pydantic 配置，还提供load_state / save_state（跨会话检查点恢复）、offload_messages（将消息转存到内存或文件系统）、processor_type()（返回类名，用于注册与查找）与只读config属性。两个核心钩子的签名：

async on_add_messages(context, messages_to_add, **kwargs) -> Tuple[ContextEvent | None, List[BaseMessage]] # 返回空列表可取消本次追加；消息传给下一处理器 async on_get_context_window(context, context_window, **kwargs) -> Tuple[ContextEvent | None, ContextWindow] # 对窗口进行压缩/重排/替换；禁止返回 None

默认trigger_*恒返回False表示不介入。处理器实现源码位于 openjiuwen/core/context_engine/processor/，其中compressor/与offloader/两个子目录分别存放压缩器与卸载器。

默认处理器链（Harness preset）

Harness 通过ContextProcessorRail(preset=True)（context_processor_rail.py）组装默认上下文处理器链，按固定顺序执行：

1. MessageSummaryOffloader
2. SessionMemoryCompressor（默认关闭）
3. ReasoningToolLoopCompactProcessor
4. DialogueCompressor
5. CurrentRoundCompressor
6. RoundLevelCompressor

默认处理链开箱即用地启用上下文卸载与压缩能力；Session Memory 与压缩原文召回保持关闭，不会产生对应的后台模型调用或归档开销。

卸载器（Offloader）：不调用 LLM 的容量控制

MessageOffloader

MessageOffloaderConfig（message_offloader.md）在消息数量或 token 数超过阈值时对符合条件的大消息裁剪并 offload：

约束：trim_size必须小于large_message_threshold；若同时设置messages_to_keep与messages_threshold，则messages_to_keep必须小于messages_threshold。offload 后的消息内容含[[OFFLOAD: ...]]占位符（见 示例）。

MessageSummaryOffloader

MessageSummaryOffloader是默认处理链使用的卸载器，不依赖 LLM：当 tool 消息过大时，先按内置规则压缩，再将原始内容 offload 到文件系统，上下文中仅保留头尾预览与[[OFFLOAD: ...]]占位符。其MessageSummaryOffloaderConfig字段如下：

约束：offload 文件默认写入{workspace}/context/{session_id}_context/offload/；上下文未关联 workspace 时 offload 不会生效，消息保持原样。

迁移注意：旧模块路径openjiuwen.core.context_engine.processor.offloader.message_summary_offloader下的同名类仍然存在，但配置字段与行为不同（旧版经 LLM 生成摘要后 offload，使用large_message_threshold、offload_message_type等字段），两者不能混用。迁移时请按新模块路径openjiuwen.core.context_engine.processor.forked.offloader.message_offloader导入并按新字段配置。

压缩器（Compressor）：经 LLM 将历史压缩为摘要

三个主压缩器均需同时配置model（ModelRequestConfig）与model_client（ModelClientConfig），否则处理器永远不会触发。保留消息边界会自动向前扩展，保证 tool call 与其 tool 结果不被拆散。

DialogueCompressor：对话历史压缩器

当上下文 token 数达到上下文容量的一定比例时，将「当前对话轮」（最后一条真实用户消息）之前的全部历史消息经 LLM 压缩为一条<memory_block_dialogue>摘要消息，当前轮消息完整保留，摘要中附带含义说明与冲突处理策略（更新的原始消息与最新用户意图优先于摘要）。

迁移注意：旧版DialogueCompressor按完整 tool-call 对话轮逐轮压缩，配置字段为messages_threshold、tokens_threshold、messages_to_keep、keep_last_round等；当前实现改为将「当前轮之前的全部历史」压缩为一条摘要，触发条件改为上下文容量比例，配置字段不兼容。

CurrentRoundCompressor：当前轮压缩器

将「当前轮」内已经完成的工作（最后一条真实用户消息之后产生的推理、工具调用与工具结果等消息）经 LLM 压缩为一条<memory_block_current>摘要消息，用于恢复已完成的分析、工具调用、代码改动、测试结果与后续步骤；用户请求本身保留不动。配置字段与DialogueCompressor相同，另有：

RoundLevelCompressor：轮级压缩器

仅保留最近keep_recent_messages条消息，将更早的全部历史消息经 LLM 压缩为一条<memory_block_round>跨轮检查点摘要（可包含长期项目状态、历史决策、当前目标、已完成与未完成的工作等），并附带含义说明与冲突处理策略。与CurrentRoundCompressor相比，默认keep_recent_messages=4，是面向「跨轮长期任务」的兜底压缩。

压缩原文召回与 Session Memory：两个可选进阶能力

启用压缩原文召回（Compression Recall）

将ContextEngineConfig(compression_recall_config=CompressionRecallConfig(enabled=True))传给DeepAgentConfig即可。启用后：

• 默认处理链中的压缩器（DialogueCompressor/CurrentRoundCompressor/RoundLevelCompressor）在压缩前会把被替换的原始消息归档到 workspace 会话上下文目录下的compression_recall/中，摘要里附带[[COMPRESSION_RECALL: id=...]]标记；
• Harness 自动注册recall_compressed_context工具，模型可凭标记中的memory_id按需找回被压缩掉的内容。

该开关默认关闭，与 Session Memory 相互独立，可单独或同时启用。

启用 Session Memory

SessionMemoryCompressor将会话级长期笔记（session memory）物化为一条<memory_block_session>摘要消息，替换笔记已覆盖的历史消息前缀。该处理器默认关闭（enabled=False），需要通过配置覆盖显式启用：

from openjiuwen.core.context_engine.context.session_memory_manager import ( SessionMemoryConfig, ) from openjiuwen.core.context_engine.processor.forked.compressor.session_memory_compressor import ( SessionMemoryCompressorConfig, ) from openjiuwen.harness.rails.context_engineer.context_processor_rail import ( ContextProcessorRail, ) context_rail = ContextProcessorRail( preset=True, processors=[ ( "SessionMemoryCompressor", SessionMemoryCompressorConfig( enabled=True, memory=SessionMemoryConfig( update_trigger_context_ratio=0.7, ), ), ), ], )

SessionMemoryCompressorConfig关键字段：enabled（默认False）、trigger_context_ratio（默认0.8，上下文 token 占容量达到该比例才允许物化）、session_memory_path（默认{workspace}/context/{session_id}_context/session_memory/session_context.md）、max_notes_chars（默认120000）、memory（SessionMemoryConfig）。

配套的SessionMemoryConfig（后台笔记更新任务SessionMemoryManager，源码见 session_memory_manager.py）：

物化只替换笔记已覆盖的消息前缀，未覆盖的消息原样保留。

在 Agent 与 Workflow 中落地上下文引擎

Agent 使用上下文：典型调用流程

ContextEngine 是 Agent 的成员，在 Agent 初始化时创建。Agent 在invoke中通过context_engine管理上下文：

1. Agent 初始化：创建并持有context_engine实例（如ContextEngine(ContextEngineConfig())）；
2. invoke 开始：调用context_engine.create_context(session=agent_session)建立当前会话的上下文环境；
3. invoke 执行：将请求委托给 Controller（如 ReActController、WorkflowController），Controller 内部按需使用 context；
4. invoke 结束：调用context_engine.save_contexts(session)持久化上下文，并在适当时机clear_context释放资源。

# Agent 初始化时（以 BaseAgent / ReActAgent 等为例） from openjiuwen.core.context_engine import ContextEngine, ContextEngineConfig class MyAgent(BaseAgent): def __init__(self, config): super().__init__(config) self.context_engine = ContextEngine(config=ContextEngineConfig( default_window_message_num=100, )) # Agent invoke 中（示意） async def invoke(self, inputs: Dict, session: Session = None) -> Dict: agent_session = session or await self._session.pre_run(session_id=...) await self.context_engine.create_context(session=agent_session) try: result = await self.controller.invoke(inputs, agent_session) await self.context_engine.save_contexts(agent_session) return result finally: await agent_session.post_run()

要点：ContextEngine 由 Agent 创建并持有，不推荐在工作流组件或外部单独创建全局实例；create_context、save_contexts在 Agent 的invoke入口统一调用，确保会话隔离与生命周期一致。

Workflow 使用上下文

Workflow 执行发生在Agent 的 invoke 内部，由 Controller（如 WorkflowController、LLMController）调度。Workflow 使用的ModelContext由Agent 的 context_engine创建，并通过Runner.run_workflow透传，而不是由 Workflow 组件自行创建：

from openjiuwen.core.runner import Runner # Controller 中：通过 Agent 的 context_engine 创建 context 并透传 result = await Runner.run_workflow( workflow, inputs=task.input.arguments, session=workflow_session, context=await self._context_engine.create_context( context_id=workflow_id, session=session, ), )

Workflow 组件从invoke参数中接收 context 直接使用，不应自行调用context_engine.create_context：

from openjiuwen.core.context_engine import ModelContext from openjiuwen.core.graph.executable import Input, Output from openjiuwen.core.foundation.llm import UserMessage class ChatNode(WorkflowComponent): async def invoke(self, inputs: Input, session, context: ModelContext) -> Output: user_query = inputs.get("query", "") await context.add_messages(UserMessage(content=user_query)) window = await context.get_context_window() messages = window.get_messages() # llm_response = await model.ainvoke(messages=messages, ...) return {"answer": "..."}

清理与生命周期管理

在长时间运行的服务中，建议：

• clear_context(session_id=...)：清理某个会话下所有上下文（例如用户主动结束会话时）；
• clear_context(session_id=..., context_id=...)：只清理某个子对话；
• 在 Runner / 服务关闭前，先save_contexts再clear_context，将必要信息落盘后释放内存。

context_engine.clear_context(session_id="s1") # 用户结束会话 context_engine.clear_context(session_id="s1", context_id="chat") # 仅结束子对话

调试与排障

• 上下文处理器统一调试：开启ContextEngineConfig(enable_context_debug=True)后，各处理器在阈值检查、分段、压缩重试、前后对比等流水线阶段落盘 JSONL 调试记录；目录通过context_debug_dir指定，或回退到环境变量OPENJIUWEN_CONTEXT_DEBUG_DIR与默认目录{workspace}/context/{session_id}_context/context_debug/。
• 压缩落盘：三个压缩器均支持enable_compression_dump=True，将每次真实压缩的请求与压缩后上下文落盘，用于离线效果分析。
• 卸载调试：MessageSummaryOffloaderConfig.enable_debug_dump控制规则压缩与 offload 的调试记录落盘。
• 处理器不触发排查：压缩器必须同时配置model与model_client；卸载器须确认上下文已关联 workspace，否则 offload 不生效。

小结

• ContextEngine 是 Agent 层面的成员，由 Agent 初始化时创建并持有，按session_id + context_id管理多个ModelContext。
• 上下文处理器在消息加入或窗口物化时介入：默认链为MessageSummaryOffloader→SessionMemoryCompressor（默认关闭）→ReasoningToolLoopCompactProcessor→DialogueCompressor→CurrentRoundCompressor→RoundLevelCompressor，并可通过ContextEngine.register_processor注册自定义处理器。
• Agent 使用上下文：在invoke中调用create_context、save_contexts，委托 Controller 执行。
• Workflow 使用上下文：由 Controller 通过 Agent 的context_engine.create_context创建，透传给Runner.run_workflow，Workflow 组件从参数接收 context，不应自行创建。
• 通过create_context / get_context / save_contexts / clear_context完成多会话隔离、多上下文 ID 管理及 Token 级别的窗口控制；可选的压缩原文召回与 Session Memory 分别解决「压缩后信息找回」与「跨轮长期记忆」两个进阶诉求。

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