数据截至 (上游 commit 4a030776a31a)
引用(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:137 |
format_llm_source / format_context | 把源块前缀 [id] 拼进 prompt | components/engines/citations/format.py:41 :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:246 |
process_citations | 在流 式事件里逐 delta 调上面那个解析器 | components/chat/processors/events/citations/citations.py:22 |
ZylonCitation | 引用的对外(API/前端)表示 | chat/extensions/citation.py:6 |
SourceBlock / TextBlock.citations | 引用随事件流出去的载体 | events/models/_content_blocks.py:656 :47 |
3. 核心原理
下面按"引用的一生"的四个阶段,逐个讲清楚。
3.1 特殊 token 包裹:让模型能"引用回来"
要解决的小问题: 怎么让模型在自由生成的文本里,精确地指回"就是刚才第 3 个源"?
思路。 给每个源一个唯一、极短、模型不会打错的代号,把这个代号用一对醒目符号包起来, 连同源内容一起放进 prompt,再明确告诉模型:"要引用时,原样写这个代号"。PrivateGPT 选的 包裹符号就是最普通的方括号 —— 起止 token 都只有一个字符,几乎不耗 token:
# components/engines/citations/utils.py:43
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:122):
- 数值模式(
numerical_shorter_citations开):直接f"{index:04}",即0000、0001…… - 默认模式:以
node_id为随机种子,生成 4 位大写字母+数字,如A3F9。用node_id作种子 保证同一节点每次得到同一个短 id(确定性),而不是每次随机。
# components/engines/citations/utils.py:122 —— 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:137,
在检索管线 workflows/retrieval/retrieval.py:211 被调用)在发随机号之前,先试着从每个源块里
抽一个只属于它、别的块没有的独特术语当代号:
# components/engines/citations/utils.py:154 —— 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:144)。
收尾一步很重要: 短 id 是临时给模型看的脚手架,不能污染节点自身的元数据展示。所以
发完号立刻把 shorter_id 字段加进 excluded_llm_metadata_keys(utils.py:162),
让它不会作为普通元数据再被塞进别处。
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:84)把它放进线程池并设 5 秒超时,超时或异常就返回空 dict,让 3.2 优雅退回
随机短 id。术语抽取还受 settings().data.enable_term_extractor 开关和 NLTK 依赖是否可用的保护
(utils.py:64、utils.py:32 的 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:88 |
MetadataChunk | shorter_id | 短 id 字段(与 MetadataFlags.SHORTER_ID 同值) | metadata_helper.py:87 |
MetadataNode | score | 检索相关度 —— 排版时按它排序/裁剪 | metadata_helper.py:58 |
MetadataNode | correlation_id | 关联 id —— 把引用绑回本次请求/回合 | metadata_helper.py:64 |
这些锚点是怎么挂上节点的。 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:246)是解析入口。它扫描文本,
把每个合法的 [id](id 命中当前这批 documents 里某个 doc.id)替换成一个 <citation> 标签,
非法/未知的代号则清掉(逐条扫描的机制 已抽成 citations/parser.py 的 CitationTextParser,
内部再委托 components/text_processing 的流式规则引擎,下面几处工程细节就落在这两处)。
构造标签的是 format_cite(utils.py:186):
# components/engines/citations/utils.py:186 —— 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 表也能直接跳源。
这里藏着几处工程细节(踩过的坑):
- 模型有时吐出全角括号
【】而非半角[],解析前先统一替换(parser.py:58)。 - 反引号
`包裹的引用先"拆包"再解析,半截引用靠增量处理器的 pending 缓冲跨 delta 续读,避免误伤(parser.py:52的规则栈)。 - 遇到没闭合的引用(有
[无]),停在need_more不再前进,最终留在 pending 里不输出,防止把半截标记漏给前端(components/text_processing/rules.py:57)。 - 同一文档在一次回答里多次被引,
index只分配一次、复用同号(parser.py:79的citation_indices)。
第二步:流式地调它。 解析器是纯函数,真正把它接进"边生成边处理"的是
process_citations(chat/processors/events/citations/citations.py:22)。它包住模型的事件流,
对每个文本 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:48)。 序列化时若为空则整个字段省略(_content_blocks.py:56的custom_model_dump),不给前端塞噪声。 - 源本身单独走
SourceBlock(_content_blocks.py:656)——它装的是list[SourceType](即被引的文档 chunk),是"引用编号 → 实际源内容"映射的载体。ChatService从内容块里 分别汇总sources(server/chat/chat_service.py:83)和citations(chat_service.py:103)两个属性。
闭环:下一轮怎么认得上一轮的引用。 多轮对话里,上一轮回答里已有的 <citation> 会被
CitationRequestInterceptor(server/chat/interceptors/citation_interceptor.py:26)在新一轮开始前
从历史里抽回来:process_history_citations(utils.py:406)把历史中的源重建成 Document,
replace_citations_in_text(utils.py:436)把还认得的引用保留、认不得的换成 UNK 清掉——
防止模型凭空复用一个当前上下文里已不存在的旧代号。