ContextEngine 可插拔上下文处理管道

ContextEngine 可插拔上下文处理管道

1. 问题定义

在 Agent 调用 LLM 之前，消息列表（messages）需要经过一系列预处理步骤：截断过长的历史、处理工具调用顺序、替换占位符变量、清理过期消息等。如果将这些逻辑写在一个函数中，会导致代码难以维护和测试。可插拔上下文处理管道将每个预处理步骤封装为独立的处理器，按顺序串联执行。

2. 核心架构

PipelineContext

管道执行过程中的中间状态，包含：

// 伪代码：管道上下文
interface PipelineContext {
  messages: ChatMessage[];        // 当前消息列表（每个处理器可修改）
  initialState: { messages: ChatMessage[] }; // 原始快照，用于回滚
  metadata: Record<string, any>;  // 处理器之间的共享状态
  isAborted: boolean;             // 中断信号
  abortReason?: string;           // 中断原因
}

Processor 接口

每个处理器实现统一的接口：

// 伪代码：处理器接口
interface ContextProcessor {
  name: string;  // 唯一标识，用于日志和错误诊断
  process(context: PipelineContext): Promise<PipelineContext>;
}

关键设计：处理器接收 PipelineContext 并返回新的（或修改后的）PipelineContext。这使得处理器之间完全解耦，可以任意组合。

3. 典型处理器

HistoryTruncateProcessor

当消息 Token 数超过限制时截断历史。与压缩（summarization）不同，截断是直接删除消息而非生成摘要。

策略：

• 保留最近的 N 条消息
• 删除最早的消息直到 Token 数在限制内
• 始终保留系统消息（system role）

陷阱：截断后丢失的对话上下文不可恢复。通常截断是管道中的最后一个降级手段——先尝试压缩，再截断。

MessageContentProcessor

处理消息内容的标准化：

• 将多模态内容（图片、文件）转换为 LLM 可接受的格式
• 处理 content 的 string vs array 类型差异
• 过滤不支持的内容类型

ToolCallProcessor

处理工具调用相关消息：

• 将工具调用和工具结果配对
• 过滤已取消的工具调用
• 处理工具调用失败的情况

ToolMessageReorder

重新排序工具消息以确保正确的消息顺序。某些 LLM 要求 tool 消息必须紧跟在 assistant 的 tool_calls 之后。

InputTemplateProcessor

在消息模板中替换占位符变量（如 {{username}}、{{date}}），支持条件渲染和默认值。

PlaceholderVariablesProcessor

将系统提示中的变量占位符替换为实际值。与 InputTemplateProcessor 的区别在于作用于系统提示而非用户消息。

4. 管道编排逻辑

// 伪代码：管道执行
async process(input: { messages: ChatMessage[] }): Promise<PipelineResult> {
  let context: PipelineContext = {
    messages: [...input.messages],
    initialState: input,
    metadata: {},
    isAborted: false,
  };

  for (const processor of this.processors) {
    if (context.isAborted) break;  // 中断信号检查

    const start = Date.now();
    try {
      context = await processor.process(context);
    } catch (error) {
      // 错误隔离：一个处理器失败不影响其他处理器
      throw new PipelineError(processor.name, error);
    }
    this.metrics[processor.name] = Date.now() - start;
  }

  return { messages: context.messages, metadata: context.metadata };
}

关键特性：

1. 顺序执行：处理器按注册顺序执行，前一个的输出是后一个的输入
2. 中断信号：任何处理器可以设置 isAborted = true 来终止管道
3. 错误隔离：单个处理器失败抛出 PipelineError，包含失败的处理器名称
4. 性能监控：每个处理器的执行时间被记录

5. 管道配置示例

