引用(citations):把答案锚回原文
30 秒导读: RAG 的答案要可信,读者得能点开每句话背后的原文。PrivateGPT 的做法是: 把检索到的每个源块用一对方括号 token
[]包住、配一个 4 字符的短 id 喂给模型, 模型在正文里原样吐出这个 id,管线再在流式生成的同时把[XXXX]解析成一个携带artifact_id/source_id/index的<citation>结构,随事件流回传给前端。本章只讲 「引用怎么生成、怎么表达」,不重复 02 章的检索算法。
1. 这是什么(零基础也能懂)
一句话定义: 引用(citation)= 在生成的答案里,给每个论断挂一个能点回具体源文档、 具体片段的可追溯标记。
它解决谁的什么问题。 RAG 系统会先检索一堆文档片段,再让大模型基于它们作答。 问题是:模型输出的是一段流畅的自然语言,读者无法知道"这句话到底出自哪个文件的哪一段"—— 也就无从判断它是真有依据,还是模型编的。引用就是把这条追溯链补回来。
用起来什么样。 开启引用后,一段回答长这样(方括号里是源标记):
Solar capacity grew 25% [A3F9]. Wind investments rose 50% [B7K2].
前端拿到的不是这串裸文本,而是每个 [A3F9] 已经被替换成一个结构化引用对象,
点一下就能跳到 A3F9 对应的源文件片段。
一句话直觉。 把它想成学术论文的脚注:正文里一个小上标数字,文末一条能查到出处的记录。 只不过这里的"脚注编号"是模型在生成时自己写进正文的,而"编号到出处的映射"由管线维护。
和 01 章/02 章的关系:
01 章把文件变成带 artifact_id 的可检索节点,02 章检索出相关片段;本章讲这些片段如何被
打上标记喂给模型、模型如何回引、管线如何把回引还原成可点击的出处。
本节不出现底层代码。目标:完全不懂的人读完知道"引用是干嘛的"。
2. 顶层全景(引用的一生)
引用不是一个函数,而是一条贯穿"进 prompt → 模型生成 → 流式回传"的闭环。先看这张图, 再逐段拆。
检索到的节点 (NodeWithScore)
│ ① 发号:给每个节点配一个 4 字符短 id
▼ init_nodes_with_shorter_ids
┌─────────────────────────┐ (随机 / 或抽一个"独特术语")
│ node.metadata │
│ shorter_id = "A3F9" │
│ artifact_id, abs_idx… │ ← ② 元数据锚点(定位回原文用)
└───────────┬─────────────┘
│ ③ 包装:每个源块前缀 [A3F9],拼进上下文
▼ format_context / format_llm_source
┌─────────────────────────┐
│ prompt 里的源: │ "Citation identifier [A3F9]
│ [A3F9] --- Content: …" │ --- Content: 太阳能装机…"
└───────────┬─────────────┘
│ ④ 模型读到源 + 引用协议(citations.j2),
│ 在正文里回吐 "… grew 25% [A3F9]."
▼
┌─────────────────────────┐ ⑤ 流式解析:把 [A3F9] 换成
│ extract_citations_... │ <citation id='A3F9' index='0'
│ (逐 delta 处理) │ artifact_id=… source_id=…>
└───────────┬─────────────┘
│ ⑥ 对外表示:Citation → ZylonCitation
▼ 挂在 TextBlock.citations,
前端事件流 源块单独走 SourceBlock
怎么读这张图: 从上到下是时间顺序;①②③ 发生在送进模型之前,④是模型生成,
⑤⑥发生在模型往外吐字的同时。核心巧思是:引用标记用最省 token 的 [XXXX] 进 prompt,
但回来时被"升级"成一个信息完整的结构。
部件一句话职责:
| 部件 | 干什么 | 文件 |
|---|---|---|
init_nodes_with_shorter_ids | 给每个检索节点发一个 4 字符短 id | components/engines/citations/utils.py:136 |
format_llm_source / format_context | 把源块前缀 [id] 拼进 prompt | components/engines/citations/format.py:34 :471 |
Document / Citation | 引用用的两个数据模型 | components/engines/citations/types.py:14 :36 |
MetadataKeys / MetadataChunk / MetadataNode | 挂在节点上的定位锚点(artifact_id / abs_idx / score) | components/ingest/metadata_helper.py:6 |
extract_citations_by_original_text | 把模型吐的 [id] 解析成 <citation> 结构 | components/engines/citations/utils.py:245 |
process_citations | 在流式事件里逐 delta 调上面那个解析器 | components/chat/processors/events/citations/citations.py:21 |
ZylonCitation | 引用的对外(API/前端)表示 | chat/extensions/citation.py:6 |
SourceBlock / TextBlock.citations | 引用随事件流出去的载体 | events/models/_content_blocks.py:653 :47 |
3. 核心原理
下面按"引用的一生"的四个阶段,逐个讲清楚。
3.1 特殊 token 包裹:让模型能"引用回来"
要解决的小问题: 怎么让模型在自由生成的文本里,精确地指回"就是刚才第 3 个源"?
思路。 给每个源一个唯一、极短、模型不会打错的代号,把这个代号用一对醒目符号包起来, 连同源内容一起放进 prompt,再明确告诉模型:"要引用时,原样写这个代号"。PrivateGPT 选的 包裹符号就是最普通的方括号 —— 起止 token 都只有一个字符,几乎不耗 token:
# components/engines/citations/utils.py:42
ORIGINAL_START_TOKEN = "["
ORIGINAL_END_TOKEN = "]"
包裹的动作只有一行,format_llm_source_str 把内容夹在起止 token 之间:
# components/engines/citations/format.py:22 —— format_llm_source_str
def format_llm_source_str(content, start_token="[", end_token="]", generate_citations=True):
if not generate_citations:
return "" # 关掉引用时,不写任何标记
return f"{start_token}{content}{end_token}" # → "[A3F9]"
format_llm_source(format.py:34)在此之上加一层:传入一个 Document,
取它的 document.id 作为 content 去包裹。这个 .id 很关键——它是短 id 优先的:
# components/engines/citations/types.py:60 —— Document.id
@property
def id(self) -> str:
return self.shorter_id or self.id_ # 有短 id 就用短 id,否则退回完整节点 id
为什么用短 id 而不是完整节点 id? 完整节点 id 是个长 UUID,模型既容易抄错、又白白烧 token。 短 id 只有 4 个字符(见 3.2),模型抄得准、prompt 也省。
源块最终在 prompt 里长什么样。 format_context(format.py:471)是对外入口,它按
settings().chat.format_context_strategy 选 list / xml / json 三种排版之一。以最直白的
list 版为例(_format_documents_as_list,format.py:51),每个源被拼成:
Citation identifier [A3F9]
---
Content:
太阳能装机容量同比增长 25% …
===
模型于是同时看到"这段内容"和"它的代号 [A3F9]",引用时照抄代号即可。三种排版策略只是把
同样的 [id] + 内容 换成列表 / <node id='[A3F9]'> XML / JSON 结构, 核心的 token 包裹不变。
模型被怎么教的。 光给代号还不够,还得约法三章。引用协议写在模板 citations.j2 里,
明确要求"恰好一对方括号 [XXXX]、别用双括号、别用括号、别放在标题/代码/公式里",并给了
few-shot 例子(prompts/templates/chat/guidelines/citations.j2:4)。create_citation_guidelines
(prompts/prompt_builder.py:515)负责把当前这批源的真实代号填进模板的 {{ all_cites }}。
3.2 短 id 从哪来:随机,还是"一个独特的词"
要解决的小问题: 4 字符代号怎么生成,才既唯一又不容易被模型搞混?
两条路。 发号入口是 generate_shorter_id(utils.py:121):
- 数值模式(
numerical_shorter_citations开):直接f"{index:04}",即0000、0001…… - 默认模式:以
node_id为随机种子,生成 4 位大写字母+数字,如A3F9。用node_id作种子 保证同一节点每次得到同一个短 id(确定性),而不是每次随机。
# components/engines/citations/utils.py:121 —— generate_shorter_id
def generate_shorter_id(index, node_id, length=4):
if NUMERICAL_SHORTER_ID:
return f"{index:0{length}}" # 0000, 0001 …
rng = random.Random(x=node_id) # 用节点 id 当种子 → 确定性
return "".join(rng.choices(string.ascii_uppercase + string.digits, k=length))
一个巧思:短 id 可以是"一个有意义的词"。 init_nodes_with_shorter_ids(utils.py:136,
在检索管线 workflows/retrieval/retrieval.py:200 被调用)在发随机号之前,先试着从每个源块里
抽一个只属于它、别的块没有的独特术语当代号:
# components/engines/citations/utils.py:151 —— init_nodes_with_shorter_ids(节选)
related_terms = potential_shorted_ids.get(i) # 该块抽到的独特术语
shorted_id = (
related_terms[0] # 有独特术语 → 用它当代号
if related_terms
else generate_shorter_id(index, node.node_id, 4) # 没有 → 退回随机 4 字符
)
node.metadata[SHORTER_ID_FIELD] = shorted_id.upper()
术语抽取由 term_extractor.py 的 TextAnalyzer 完成(下一节)。为什么费这个劲? 因为
一个"有意义的词"当代号,比 A3F9 更利于模型理解和正确回引,也更省 latency——
注释原话说这是为了 "generating shorter citation references and reducing the token use"(utils.py:142)。
收尾一步很重要: 短 id 是临时给模型看的脚手架,不能污染节点自身的元数据展示。所以
发完号立刻把 shorter_id 字段加进 excluded_llm_metadata_keys(utils.py:161),
让它不会作为普通元数据再被塞进别处。
3.3 术语抽取:挑一个"只属于这块"的词
要解决的小问题: 给定一批源块,怎么给每块找一个别的块都没有、又有代表性的词?
思路(三步)。 TextAnalyzer.get_unique_terms(term_extractor.py:159)是核心:
- 抽词。 对每块文本清洗(只留字母数字连字符)、分词、词形还原(lemmatize)、去停用词,
得到该块的词集合(
get_terms,term_extractor.py:123)。 - 求"独有"。 对第 i 块,减去所有其它块的词并集,剩下的就是只有它有的候选词
(
term_extractor.py:188)。 - 打分排序。 用
score_term(term_extractor.py:138)给候选词打分——出现在行首得 0.8、 在表格首列得 0.6、其它 0.4;取分数 > 0.2 的前max_terms个。
# components/engines/citations/term_extractor.py:186 —— 求各块独有词(节选)
for i, terms in enumerate(all_terms):
other_terms = set().union(*(s for j, s in enumerate(all_terms) if j != i))
unique = terms - other_terms # 只属于第 i 块的词
scored = [(t, self.score_term(t, texts[i])) for t in unique]
best = [t for t, s in sorted(scored, key=lambda x: x[1], reverse=True)
if s > 0.2][:max_terms] # 挑分最高、够显著的
边界与容错。 这一步是"锦上添花"、不能拖慢主流程,所以调用它的 analyze_texts_with_timeout
(utils.py:83)把它放进线程池并设 5 秒超时,超时或异常就返回空 dict,让 3.2 优雅退回
随机短 id。术语抽取还受 settings().data.enable_term_extractor 开关和 NLTK 依赖是否可用的保护
(utils.py:62、utils.py:31 的 try/except import)。
3.4 元数据锚点:引用凭什么能定位回具体源
要解决的小问题: [A3F9] 只是个代号,前端点它时,系统凭什么知道要跳到"哪个文件、哪一段"?
思路。 答案不在代号本身,而在每个节点从摄取期就挂着的一串元数据键。这些键是引用的
"定位坐标",在 metadata_helper.py 里用几个 StrEnum 定义:
| 元数据枚举 | 关键键 | 含义(定位作用) | 定义 |
|---|---|---|---|
MetadataKeys | artifact_id | 源文档/工件的 id —— 指回是哪个文件 | metadata_helper.py:14 |
MetadataKeys | file_name | 文件名 —— 排版时做文档分组表头 | metadata_helper.py:16 |
MetadataChunk | abs_idx | 该片段在文档里的绝对序号 —— 定位是第几段 | metadata_helper.py:87 |
MetadataChunk | shorter_id | 短 id 字段(与 MetadataFlags.SHORTER_ID 同值) | metadata_helper.py:86 |
MetadataNode | score | 检索相关度 —— 排版时按它排序/裁剪 | metadata_helper.py:57 |
MetadataNode | correlation_id | 关联 id —— 把引用绑回本次请求/回合 | metadata_helper.py:63 |
这些锚点是怎么挂上节点的。 Document.from_node(types.py:65)从 NodeWithScore 造
Document 时,把 artifact_id、shorter_id 从 node.metadata 里取出、并补上 score;
Document.metadata(types.py:136)反过来保证 artifact_id/shorter_id 一定回写进元数据 dict。
于是无论排版还是解析,都能从一个 Document 上稳定地读到这几个坐标。
排版时锚点就在用。 比如 XML 策略把 artifact_id 直接写进文档头、按 abs_idx 给同一文档内
的片段排序,让模型读到连贯的原文(format.py:165、format.py:217);token 超限时按 score
从高到低裁剪(format.py:148)。
关键区分:哪些元数据给模型看、哪些藏起来。 MetadataHelper.exclude_metadata
(metadata_helper.py:97)会把 artifact_id、score、各种 flag 加进
excluded_llm_metadata_keys / excluded_embed_metadata_keys——它们是给管线定位用的,
不该出现在喂给模型的文本里,否则既浪费 token 又干扰模型。这条"锚点存在但对模型隐身"的设计,
是引用能精确定位却不污染 prompt 的前提。
3.5 对外表示:从 [A3F9] 到 <citation> 到前端
要解决的小问题: 模型吐出的还是裸文本 … 25% [A3F9].,怎么把它变成前端能渲染、能点击的
结构化引用?而且是在流式输出、文本一个字一个字往外冒的时候?
第一步:解析。 extract_citations_by_original_text(utils.py:245)是解析器。它扫描文本,
把每个合法的 [id](id 命中当前这批 documents 里某个 doc.id)替换成一个 <citation> 标签,
非法/未知的代号则清掉。构造标签的是 format_cite(utils.py:185):
# components/engines/citations/utils.py:185 —— format_cite
def format_cite(i, doc, index):
data = {
"id": doc.id, # 短 id,如 A3F9
"index": index, # 该引用在本轮的序号(去重后)
"artifact_id": doc.document_id, # 指回哪个文件
"source_id": doc.id_, # 指回哪个具体节点/片段
"correlation_id": doc.metadata.get(MetadataNode.CORRELATION_ID.value),
}
attrs = "".join(f" {k}='{v}'" for k, v in data.items() if v is not None)
return f"<citation{attrs}></citation>"
于是 … 25% [A3F9]. 变成 … 25% <citation id='A3F9' index='0' artifact_id='…' source_id='…'></citation>.
——代号被"升级"成 3.4 里那些定位锚点的完整快照,前端不再需要短 id 表也能直接跳源。
这里藏着几处工程细节(踩过的坑):
- 模型有时吐出全角括号
【】而非半角[],解析前先统一替换(utils.py:263)。 - 用一个 buffer 处理被反引号
`包裹或跨 delta 边界切断的半截引用,避免误伤(utils.py:270)。 - 遇到没闭合的引用(有
[无]),直接丢弃不输出,防止把半截标记漏给前端(utils.py:353)。 - 同一文档在一次回答里多次被引,
index只分配一次、复用同号(utils.py:394的citation_indices)。
第二步:流式地调它。 解析器是 纯函数,真正把它接进"边生成边处理"的是
process_citations(chat/processors/events/citations/citations.py:21)。它包住模型的事件流,
对每个文本 delta 累积 current_text,整体重新解析一遍,再只把新增的干净文本和新增引用
作为这次的 delta 发出去:
# components/chat/processors/events/citations/citations.py:54(节选)
result = await asyncio.to_thread(
extract_citations_by_original_text,
text=current_text, documents=current_documents, citation_indices=citation_indices,
)
cleaned_text, current_citations, citation_indices = result
delta_text = cleaned_text[len(send_text):] # 只发相对上次的增量
delta_citation = current_citations[len(send_citations):]
event.delta = TextDelta.from_citations(delta_text, delta_citation)
放进 asyncio.to_thread 是因为解析是 CPU 活,别阻塞事件循环。同一套逻辑对"思考"(thinking)
delta 也走一遍(citations.py:74)。
第三步:对外类型。 内部的 Citation(types.py:36,带 text/value/doc_id/artifact_id/source_id)
是管线用的,对外(API/前端)则转成 ZylonCitation(chat/extensions/citation.py:6)——一个
只保留 id/index/artifact_id/source_id 四个字段的 pydantic 模型。两个方向的转换是对称的:
# chat/extensions/citation.py:26 —— 内部 Citation ↔ 对外 ZylonCitation
@classmethod
def from_citation(cls, citation): # 内 → 外
return cls(id=citation.doc_id, index=..., artifact_id=..., source_id=...)
@classmethod
def to_citation(cls, zylon_citation): # 外 → 内(如从历史里读回)
return Citation(text="", doc_id=zylon_citation.id, ...)
第四步:随事件流出去。 引用和源在事件流里走两条并行的线:
- 引用挂在文本块上:
TextBlock.citations: list[ZylonCitation](events/models/_content_blocks.py:47)。 序列化时若为空则整个字段省略(_content_blocks.py:55的custom_model_dump),不给前端塞噪声。 - 源本身单独走
SourceBlock(_content_blocks.py:653)——它装的是list[SourceType](即被引的文档 chunk),是"引用编号 → 实际源内容"映射的载体。ChatService从内容块里 分别汇总sources(server/chat/chat_service.py:70)和citations(chat_service.py:90)两个属性。
闭环:下一轮怎么认得上一轮的引用。 多轮对话里,上一轮回答里已有的 <citation> 会被
CitationRequestInterceptor(server/chat/interceptors/citation_interceptor.py:26)在新一轮开始 前
从历史里抽回来:process_history_citations(utils.py:557)把历史中的源重建成 Document,
replace_citations_in_text(utils.py:588)把还认得的引用保留、认不得的换成 UNK 清掉——
防止模型凭空复用一个当前上下文里已不存在的旧代号。
4. 与 Anthropic citations 语义的对应
同一个目标,两种实现层次。 Anthropic 官方的 citations 是API 原生能力:你把文档作为
结构化 document 块传进去、开 citations.enabled,模型返回时会自动附上
cited_text + document_index + 精确的字符/页码定位(char_location/page_location),
引用由 API 保证、模型不需要自己写编号。
PrivateGPT 的方案是应用层"自建"同一套语义,以便对任何模型都能用、并能精细控制排版:
| 关切 | Anthropic 原生 citations | PrivateGPT 的对应实现 |
|---|---|---|
| 谁产生引用标记 | API 自动附加 | 模型按 citations.j2 协议在正文里写 [XXXX] |
| 引用如何编号 | document_index 由 API 给 | 短 id [A3F9] + 去重后的 index(format_cite) |
| 指回哪个源 | document_index | artifact_id(文件) + source_id(片段) |
| 引用的原文定位 | cited_text + start/end_char/page | abs_idx 片段序号 + 源块内容(粒度到 chunk) |
| 结果结构 | 响应里的 citation 对象 | <citation> 标签 → Citation → ZylonCitation |
代码里两套语义并存的证据。 PrivateGPT 的事件模型里同时定义了对齐 Anthropic 原生 schema 的
CitationCharLocation / CitationPageLocation / CitationsConfig 等类
(events/models/_content_blocks.py:83、:89、:100),用于承接支持原生 citations 的模型
返回的字符/页码定位;而 [XXXX] + <citation> 这一套(本章主线)则是不依赖模型原生能力的
通用回退与统一表示层。
一句话对应关系: Anthropic 把"答案锚回原文"做进了 API,PrivateGPT 把同一语义拆成
「进 prompt 时用 token 发号 → 出 prompt 时把号解析成带 artifact_id/source_id 的结构」,
从而在任意模型上复刻出"每句话可点回具体源片段"的可信度支柱。
5. 巧妙之处(可借鉴)
-
代号进/出不对称——省 token 又信息完整。 进 prompt 用最短的
[A3F9](1+4+1 字符); 出 prompt 时才"升级"成携带全套定位锚点的<citation>。模型只需抄 4 个字,前端却拿到完整坐标。 见format_llm_source_str(format.py:22)对format_cite(utils.py:185)。 -
短 id 优先用"独特术语",随机是兜底。 有意义的词比
A3F9更利于模型正确回引,还带 5 秒 超时保护、失败静默降级,不拖慢主流程。见init_nodes_with_shorter_ids(utils.py:136)+analyze_texts_with_timeout(utils.py:83)。 -
定位锚点"存在但对模型隐身"。
artifact_id/score等键始终挂在节点上供管线定位,却被exclude_metadata(metadata_helper.py:97)从喂给模型的文本里排除,兼顾精确定位与干净 prompt。 -
流式增量解析。 每个 delta 都对累积全文重解析、只发增量(
process_citations,citations.py:54), 配合 buffer + 未闭合引用丢弃,做到"边流边把裸代号变成结构化引用"而不吐半截标记。 -
跨轮引用防幻觉。
replace_citations_in_text(utils.py:588)把历史里认不得的旧代号换成UNK清除,堵住"模型复用一个已不在上下文里的编号"这个 RAG 常见坑。
6. 边界与局限
-
webpage 类型的引用暂被前端丢弃。
extract_sources_from_history里明确注释 "FE doesn't support citations of webpage types",把type == "webpage"的源过滤掉(utils.py:508)。 -
引用粒度到 chunk,不到字符。 自建方案的定位精度是"哪个文件、哪个片段(
abs_idx)", 不像 Anthropic 原生的char_location能精确到起止字符区间——后者需模型原生 citations 支持。 -
术语抽取只覆盖 4 种语言。
SUPPORTED_LANGUAGES仅 en/es/fr/de(term_extractor.py:10), 其它语言会退回随机短 id(功能不受损,只是代号不再"有意义")。 -
依赖模型守协议。
[XXXX]由模型手写,若模型写成双括号、放进代码块或代号打错,解析器 会尽力容错(全角替换、buffer、未知代号清除),但本质上引用的正确性仍部分依赖模型遵守citations.j2的约束。
7. 横向对比
同 shelf 的兄弟章节里,引用不是孤立的一环:
- 源从哪来:01-ingestion-tree-nodes.md 定义了
artifact_id等 本章赖以定位的元数据,是引用锚点的源头。 - 引用挂在哪些内容上:02-semantic-retrieval-tree-expansion.md 产出被引的片段(本章只用其结果,不重复其算法)。
- 引用如何随工具结果流出:04-agentic-rag-chat-loop.md 讲检索被封成
工具、
SourceBlock如何在 agent 循环里回传。 - 源块进 prompt 的分层:05-context-stack-assembly.md 讲
DocumentLayer等如何把带引用的源装进上下文栈(本章的CitationRequestInterceptor就往那个栈里塞源)。
8. 代码地图(导航索引)
| 主题 | 文件路径 | 关键符号 |
|---|---|---|
| token 包裹源、拼进 prompt | private_gpt/components/engines/citations/format.py | format_llm_source_str format_llm_source format_context _format_documents_as_list/xml/json |
| 起止 token 常量 | private_gpt/components/engines/citations/utils.py | ORIGINAL_START_TOKEN ORIGINAL_END_TOKEN |
| 短 id 生成 / 发号 | private_gpt/components/engines/citations/utils.py | generate_shorter_id init_nodes_with_shorter_ids |
| 术语抽取(独特词当代号) | private_gpt/components/engines/citations/term_extractor.py | TextAnalyzer get_unique_terms get_terms score_term |
| 引用数据模型 | private_gpt/components/engines/citations/types.py | Document(.id from_node) Citation |
| 元数据锚点 | private_gpt/components/ingest/metadata_helper.py | MetadataKeys MetadataChunk(ABS_IDX) MetadataNode(SCORE CORRELATION_ID) MetadataHelper.exclude_metadata |
把 [id] 解析成 <citation> | private_gpt/components/engines/citations/utils.py | extract_citations_by_original_text format_cite _extract_citations_from_text |
| 流式逐 delta 解析 | private_gpt/components/chat/processors/events/citations/citations.py | process_citations |
| 对外引用表示 | private_gpt/chat/extensions/citation.py | ZylonCitation.from_citation to_citation |
| 事件流载体 | private_gpt/events/models/_content_blocks.py | SourceBlock TextBlock.citations CitationCharLocation CitationPageLocation |
| 汇总 sources / citations | private_gpt/server/chat/chat_service.py | Completion.sources Completion.citations |
| 跨轮引用接续 | private_gpt/server/chat/interceptors/citation_interceptor.py | CitationRequestInterceptor |
| 引用协议(教模型) | private_gpt/components/prompts/templates/chat/guidelines/citations.j2 · prompts/prompt_builder.py | create_citation_guidelines |