数据截至 (上游 commit 4a030776a31a)
上下文装配:检索结果、工具、指令如何分层进 prompt
30 秒导读: 前四章分别产出了「可检索的节点」「检索回来的片段」「引用锚点」「挂进循环的工具」。 这一章讲最后一步:每一轮把请求发给模型之前,这些产物怎么被组织成一份完整的 prompt。 PrivateGPT 不用字符串拼接,而是把每一块上下文做成一个带类型 + 优先级的「层」,叠成一个 不可变的
ContextStack,再按规则渲染成系统提示 + 工具列表 + 文档。
本章聚焦上下文如何组织。循环怎么转(迭代、工具执行、拦截器时序)是 04 章的事; 检索算法本身是 02 章的事——这里都不重复。
1. 这是什么(零基础也能懂)
一句话定义
上下文装配 = 在调用大模型之前,把「给模型看的所有东西」按类别摆好、排好序,拼成一份最终请求。
给模型看的东西通常有这么几类:
| 类别 | 例子 | 谁产出的 |
|---|---|---|
| 指令 | 系统提示、平台守则、单个工具的用法说明 | 平台 / 拦截器 |
| 技能 | 可用技能目录、已激活技能的完整说明书 | 技能子系统 |
| 工具 | 工具的名字 + 参数 schema | 04 章的工具封装 |
| 文档 | 检索回来的片段(带引用锚点) | 02 / 03 章 |
解决什么问题
假设你在写一个 RAG agent,每轮请求要发给模型的内容来自四面八方:一段系统提示、几个检索到的文档、
一批工具定义、可能还临时加一句「注意省点 token」。最朴素的做法是把它们 + 成一个大字符串——
但很快就会失控:
- 谁排在前、谁排在后?顺序一乱,模型的注意力就被带偏。
- 同一轮里两个来源都注册了一个叫
search的工具,冲突了怎么发现? - 这一轮要临时去掉某类内容(比如关掉文档),字符串拼好了就很难拆回来。
- 循环跑到第 3 轮,又检索回来两篇新文档,要增量加进去,而不是重拼一遍。
PrivateGPT 的答案:别拼字符串,拼「层」。 每块上下文是一个有类型的对象,整体是一叠有序的层。
一句话直觉
把最终 prompt 想成一份分栏的报纸:头版是系统指令,中间是技能和文档,工具是夹在里面的 一张「能点的按钮清单」。你不是把所有字揉成一团,而是一栏一栏码好,最后印刷时按版面顺序排下来。
用起来什么样
一段示意代码,先建立直觉(真实实现见后文):
# 示意,非源码:装配一份上下文,再渲染成请求
stack = (ContextStackBuilder()
.set_user_instructions("你是一个只依据文档回答的助手") # 一层:指令
.add_document(retrieved_doc, source="citations") # 一层:检索文档
.add_tools([search_tool], source="rag") # 一层:工具
.build()) # 冲突校验 + 冻结
request = build_request_from_context_stack(base_request, stack)
# request.system.prompt ← 指令层按优先级渲染出的系统提示
# request.tool_config.tools ← 所有工具层去重后的扁平列表
# request.context.documents ← 所有文档层
重点看:指令、文档、工具是三种不同的层,build() 时统一校验,渲染时各走各的通道。
2. 顶层全景(它大概怎么转)
一张图:从「层栈」到「一份请求」
怎么读这张图:上半部是装配(把各种层叠起来),下半部是渲染(把层栈拆成模型能吃的三样东西)。
各路来源(拦截器 / 请求本身 / 技能子系统 / 检索)
│ │ │ │
▼ ▼ ▼ ▼
┌───────────────────────────────────────────────┐
│ ContextStack (不可变层栈) │
│ ┌─────────────────────────────────────────┐ │
│ │ 指令层 USER_INSTRUCTIONS priority 100 │ │ ← 渲染进系统提示
│ │ 技能目录 SKILL_CATALOG priority 300 │ │ ← 渲染进系统提示
│ │ 技能正文 SKILL_BODY priority 400 │ │ ← 渲染进系统提示
│ │ 工具说明 TOOL_INSTRUCTIONS priority 450 │ │ ← 渲染进系统提示
│ │ ─────────────── 状态层(不渲染文本)─────── │ │
│ │ 工具定义 TOOL_DEFINITIONS │ │ ← 走 all_tools()
│ │ 文档 DOCUMENT │ │ ← 走 all_documents()
│ └─────────────────────────────────────────┘ │
└───────────────────────────────────────────────┘
│ │ │
to_system_prompt() all_tools() all_documents()
│ │ │
▼ ▼ ▼
┌──────────────────────────────────────────────┐
│ ResolvedChatRequest(发给模型的一份请求) │
│ system.prompt tool_config.tools context.documents │
└──────────────────────────────────────────────┘
一个关键分野:层分两类——「提示层」渲染成文本进系统提示,「状态层」被程序读取(工具、文档),
从不变成系统 提示里的文字。layer_type.py:4 的 LayerType 注释把这条界线写死了。
部件一句话职责
| 部件 | 干什么 | 在哪个文件 |
|---|---|---|
LayerType | 枚举所有层类型,并划分「提示层 / 状态层」 | context/models/layer_type.py:4 |
各 *Layer 类 | 每种上下文一个带类型的对象,各自定义 render() | context/models/context_layer.py |
ContextStack | 不可变、有序的层集合;提供 to_system_prompt / all_tools 等读取器 | context/models/context_stack.py:17 |
ContextStackBuilder | 用链式 API 攒层,build() 时做工具名唯一性校验 | context/services/context_stack_builder.py:23 |
build_initial_context_stack | 从原始请求造出第一版层栈 | chat/utils/request_builder.py:16 |
build_request_from_context_stack | 每轮把层栈重新渲染成一份 ResolvedChatRequest | chat/utils/request_builder.py:51 |
_build_tools | 按 tool_choices 过滤出本轮真正可用的工具 | chat/chat_engine.py:939 |
context/utils/ | 横切:时钟 utc_now、工具编解码 serialize_tool_specs | context/utils/context_clock.py、toolset_codec.py |
主线走一遍(高层,不进代码)
- 请求进来 →
build_initial_context_stack把请求里已有的系统提示、工具、文档各自包成一层,叠成初始栈。 - 循环每迭代一次,拦截器可能往栈里追加层(平台守则、检索回来的文档、激活的技能……)—— 每次追加都返回一个新栈(旧栈不可变)。
- 每轮 LLM 调用前,
build_request_from_context_stack把当前栈重新渲染成一份请求: 指令层排序拼成系统提示,工具层去重成工具列表,文档层收集成文档。 _build_tools再按tool_choices从工具列表里筛掉本轮不该给的,交给模型。
记住这条主线:层栈是「真相」,请求是每轮从它派生出来的「快照」。
3. 核心原理(逐个机制,由浅入深)
3.1 为什么用「层栈」而不是字符串拼接
它要解决的小问题: 上下文来自多个来源、要能增量增删、要能排序、要能校验——纯字符串这些全做不到。
思路: 把每块上下文升格成一个有类型的对象。类型带来三样字符串没有的能力:
- 可路由:文档走文档通道、工具走工具通道,而不是全挤进一个字符串。
- 可排序:每层带
priority,渲染时按优先级排,顺序稳定可控。 - 可校验/可增删:能按类型筛选、按来源移除、追加时做冲突检查。
层的两个基类字段——每一层都从 BaseContextLayer 继承 source(这层从哪来)和 priority
(渲染优先级,越小越靠前):
# context/models/context_layer.py:13
class BaseContextLayer(BaseModel):
source: str = Field(default="request", ...) # 来源:'platform' / 'skill:git' / 'citations'
priority: int = Field(default=1000, ...) # 越小越先渲染;状态层忽略此字段
model_config = ConfigDict(frozen=True, ...) # 冻结:每层不可变
def render(self) -> str:
return "" # 默认不产出文本(状态层就用这个默认)
BaseContextLayer 在 context_layer.py:12,render() 默认返回空串——这正是「状态层」的默认行为:
不进系统提示。
九种层,各是一类上下文(都在 context_layer.py):
| 层类 | 类型 | priority | 提示/状态 | 装什么 |
|---|---|---|---|---|
UserInstructionsLayer | USER_INSTRUCTIONS | 100 | 提示 | 基线系统指令 |
RuntimeInstructionsLayer | RUNTIME_INSTRUCTIONS | 200 | 提示 | 临时运行时提示(如压缩提示) |
SkillCatalogLayer | SKILL_CATALOG | 300 | 提示 | 可用但未加载的技能目录 |
SkillBodyLayer | SKILL_BODY | 400 | 提示 | 已激活技能的完整说明 |
ToolInstructionsLayer | TOOL_INSTRUCTIONS | 450 | 提示 | 某个工具的专属用法说明 |
ContextPromptLayer | CONTEXT | 600 | 提示 | 渲染好的上下文段落 |
ToolDefinitionsLayer | TOOL_DEFINITIONS | 2000 | 状态 | 一批 ToolSpec 工具定义 |
DocumentLayer | DOCUMENT | 2000 | 状态 | 一个检索/引用来的 Document |
ContentBundlesLayer | CONTENT_BUNDLES | 2000 | 状态 | 技能附带的文件 包 |
注意 priority 的递增设计:指令(100)一定在技能目录(300)前,技能目录在技能正文(400)前…… 这不是巧合,是刻意的版面顺序。状态层统一给 2000 且不参与文本渲染。
AnyContextLayer 用 pydantic 的判别式联合(Field(discriminator="type"),context_layer.py:239)
把九种层收进一个类型——序列化/反序列化时靠 type 字段自动分派到正确的类。
提示层怎么各自渲染: 每种提示层重写 render()。举个非平凡的例子,技能目录会把每个条目渲染成 XML:
# context/models/context_layer.py:95 SkillCatalogLayer.render()
def render(self) -> str:
if not self.entries:
return ""
lines = ["<available_skills>"]
for entry in self.entries:
lines.append(" <skill>")
lines.append(f" <name>{entry.name}</name>")
lines.append(f" <description>{entry.description}</description>")
# ...(location / resources 可选)
lines.append("</available_skills>")
return "\n".join(lines)
这段把结构化的技能目录变成 <available_skills>…</available_skills> 文本——字符串拼接做不到的
「按类型定制渲染」,层模型免费给了你。
3.2 层栈:不可变、有序、按类型读取
它要解决的小问题: 循环每轮都在改上下文;如果就地改一个共享对象,快照、回放、并发都容易踩坑。
思路: ContextStack 冻结(frozen=True),所有「修改」都返回新实例:
# context/models/context_stack.py:92
def append_layer(self, layer: AnyContextLayer) -> "ContextStack":
"""Return a new stack with *layer* appended."""
return ContextStack(layers=[*self.layers, layer])
append_layer(context_stack.py:115)、insert_layer(:88)、remove_layers_of_type(:100)、
remove_layers_of_source(:106)——每一个都不改原栈,而是造一个新栈。所以拦截器链里
stack = stack.append_layer(...) 这种写法,天然是一串不可变快照(与 04 章循环里
「所有 mutation 返回新实例」的约定一致,见 chat_state.py:120 注释)。
三个核心读取器,把「一叠混合的层」拆成模型要的三样:
第一样,系统提示——只挑提示层,按 (priority, 插入序) 排序后渲染、去空:
# context/models/context_stack.py:35 ContextStack.to_system_prompt()
def to_system_prompt(self) -> list[TextBlock]:
ordered_layers = sorted(
enumerate(self.layers),
key=lambda item: (item[1].priority, item[0]), # 先按优先级,再按插入顺序
)
chunks = [rendered
for _, layer in ordered_layers
for rendered in [layer.render()]
if rendered.strip()] # 空渲染(状态层)自动被过滤掉
return [TextBlock(text=chunk) for chunk in chunks]
重点看排序键 (priority, index):优先级相同的层,保持它们被加入的先后——顺序完全确定,不随机。
状态层因为 render() 返回空串,if rendered.strip() 这一步把它们自动排除在系统提示之外。
第二样,工具列表——扫所有 ToolDefinitionsLayer,按名字去重后扁平化:
# context/models/context_stack.py:49 ContextStack.all_tools()
def all_tools(self) -> list[ToolSpec]:
seen: set[str | None] = set()
result: list[ToolSpec] = []
for layer in self.layers:
if not isinstance(layer, ToolDefinitionsLayer):
continue
for tool in layer.tools:
if tool.name not in seen: # 首次出现的名字才保留
seen.add(tool.name)
result.append(tool)
return result
第三样,文档——all_documents()(context_stack.py:76)按插入序收集所有 DocumentLayer 里的
Document(就是 03 章里带 shorter_id 引用锚点的那个实体)。这样,
检索出来的文档层和工具定义层可以在同一个栈里共存,却各走各的读取器、互不干扰。
3.3 构建器:攒层 + 工具名唯一性校验
它要解决的小问题: 多个来源各自注册工具,很容易两处都叫 search——模型一旦看到重名工具,
调用路由就会歧义。这种冲突必须在装配时就炸出来,而不是等模型调用才出错。
思路: ContextStackBuilder 提供链式 API 逐层添加,build() 时统一做校验再冻结成 ContextStack。
链式方法一览(全在 context_stack_builder.py):
| 方法 | 加的层 | 行 |
|---|---|---|
set_user_instructions(text) | UserInstructionsLayer | :29 |
add_skill_catalog(skills) | SkillCatalogLayer | :40 |
add_activated_skill(...) | SkillBodyLayer | :47 |
add_tools(tools, source) | ToolDefinitionsLayer | :62 |
add_document(document, source) | DocumentLayer | :71 |
build() | 校验 + 冻结 | :80 |
唯一性校验发生在 build() → validate_tool_name_uniqueness():跨所有工具定义层建一张
名字 → 来源 的表,撞名就抛 ToolNameConflictError,并带上两个来源方便定位:
# context/services/context_stack_builder.py:85 validate_tool_name_uniqueness()
seen: dict[str, str] = {}
for layer in self.layers:
if layer.type is not LayerType.TOOL_DEFINITIONS:
continue
for tool in layer.tools:
if tool.name is None:
continue
existing = seen.get(tool.name)
if existing is not None:
raise ToolNameConflictError( # 见 context/errors.py
f"Tool name conflict: name='{tool.name}', "
f"layer_a='{existing}', layer_b='{layer.source}'")
seen[tool.name] = layer.source
ToolNameConflictError 是带稳定错误码 TOOL_NAME_CONFLICT 的类型化领域错误
(context/errors.py),错误信息里 layer_a / layer_b 直接点出哪两层撞名。
注意校验时机的区别:
all_tools()(3.2)在重名时静默保留第一个——那是运行时防御;validate_tool_name_uniqueness()(这里)在构建时主动抛错——那是装配期的严格门禁。两者互补。
3.4 每轮如何从层栈重建 LLM 请求
它要解决的小问题: 层栈在整个循环里活着并不断长大,但每次 LLM 调用要的是一份扁平、当下的请求。
思路: 每次迭代,把当前层栈重新渲染成一份 ResolvedChatRequest——层栈是源,请求是派生快照。
入口一:初始栈。 请求刚进来时,build_initial_context_stack 把请求里已有的系统提示、工具、
文档各包成对应层:
# chat/utils/request_builder.py:16 build_initial_context_stack()
if request.system.prompt:
stack = stack.append_layer(UserInstructionsLayer(text=request.system.prompt, source=source))
if request.tool_config.tools:
stack = stack.append_layer(ToolDefinitionsLayer(tools=list(request.tool_config.tools), source=source))
if request.context.documents:
for document in request.context.documents:
stack = stack.append_layer(DocumentLayer(document=document, source=source))
这份初始栈存进循环状态 ChatInputState.context_stack(chat_state.py:35;上游已把循环状态从
ChatLoopState 系列改名成 ChatState 系列);之后拦截器往里追加层。若调用方没给栈,
initialize_run 会用它兜底(chat_engine.py:893)。
入口二:每轮重建。 迭代循环里,_run_iteration 在调用模型前先重建请求
(chat_engine.py:388):
# chat/utils/request_builder.py:51 build_request_from_context_stack()
request = ResolvedChatRequest.model_validate(base_request, from_attributes=True)
request.tool_config.tools = list(context_stack.all_tools()) # 工具:去重扁平列表
request.context.documents = context_stack.all_documents() or None # 文档
request.context.content_bundles = context_stack.all_bundles() # 技能文件包
request.messages = [m for m in request.messages if m.role != MessageRole.SYSTEM] # 清掉旧系统消息
request.system.prompt = _render_system_prompt_text(context_stack) # 用层栈重渲染系统提示
看第 4、5 行:先删除历史里所有 SYSTEM 消息,再用当前层栈重新渲染系统提示——保证系统提示
永远是「当下层栈的最新投影」,不会有上一轮的残留。_render_system_prompt_text(:64)只是把
to_system_prompt() 出来的块拼起来、 滤空。
3.5 tool_choices:从「所有工具」里筛出本轮可用的
它要解决的小问题: 层栈里可能挂着一堆工具,但某一轮你想强制模型只用其中某个、或干脆不给工具。
思路: all_tools() 给的是全集;_build_tools 再叠一层本轮策略过滤,两者分工明确。
# chat/chat_engine.py:939 _build_tools()
tool_specs = state.input.context_stack.all_tools() # 全集(已去重)
allowed_names = set(select_tool_names(
tool_choices=state.input.request.tool_config.tool_choices,
tool_names=[tool.name or "" for tool in tool_specs]))
selected = [t for t in tool_specs if (t.name or "") in allowed_names]
return [adapt_to_async_tool(t.to_function_tool()) for t in selected]
过滤规则在 select_tool_names,很直白:
# chat/utils/tool_utils.py:32 select_tool_names()
if tool_choices in ("auto", "any"):
return tool_names # 全给
if isinstance(tool_choices, str):
return [n for n in tool_names if n == tool_choices] # 只给这一个
return [n for n in tool_names if n in tool_choices] # 只给这个白名单
tool_choices 取值 | 效果 |
|---|---|
"auto" / "any" | 层栈里全部工具都给模型 |
单个字符串(如 "search") | 只暴露这一个工具 |
| 字符串列表 | 只暴露列表里的这些 |
选出的 ToolSpec 再经 to_function_tool()(chat_config_models.py:373)转成 llama-index 工具、
adapt_to_async_tool 包成异步工具——这一步之后就交给 04 章的循环去执行了,不在本章范围。
分层的意义: 「有哪些工具」由层栈决定(装配期),「这一轮给哪些」由 tool_choices 决定(运行期)。同一个层栈,不同轮可以给模型不同的工具子集,而层栈本身不用动。
4. 横切工具:时钟与编解码(context/utils/)
这两个是装配层的辅助件,简述即可。
时钟 context_clock.py: 只有一个 utc_now()(:6),返回带时区的当前 UTC 时间。
把「取当前时间」收敛到一个函数,好处是可 mock、时区统一——上下文里凡是要打时间戳的地方都走它。
# context/utils/context_clock.py:6
def utc_now() -> datetime:
return datetime.now(UTC) # aware(带时区)时间戳
工具编解码 toolset_codec.py: 把 ToolDefinitionsLayer 里的 ToolSpec 在「对象 ↔ JSON」之间转换,
用于把工具定义持久化/传输(层要存下来或跨进程带走时)。
serialize_tool_specs(:10)只挑name / type / description / input_schema四个字段 (_TOOL_SPEC_KEYS)、丢掉None,压成紧凑 JSON——不序列化函数体,只留模型需要看的元数据。deserialize_tool_specs(:23)反向解析回ToolSpec列表,非 dict 的脏项会被跳过。
为什么只留四个字段?因为工具的可执行部分(
async_fn等)在装配侧靠代码重建,存储时只需要 「模型看得懂的描述性契约」。这跟 3.2 里all_tools()只关心name去重是一脉相承的思路。
5. 巧妙之处(可借鉴的技术)
-
提示层 vs 状态层,用「render 是否为空」自然分野。 状态层不重写
render()(默认返回空串),to_system_prompt()的if rendered.strip()顺手就把它们滤掉——不需要额外的 if/else 判断类型。 见context_stack.py:63+context_layer.py:29。 -
优先级用递增常量硬编码进各层类。 指令 100 → 技能 300/400 → 工具说明 450 → 上下文 600, 版面顺序写死在类型里而非调用点,任何来源加同类型层都自动落到正确位置。见
context_layer.py各priority默认值。 -
排序键带插入序做 tiebreak。
(priority, index)(context_stack.py:51)让同优先级层顺序确定, 消除了「顺序不稳定」这类难复现的 bug。 -
不可变栈 = 免费的快照。 每次
append_layer返回新栈(context_stack.py:115),循环各轮天然 持有各自的上下文快照,回放和调试都简单。 -
两级工具名防护。 构建期
validate_tool_name_uniqueness硬抛错(context_stack_builder.py:85), 运行期all_tools静默去重(context_stack.py:63)——严格门禁 + 兜底防御,各管一段。 -
判别式联合收九种层。
Field(discriminator="type")(context_layer.py:239)让层栈能被 pydantic 无损序列化/反序列化,是「层能被持久化并跨轮携带」的基础。
6. 边界与局限(诚实)
-
构建器与循环入口是两条路。 主循环实际走的是
build_initial_context_stack+ 拦截器直接append_layer(request_builder.py:28、各server/chat/interceptors/*),ContextStackBuilder作为服务(services/__init__.py导出)提供带校验的装配 API。因此唯一性校验主要覆盖走构建器的路径; 走all_tools()的路径靠静默去重兜底,不会主动报冲突(inferred,基于两处代码的不同行为)。 -
系统提示是「块拼接」,不做去重/ 裁剪。
to_system_prompt只按优先级排序、滤空 (context_stack.py:63),不判断两层内容是否重复、也不做 token 预算裁剪——预算/压缩是别处的事。 -
all_tools的去重只看name。 两个不同定义但同名的工具,后一个会被静默丢弃 (context_stack.py:63),这轮不会有任何提示。 -
文档/工具/技能层本身不「排序」进 prompt。 它们是状态层,进的是
context.documents/tool_config.tools通道,由下游(引用渲染、工具封装)决定怎么呈现给模型——本章只负责把它们 正确归位到对应通道。
7. 横向对比(同组其它章)
| 想了解 | 去这章 |
|---|---|
| PrivateGPT 整体是什么、检索怎么转 | index.md |
| 文件如何变成可检索的树状节点 | 01-ingestion-tree-nodes.md |
| 一次查询如何取回带上下文的片段 | 02-semantic-retrieval-tree-expansion.md |
Document、shorter_id 引用锚点从哪来 | 03-citations.md |
| 工具怎么挂进流式 agent 循环、循环怎么迭代 | 04-agentic-rag-chat-loop.md |
衔接关系: 03 章产出的 Document(带 shorter_id 引用锚点)在本章被包进 DocumentLayer;
02 章的检索片段最终也以文档层形式进栈;04 章封装的工具以 ToolDefinitionsLayer 进栈。
本章是把前面所有产物「归位、排序、渲染成一份请求」的收口环节。
8. 代码地图(导航索引)
| 主题 | 文件 | 关键符号 |
|---|---|---|
| 层类型枚举 / 提示层vs状态层 | private_gpt/components/context/models/layer_type.py | LayerType |
| 九种层与各自 render | private_gpt/components/context/models/context_layer.py | BaseContextLayer、UserInstructionsLayer、SkillCatalogLayer、SkillBodyLayer、ToolInstructionsLayer、DocumentLayer、ToolDefinitionsLayer、ContentBundlesLayer、AnyContextLayer |
| 不可变层栈与读取器 | private_gpt/components/context/models/context_stack.py | ContextStack、to_system_prompt、all_tools、all_documents、append_layer、insert_layer |
| 构建器与唯一性校验 | private_gpt/components/context/services/context_stack_builder.py | ContextStackBuilder、set_user_instructions、add_tools、add_document、add_skill_catalog、build、validate_tool_name_uniqueness |
| 冲突错误类型 | private_gpt/components/context/errors.py | ToolNameConflictError、ContextErrorCode |
| 初始栈 / 每轮重建请求 | private_gpt/components/engines/chat/utils/request_builder.py | build_initial_context_stack、build_request_from_context_stack、_render_system_prompt_text |
| tool_choices 过滤 | private_gpt/components/engines/chat/chat_engine.py | _build_tools |
| 工具选择策略 | private_gpt/components/engines/chat/utils/tool_utils.py | select_tool_names |
| 横切:时钟 | private_gpt/components/context/utils/context_clock.py | utc_now |
| 横切:工具编解码 | private_gpt/components/context/utils/toolset_codec.py | serialize_tool_specs、deserialize_tool_specs |