// 伪代码：典型管道配置
const contextEngine = new ContextEngine({
  pipeline: [
    new PlaceholderVariablesProcessor(),      // 1. 替换系统提示变量
    new InputTemplateProcessor(),             // 2. 替换用户消息模板
    new GroupMessageFlattenProcessor(),       // 3. 展平消息组
    new MessageContentProcessor(),            // 4. 标准化内容格式
    new ToolCallProcessor(),                  // 5. 处理工具调用
    new ToolMessageReorder(),                 // 6. 重排序工具消息
    new MessageCleanupProcessor(),            // 7. 清理过期消息
    new HistoryTruncateProcessor(),           // 8. 最后手段：截断历史
  ],
  maxTokens: 128000,
});

处理器顺序的重要性：

• 变量替换必须在内容标准化之前（否则占位符可能已被转换）
• 工具消息重排序必须在工具调用处理之后（需要先配对再排序）
• 截断必须是最后一个处理器（所有其他处理后 Token 仍超限才截断）

6. 陷阱与对策

陷阱	后果	对策
处理器之间有隐式依赖	移除一个处理器导致后续失败	文档化每个处理器的输入/输出契约
处理器修改 messages 引用	后续处理器收到意外的对象	使用结构化克隆或不可变更新
管道配置过长	难以理解执行流程	将相关处理器分组为子管道
处理器异步操作未 await	消息状态不一致	使用 TypeScript strict 模式 + ESLint 检测未 await 的 Promise

7. 与其他模式的对比

模式	优势	劣势	适用场景
管道模式	可组合、可测试、可扩展	顺序执行、需要理解依赖顺序	需要多个独立预处理步骤
单一处理函数	简单直接、无顺序问题	难以维护、难以测试	只有 1-2 个简单预处理步骤
规则引擎	声明式、可动态配置	复杂规则表达困难	预处理规则需要运行时动态修改

8. 参考来源

• LobeHub ContextEngine — packages/context-engine/src/pipeline.ts 完整实现
• LobeHub Processors — packages/context-engine/src/processors/ 全部处理器实现
• Chain-of-Thought Prompting — 上下文处理中的思维链保留策略

---

补充：Agno 的 ContextProvider 协议与三种暴露模式

来源：Agno libs/agno/agno/context/provider.py，2026-05-30

统一的 ContextProvider 抽象

Agno 将所有外部上下文源统一为 ContextProvider 抽象基类：

# 伪代码：ContextProvider 协议
class ContextProvider(ABC):
    @abstractmethod
    def query(self, question: str) -> Answer: ...
    @abstractmethod
    async def aquery(self, question: str) -> Answer: ...
    @abstractmethod
    def status(self) -> Status: ...
    @abstractmethod
    async def astatus(self) -> Status: ...
    # 可选：写入支持
    async def aupdate(self, document: Document) -> None:
        raise NotImplementedError(f"{self.id} is read-only")

关键设计：Answer 返回结构化的 results（Document 列表）和 text（自然语言摘要）。查询方可以选择使用结构化结果或直接使用文本。

三种暴露模式

ContextMode 控制 Provider 如何暴露给调用 Agent：

模式	行为	适用场景
default	Provider 自己决定最佳暴露方式	通用场景
agent	包装为子 Agent，调用方获得单一 query_<id> 工具	Agent 需要自然语言查询能力
tools	直接暴露 Provider 的底层工具	调用方需要精细控制

mode=agent 的包装逻辑：

# 伪代码：将 Provider 包装为子 Agent
sub_agent = Agent(
    name=f"{provider.id}_context",
    tools=[provider.query_tool],
    system_message=provider.instructions(),
)
# 调用方获得一个 query_<id> 工具

内置提供者

Agno 内置 10+ 种 Context Provider：

• Web：Exa 搜索、Parallel 搜索（均支持 MCP 后端）
• Slack：频道浏览、消息搜索、媒体提取
• Google Drive：文件浏览、搜索、读取
• Gmail：邮件搜索、读取
• 文件系统：本地文件浏览和读取
• 数据库：SQL 查询（只读）
• Wiki：知识库查询
• MCP：通过 MCP 协议连接任意外部服务器
• Calendar：日历查询
• Workspace：工作区文件管理