Context cn
上下文构建器(Context Builder)是 Jianmu 在每次 LLM 调用前将多源信息装配为完整 Prompt 的运行时引擎。它采用 Provider-Filter 管道架构——先由一组 Provider 按序贡献消息块,再由一组 Filter 对聚合后的消息序列施加预算控制。三个工厂预设(for_chat、for_react、for_skill)覆盖了从简单对话到技能增强 Agent 的完整梯度,而 resolve_for_llm 的分发逻辑确保了显式注入 > 运行时覆盖 > 自动推导的优先级链。
架构全景:Provider-Filter 管道¶
上下文构建的核心是一个两阶段流水线。第一阶段,多个 MessageProvider 按注册顺序依次产出消息块——系统提示词、ReAct 协议、工具描述、技能指引、对话历史等。第二阶段,ContextFilter 链对聚合后的消息序列施加去尾控制:MaxMessagesFilter 按数量裁剪,TokenBudgetFilter 按 Token 预算截断,两者均保留以 system 角色为首的"钉住前缀(pinned prefix)"不被裁剪。
graph LR
subgraph Providers["Provider 阶段(有序)"]
P1["StaticPromptProvider<br/>系统提示词"] --> P2["ToolsDescProvider<br/>工具描述"]
P2 --> P3["BootstrapFilesProvider<br/>工作空间文件"]
P3 --> P4["SkillSetPromptProvider<br/>技能指引"]
P4 --> P5["StateHistoryProvider<br/>对话历史"]
end
subgraph Filters["Filter 阶段(有序)"]
F1["MaxMessagesFilter<br/>消息数量裁剪"] --> F2["TokenBudgetFilter<br/>Token 预算截断"]
end
Providers --> Filters --> LLM["LLM 调用"]
这条管道的核心优势在于 关注点分离:Provider 只关心"从哪取数据",Filter 只关心"数据是否超标"。用户可以通过组合不同的 Provider 和 Filter 实现任意复杂的 Prompt 装配逻辑,而三个工厂预设封装了最常见的组合模式。
核心抽象:三个 Protocol¶
上下文构建器体系建立在三个轻量 Protocol 之上,它们定义了 Provider-Filter 管道的契约。
| 抽象 | 核心方法 | 职责 |
|---|---|---|
MessageProvider |
get_messages(local_state, global_state, ctx) → Sequence[Message] |
从状态和运行时上下文中产出消息块 |
ContextFilter |
apply(messages, local_state, global_state, ctx) → Sequence[Message] |
对聚合后的消息序列进行后处理(裁剪/增强/清理) |
TokenCounterProtocol |
count_message(Message) → int / count_messages(Iterable[Message]) → int |
估算单条或一组消息的 Token 数量 |
ContextBuilderProtocol 本身暴露两个入口:build() 走完整的 Provider → Filter 管道,transform() 仅走 Filter 阶段——后者允许对已有消息列表直接施加预算控制而不重新生成 Prompt。
消息基础:Message 模型与存储¶
在深入 Provider 之前,需要理解 Jianmu 的消息基元。Message 是一个 Pydantic 模型,承载 role(system/user/assistant/tool)、content(文本或结构化块)、tool_call_id、tool_calls 和 metadata 字典。每个消息实例拥有一个稳定的 id(UUID hex),用于消息级差量和事件追踪。
# 轻量工厂函数
from jianmu.message import system, human, ai, tool
system("You are a helpful assistant.") # role="system"
human("What is 2+2?") # role="user"
ai("The answer is 4.") # role="assistant"
tool('{"result": 4}', name="calculator") # role="tool"
StateMessageStore 是默认的消息存储后端,以 StateManager 为持久化载体,支持可插拔的 消息强制转换器(Coercer) 和 保留策略(Retention)。StrictMessageCoercer 只接受 Message 实例或合法字典;LenientMessageCoercer 尽力将任意负载转为 Message。MaxCountRetention 策略在写入时保留最新的 N 条消息。写入管道是线程安全的,并在每次追加后向订阅者广播 MessageEvent。
Provider 详解:五大数据源¶
StaticPromptProvider:静态提示词注入¶
最基础的 Provider,将一段静态文本包装为 role="system" 的消息。三个工厂预设均将其置于 Provider 链首位,用于注入 Persona 提示词和 ReAct 协议。
from jianmu.memory.context import StaticPromptProvider
provider = StaticPromptProvider("You are a helpful assistant.")
# → [system("You are a helpful assistant.")]
ToolsDescProvider:工具描述注入¶
将预格式化的工具描述块嵌入系统消息。使用 format_tools_prompt() 模板包装——默认为 "Available tools:\n{tools_desc}"——可通过配置覆盖。
from jianmu.memory.context import ToolsDescProvider
provider = ToolsDescProvider("- calculator: perform arithmetic\n- search: web search")
# → [system("Available tools:\n- calculator: perform arithmetic\n- search: web search")]
StateHistoryProvider:对话历史加载与语义标注¶
这是最关键也是逻辑最复杂的 Provider。它从状态中提取对话历史,执行两项关键转换:
- 工具计划标注:将
assistant角色且包含tool_calls的历史消息改写为"This assistant message records your own previous tool-call decision...",并标记metadata.semantic_kind = "self_tool_plan"。 - 工具回执标注:将
role="tool"且metadata.tool_result不为空的消息改写为包含"execution receipt"注释的结构化 JSON,并标记metadata.semantic_kind = "tool_observation"。
当存在语义标注时,Provider 还会在历史消息前方插入一条 History semantics 系统消息,向模型解释标注含义。这套机制防止模型将自身先前的工具决策误读为新的传入消息,对多 Agent Swarm 场景尤为重要。
Provider 还支持通过 working_memory 适配器调用 build_view() 进一步压缩或重组历史序列。
BootstrapFilesProvider:工作空间文件注入¶
读取工作空间根目录下的引导文件(默认为 AGENTS.md 等,通过 prompt.bootstrap_files 配置),以 ## 文件名\n\n内容 格式嵌入系统消息。当 PromptRuntimeContext.include_bootstrap_files = True 时激活。
SkillSetPromptProvider:技能指引渲染¶
最复杂的 Provider,负责将选中的技能集渲染为 Prompt 友好的系统消息块。它支持两种模式:
- summary 模式:当工具集中存在文件读取工具时,输出技能索引摘要 + 按需加载策略;当无文件读取工具时,输出摘要 + 回退策略(告知模型技能指令已内联)。
- full 模式:输出完整技能列表 + 详细技能指令(
Skill instructions:标题),可选附加运行时资源路径映射。
Provider 还会注入技能能力策略提示词(区分 Prompt 技能和 BT 技能的执行方式)、资源使用策略(工作目录、Skill Root 路径映射),以及在 always_only 模式下仅输出 # Active Skills 标题下的活动技能详情。
Filter 详解:消息数量与 Token 预算控制¶
工具调用事务完整性¶
包含 tool_calls 的 assistant 消息与其紧随的 role="tool" 结果构成一个不可拆分的历史事务。MaxMessagesFilter 和 TokenBudgetFilter 会以事务为单位裁剪,因此预算边界不会只保留调用而丢掉结果(或反过来)。请求 Provider 前,Jianmu 还会用 sanitize_tool_call_history() 投影历史:损坏的历史事务会从本次请求中移除;损坏的当前尾部事务则会在本地抛错,不会发送非法 Provider 请求。对于旧版内部调用,如果 assistant 侧缺少 ID 但紧邻工具结果的对应关系明确,系统会安全地使用结果 ID 补全调用 ID。
MaxMessagesFilter:消息数量裁剪¶
限制消息总数,同时保留领先的钉住角色(默认 system)。裁剪策略为 "保留钉住前缀 + 最近 N 条尾部消息"——这是 ReAct 和聊天场景中常见的"保留系统提示词 + 最新对话"模式。
from jianmu.memory.context import MaxMessagesFilter
f = MaxMessagesFilter(max_messages=5, pinned_roles=("system",))
# 输入: [sys1, sys2, u1, a1, u2, a2, u3, a3]
# 输出: [sys1, sys2, a1, u2, a2, u3, a3] # 钉住前2条 + 最近3条
TokenBudgetFilter:Token 预算截断¶
根据 Token 预算裁剪消息列表。核心行为:
- 钉住前缀(默认
system角色)始终保留完整。 - 尾部消息从头部(
trim_from_start=True)或尾部(trim_from_start=False)逐条移除,直到总 Token 数不超预算。 - Token 计数由可插拔的
TokenCounterProtocol实现完成,默认使用SimpleTokenCounter。
三种 Token 计数器¶
| 计数器 | 原理 | 依赖 | 适用场景 |
|---|---|---|---|
SimpleTokenCounter |
字符数 ÷ 每 Token 字符数(默认 4) | 无 | 快速估算、离线环境 |
TiktokenTokenCounter |
OpenAI 官方 tiktoken 库 | pip install tiktoken |
OpenAI 模型精确计数 |
HFTokenCounter |
HuggingFace AutoTokenizer | pip install transformers |
开源模型精确计数 |
三种计数器均实现同一接口,可互换注入 TokenBudgetFilter。
工厂预设:三个场景化入口¶
ContextBuilder 提供三个类方法工厂,封装不同场景的 Provider + Filter 组合。
for_chat:纯对话上下文¶
Provider 链:StaticPromptProvider(persona) → BootstrapFilesProvider(可选) → StateHistoryProvider
不注入工具描述或技能指引。适用于无工具的简单对话。
builder = ContextBuilder.for_chat(
system_prompt="You are a helpful assistant.",
max_messages=50,
max_tokens=8000,
runtime_prompt=PromptRuntimeContext(include_bootstrap_files=True),
)
for_react:ReAct Agent 上下文¶
Provider 链:StaticPromptProvider(persona) → StaticPromptProvider(react_protocol) → ToolsDescProvider(可选) → BootstrapFilesProvider(可选) → StateHistoryProvider
在 for_chat 基础上增加 ReAct 协议(定义 "Final Answer:" 输出格式)和工具描述块。
builder = ContextBuilder.for_react(
system_prompt="You are a math assistant.",
tools_desc="- calculator: perform arithmetic\n- search: web search",
max_messages=30,
max_tokens=16000,
)
for_skill:技能感知上下文¶
Provider 链:StaticPromptProvider(persona) → StaticPromptProvider(react_protocol)(可选) → BootstrapFilesProvider(可选) → SkillSetPromptProvider → StateHistoryProvider
for_skill 支持两种模式:
- 显式模式:直接传入
skill_set,由SkillSetPromptProvider按mode(summary/full)渲染。 - 运行时解析模式:
runtime_prompt携带skills_catalog、skills_dir或技能上下文开关,Builder 先通过SkillsCatalog.resolve_runtime()解析有效目录对象,再根据include_skills_summary/include_active_skills决定渲染策略。
运行时目录解析的优先级是:
runtime_prompt.skills_catalogruntime_prompt.skills_dir- 静态配置
paths.skills_dir - 当前工作目录下的
./skills
如果显式传入了 skill_set,它优先于运行时发现的技能。
resolve_for_llm:自动分发¶
LLM 节点(SimpleLLMNode 和 AgentLLMNode)并不直接调用工厂方法,而是通过 resolve_for_llm 的分发逻辑确定最终的 Builder。优先级链为:
graph TD
A["显式 context_builder 参数"] -->|非空| R["直接返回"]
A -->|为空| B["runtime_prompt.context_builder"]
B -->|非空| R
B -->|为空| C["allow_skill_context AND<br/>(skill_set 或 runtime skill 请求)"]
C -->|满足| S["for_skill(...)"]
C -->|不满足| D["prefer_react OR tools_desc"]
D -->|满足| T["for_react(...)"]
D -->|不满足| U["for_chat(...)"]
SimpleLLMNode._resolve_context_builder() 不传 prefer_react=True,因此默认落入 for_chat;AgentLLMNode._resolve_context_builder() 会传 prefer_react=True,因此即使工具描述文本还没有保证非空,也会优先选择 ReAct 语义。
Prompt 配置体系:默认值与覆盖¶
jianmu/config/prompts.py 提供了完整的默认 Prompt 文本库和覆盖机制。所有可配置的 Prompt 片段均通过 _prompt_override(field_name, fallback) 函数解析——优先读取 jianmu.yaml 中 prompt 段的配置值,未配置时使用内置默认值。
关键默认值:
| 配置字段 | 默认值(摘要) |
|---|---|
persona_default_prompt |
"You are a helpful assistant." |
react_protocol_prompt |
定义 Final Answer 输出格式、工具使用规范 |
history_semantics_prompt |
解释 self_tool_plan 和 tool_observation 标注 |
skill_capability_policy_prompt |
区分 Prompt 技能和 BT 技能的执行语义 |
skill_on_demand_policy_prompt |
按需加载 SKILL.md 的指导 |
skill_fallback_policy_prompt |
文件读取工具不可用时的回退说明 |
skill_resource_usage_policy_prompt |
运行时可见路径和资源使用策略 |
运行时注入:PromptRuntimeContext¶
PromptRuntimeContext 是运行时级别的 Prompt 覆盖载体,通过 RunContext.prompt_runtime 注入到节点中。它承载:
| 字段 | 类型 | 作用 |
|---|---|---|
context_builder |
ContextBuilderProtocol \| None |
直接注入 Builder,优先级最高 |
skills_dir |
str \| None |
技能文件目录路径 |
workspace_dir |
str \| None |
工作空间根路径 |
skills_catalog |
Any \| None |
预加载的技能目录对象 |
include_skills_summary |
bool |
是否渲染技能索引摘要 |
include_active_skills |
bool |
是否内联活动技能详情 |
include_bootstrap_files |
bool |
是否注入工作空间引导文件 |
不同工厂预设对 PromptRuntimeContext 的消费语义不同:for_chat 和 for_react 只消费 include_bootstrap_files、workspace_dir 这类通用字段,刻意忽略技能相关字段。只有当调用方显式选择了技能感知路径(allow_skill_context=True 或直接调用 for_skill)时,技能字段才会生效。
约束联动:max_messages 的多层控制¶
max_messages 在 Jianmu 中存在三个控制层面,各自作用于不同阶段:
| 层面 | 位置 | 作用阶段 |
|---|---|---|
Constraints.max_messages |
jianmu/config/constraints.py |
全局配置,可由 Guard 层读取 |
StateMessageStore(max_messages=N) |
jianmu/message/store.py |
消息写入时截断(保留最新 N 条) |
ContextBuilder(max_messages=N) → MaxMessagesFilter |
jianmu/memory/context/builder.py |
Prompt 装配时裁剪(保留钉住前缀 + 最新 N 条) |
三者的设计意图不同:StateMessageStore 的截断发生在持久化写入时,控制存储膨胀;ContextBuilder 的裁剪发生在 LLM 调用前的 Prompt 装配阶段,可能比存储上限更严格以控制 Token 消耗。两者可以共存——存储保留 100 条历史,但每次调用只向模型发送最近 20 条。
端到端数据流:从节点到模型调用¶
以下序列图展示了 AgentLLMNode.update_async() 中完整的上下文构建路径:
sequenceDiagram
participant Node as AgentLLMNode
participant CB as ContextBuilder
participant P1 as StaticPromptProvider
participant P2 as ToolsDescProvider
participant P3 as StateHistoryProvider
participant F1 as MaxMessagesFilter
participant F2 as TokenBudgetFilter
participant MC as ModelClient
Node->>Node: _prepare_messages() → 从 StateManager 读取原始消息
Node->>Node: _resolve_context_builder() → resolve_for_llm()
Node->>CB: build(local_state, global_state, ctx, tools_schema)
CB->>P1: get_messages() → [system(persona)]
CB->>P2: get_messages() → [system(tools_desc)]
CB->>P3: get_messages() → 标注历史语义 + 返回消息列表
CB->>CB: 聚合所有 Provider 产出
CB->>F1: apply(messages) → 按 max_messages 裁剪
CB->>F2: apply(messages) → 按 max_tokens 截断
CB-->>Node: 完整消息序列
Node->>MC: invoke(messages=full_messages, config=model_config)
每个 Provider 和 Filter 的异常均被捕获并记录日志,不会中断整体管道。
扩展指南:自定义 Provider 与 Filter¶
由于 MessageProvider 和 ContextFilter 均为 Protocol(而非抽象基类),任何实现了对应签名的对象均可参与管道。
自定义 Provider 示例(注入当前时间戳):
from jianmu.memory.context.base import MessageProvider
from jianmu.message import Message, system
from datetime import datetime
class TimestampProvider:
def get_messages(self, local_state, global_state=None, ctx=None):
ts = datetime.now().isoformat()
return [system(f"Current time: {ts}")]
自定义 Filter 示例(移除重复的连续用户消息):
class DeduplicateFilter:
def apply(self, messages, local_state=None, global_state=None, ctx=None, **kwargs):
result = []
for msg in messages:
if result and msg.role == "user" and result[-1].role == "user":
if msg.content == result[-1].content:
continue
result.append(msg)
return result
通过 ContextBuilder(providers=[...], filters=[...]) 直接组合即可使用。
导航建议¶
完成本文阅读后,建议按以下路径继续深入:
- LLM 节点:AgentLLMNode 与 SimpleLLMNode 的上下文构建与模型调用——了解
resolve_for_llm在节点中的完整调用链路 - Skill 定义与目录:SKILL.md 解析、行为树技能与 Prompt 技能——理解
SkillSetPromptProvider渲染的技能来源 - Checkpoint 与长期记忆:FileCheckpointer、Mem0 集成与记忆工具——了解
StateHistoryProvider所读取的历史消息如何被持久化 - 模型接入:ModelClient 外观——理解上下文构建的最终消费者