跳到主要内容

RAG 生成与 short-id 方括号引用系统

30 秒导读: 检索(前几章)把一堆相关 chunk 找了出来,但答案还是要 LLM 来写。 本章讲 R2R 怎么在「让 LLM 写答案」的同时,逼它给出可回溯的引用——办法是给每个 chunk 发一个 7 位的短 ID([abc1234]),塞进喂给模型的上下文里;模型引用哪句话就顺手打上这个方括号; 事后再用正则把这些方括号抓出来,精确映射回它对应的源 chunk。这样答案里的每个 [abc1234] 都能一键跳回它引用的原文。

本章聚焦「从检索结果 → 带引用的答案」这一段,承接 知识图谱检索与多源聚合 产出的 AggregateSearchResult。不涉及 agent 多轮工具循环,那是 Agentic RAG 与 Deep Research 的内容。

1. 这是什么(零基础也能懂)​

一句话定义: RAG(Retrieval-Augmented Generation,检索增强生成)= 先检索、再让 LLM 基于检索到的内容写答案。 本章讲的是其中「生成」这半段,外加一套引用溯源机制。

它解决什么问题。 普通 LLM 会「一本正经地胡说」——你没法知道它某句话是从哪来的。 RAG 把真实资料喂给模型能缓解幻觉,但还差一步:答案里到底哪句话对应哪份资料? 如果模型只是笼统地「参考了这些资料」,用户仍然无法核对。R2R 的做法是让引用精确到 chunk。

用起来什么样。 调用方给一个问题,拿回一个 RAGResponse:里面有生成的答案文本, 答案里夹着 [abc1234] 这样的方括号标记,外加一个 citations 列表——每个引用 ID 都附带它指向的那个源 chunk 的完整内容。前端渲染时就能把 [abc1234] 变成一个可点击的角标。

一句话直觉。 想象老师要求你写读书报告时必须在每句话后面标出处页码。 R2R 做的事一模一样:它先给每份「参考资料」编一个短号(chunk 的 UUID 前 7 位), 在资料上方写「Source ID [abc1234]」,然后要求 LLM「你引用哪段,就在句末打上那段的号」。 最后老师(代码)拿正则去数这些号,一一对回原始资料。

本节不涉及底层代码,只要记住:short-id = chunk 的身份证前 7 位 = 引用锚点。

2. 顶层全景(它大概怎么转)​

一次非流式 rag() 调用,数据这样流动(retrieval_service.py:985 rag):

┌─────────┐   ①检索    ┌──────────────────────┐
│  query  │──────────▶│ AggregateSearchResult │  (chunk/graph/web/doc 多源聚合)
└─────────┘            └──────────┬───────────┘
                                  │
              ②格式化(埋入 short-id)│  format_search_results_for_llm
                                  ▼
                      "Source ID [abc1234]:\n<chunk 原文>\n..."
                                  │
              ③组装 system+task 提示│  prompts_handler.get_message_payload
                                  ▼
                            ┌──────────┐
                            │   LLM    │  aget_completion
                            └────┬─────┘
                                 │ 生成的答案里夹着 [abc1234]
              ④抽取引用 ID(正则) │  extract_citations
                                 ▼
                     raw_sids = ["abc1234", ...]
                                 │
              ⑤逐个映射回源 chunk  │  _find_item_by_shortid(sid, collector)
                                 ▼
                     ┌──────────────────────┐
                     │      RAGResponse      │  answer + citations[]
                     └──────────────────────┘

怎么读这张图: 从上到下是一次请求的生命周期,命中即停。关键在 ② 和 ④/⑤—— short-id 在 ② 被「写进」上下文,在 ④/⑤ 被「读回」并还原成源对象。这两步一进一出,构成整个引用系统。

各部件一句话职责:

部件干什么文件:符号
rag()生成半段的总入口,分派流式/非流式retrieval_service.py:985 rag
format_search_results_for_llm把检索结果转成带 Source ID [xxxx] 的文本base_utils.py:29
id_to_shorthand取 UUID 前 7 位当短 IDbase_utils.py:25
extract_citations正则从答案里抠出所有引用 IDcore/utils/init.py:29
SearchResultsCollector把聚合结果拍平成 (类型, 对象) 列表,供反查base_utils.py:288
_find_item_by_shortid用短 ID 前缀匹配回源对象retrieval_service.py:1248
Citation / RAGResponse最终返回的引用与答案数据结构responses.py:36 / :138

主线走一遍(高层): 问题进来 → 检索出多源结果 → 把每条结果标上 short-id 拼成上下文 → 连同提示模板发给 LLM → LLM 写答案时按要求打 [short-id] → 代码正则回收这些 ID → 每个 ID 前缀匹配回它的源对象 → 打包成 RAGResponse 返回。

3. 核心原理(逐个机制,由浅入深)​

3.1 short-id 是怎么「发」出去的​

要解决的小问题: 怎么给 LLM 一个既简短、又能唯一指回某个 chunk 的引用锚点?

思路。 每个检索项(chunk / 实体 / 网页 / 文档)本身都有一个 UUID。 UUID 太长,不适合让模型反复誊写;但它的前几位基本就能唯一区分一次检索里的几十个项。 于是 R2R 取 UUID 的前 7 个字符当短 ID。

取前 7 位的函数极简(base_utils.py:25 id_to_shorthand):

def id_to_shorthand(id: str | UUID):
    return str(id)[:7]

这个短 ID 被写进上下文的哪里? 在 format_search_results_for_llm 里, 每个 chunk 上方都会印一行 Source ID [<短ID>]:,紧跟 chunk 原文(base_utils.py:47-50):

if results.chunk_search_results:
    lines.append("Vector Search Results:")
    for c in results.chunk_search_results:
        lines.extend(
            (f"Source ID [{id_to_shorthand(c.id)}]:", (c.text or ""))
        )

graph / web / doc 各源也照此办理(base_utils.py:53-126),所以每一类检索结果都带上了自己的方括号锚点。 喂给模型的上下文最终长这样(示意,非源码):

Vector Search Results:
Source ID [abc1234]:
爱因斯坦在 1905 年发表了狭义相对论……
Source ID [def5678]:
广义相对论则在 1915 年完成……

关键细节 / 坑: 短 ID 是 7 位,但引用正则(见 3.3)允许 7–8 位。 这不是 bug——回查时用的是「前缀匹配」(startswith),即使模型多写或格式化出 8 位也仍能匹配上第一个符合前缀的项。 但真正「发」出去的锚点始终是 7 位。

3.2 让 LLM 学会打引用:提示组装​

要解决的小问题: 光把 Source ID [abc1234] 放进上下文,模型不一定会用。得明确要求它引用时打这个标记。

思路。 rag() 用两段提示——system 提示(角色设定)+ rag 任务提示(把 query 和 context 填进模板,并交代「引用时请用 [ID] 格式」)。这两段由数据库里的 prompt 模板生成(retrieval_service.py:1031-1040):

system_prompt_name = system_prompt_name or "system"
task_prompt_name = task_prompt_name or "rag"
messages = await self.providers.database.prompts_handler.get_message_payload(
    system_prompt_name=system_prompt_name,
    task_prompt_name=task_prompt_name,
    task_inputs={"query": query, "context": context_str},
)

context_str 就是 3.1 里那段带 Source ID 的文本。上下文里的锚点 + 提示里的引用要求, 两者配合,模型才会在答案里吐出 [abc1234]。具体的引用措辞在 rag prompt 模板里(数据库/YAML,非本文重点)。

3.3 把引用「抠」回来:extract_citations 与正则​

要解决的小问题: 模型的答案是一大段自由文本,里面混着 [abc1234]。怎么可靠地把这些 ID 全抓出来?

思路。 一条正则:\[([A-Za-z0-9]{7,8})\]——匹配方括号里 7 到 8 个字母数字的串。 这条模式故意收得很紧,才能把真引用和普通方括号(如 [1]、[note]、[abc-123])区分开。

真实实现(core/utils/init.py:29 extract_citations,模式常量 :45):

def extract_citations(text: str) -> list[str]:
    if text is None or text == "":
        return []
    CITATION_PATTERN = re.compile(r"\[([A-Za-z0-9]{7,8})\]")
    sids = []
    for match in CITATION_PATTERN.finditer(text):
        sids.append(match.group(1))
    return sids

为什么是 7–8 位? 短 ID 发出去是 7 位;留一位宽容度,防止模型偶尔多带一个字符仍能被捕获。 测试用例把边界钉得很死(test_citations.py:246):[abc123](6 位,太短)、[abcdefghi](9 位,太长)、 [abc-1234](含连字符)一律不匹配;只有 [abc1234]、[abcd1234] 这类才算数。

去重发生在哪? extract_citations 本身不去重——同一个 ID 出现两次就返回两个(test_citations.py:649-653 明确断言保留重复与顺序)。 去重是在组装最终 citations 时按 ID 收敛的(见 3.4/3.5)。

3.4 非流式:一次性抠 ID、映射、打包​

要解决的小问题: 拿到完整答案后,怎么把每个引用 ID 变成一个附带源内容的 Citation 对象?

思路(三步): 抽取 → 逐个反查 → 组装。真实代码(retrieval_service.py:1052-1076):

llm_text = response.choices[0].message.content
raw_sids = extract_citations(llm_text or "")     # ① 抠出所有 ID
...
rag_resp = RAGResponse(
    generated_answer=llm_text or "",
    search_results=aggregated_results,
    citations=[
        Citation(
            id=f"{sid}",
            object="citation",
            payload=dump_obj(                        # ③ 序列化源对象为 payload
                self._find_item_by_shortid(sid, collector)   # ② 短 ID → 源对象
            ),
        )
        for sid in raw_sids
    ],
    ...
)

注意: 这里对 raw_sids 直接遍历,raw_sids 可能含重复 ID,所以非流式返回的 citations 列表每次引用出现都会生成一个条目(同一 ID 可重复出现)。真正「同一 ID 只带一次完整 payload」 的优化在流式路径里才有(见 3.5)。

collector 是一个 SearchResultsCollector,在检索后立即由 add_aggregate_result 灌入(retrieval_service.py:1026-1027):

collector = SearchResultsCollector()
collector.add_aggregate_result(aggregated_results)

它把 chunk / graph / web / doc 全拍平成 (source_type, result_obj) 的有序列表(base_utils.py:329 add_aggregate_result), 供下一步按短 ID 前缀反查。

3.5 流式:边生成、边检测新引用(SSE)​

要解决的小问题: 流式生成时,答案是一小段一小段吐出来的。怎么在 [abc1234] 刚被写完的那一刻 就发出一个「引用事件」,而不重复发、不漏发?

难点直觉。 token 是增量到达的:某一刻缓冲区是 …相对论[abc12,下一刻才补成 …相对论[abc1234]。 你不能对每个增量都重跑一遍全文正则然后把所有引用都发一遍——那会把老引用反复发出去。 需要一个**只报「新出现的引用」**的机制。这就是 CitationTracker + find_new_citation_spans。

span(跨度)= 去重的钥匙。 光看 ID 不够:同一个 [abc1234] 可能在答案里出现多次, 每次是不同位置。R2R 用 (start, end) 字符区间(span)来标识某一次具体出现。 extract_citation_spans 返回 {ID: [(start,end), ...]}(core/utils/init.py:55)。

CitationTracker 记两样东西(core/utils/init.py:90):

  • seen_citation_ids —— 见过哪些 ID(判断「这个 ID 是不是第一次出现」)。
  • processed_spans —— 每个 ID 已处理过哪些 span(判断「这一次出现是不是新的」)。

is_new_span 一个方法既查又标记(core/utils/init.py:129):

def is_new_span(self, citation_id, span):
    if citation_id is None or citation_id == "" or span is None:
        return False
    if citation_id not in self.processed_spans:
        self.processed_spans[citation_id] = set()
    if span in self.processed_spans[citation_id]:   # 见过 → 不是新的
        return False
    self.processed_spans[citation_id].add(span)      # 没见过 → 记下并判为新
    return True

find_new_citation_spans 就是「对累计缓冲区跑正则,只留 tracker 认为是新的那些 span」(core/utils/init.py:178):

all_spans = extract_citation_spans(text)             # 全文所有 span
new_spans = {}
for cid, spans in all_spans.items():
    for span in spans:
        if tracker.is_new_span(cid, span):           # 只保留新 span
            new_spans.setdefault(cid, []).append(span)
return new_spans

流式主循环怎么用它。 sse_generator(retrieval_service.py:1080)对每个 LLM 增量 delta.content:

  1. 追加到 partial_text_buffer(retrieval_service.py:1127);
  2. 调 find_new_citation_spans(partial_text_buffer, citation_tracker) 找这一刻新冒出来的引用(retrieval_service.py:1132);
  3. 对每个新 span:用 is_new_citation(cid) 判断这个 ID 是否首次出现(retrieval_service.py:1140);
  4. 只有首次才 _find_item_by_shortid 反查源对象、dump_obj 成 payload 并缓存(retrieval_service.py:1148-1159);
  5. 发一个 citation SSE 事件:带 id / is_new / span;is_new=True 时才附完整 payload(retrieval_service.py:1162-1182)。

这就是 Citation 模型注释说的**「首次带全量 payload,后续只带 ID 和位置」**(responses.py:40-41)——省流量。

收尾: LLM 发出 finish_reason == "stop" 时(retrieval_service.py:1185), 用 citation_tracker.get_all_spans() 把每个 ID 的所有 span 汇总,连同缓存的 payload 组成 consolidated_citations,发一个 rag.final_answer 事件,再发 done(retrieval_service.py:1186-1227)。

流式引用的时间线(示意):

增量到达:  "相对论"   "[abc"     "1234]是"   "…[abc1234]又"      finish=stop
buffer:    相对论    相对论[abc  相对论[abc1234]是  …两处[abc1234]
新 span?    无        无(未闭合)  ✔ (10,19) 新    ✔ 第二处新位置     —
首次 ID?    —         —          ✔ 发 payload    ✘ 只发 ID+span     —
事件:      message   message     citation(全)   citation(轻)      final_answer + done

3.6 反查:short-id 如何映射回具体检索项​

要解决的小问题: 手里只有一个 7 位前缀 abc1234,怎么在几十个检索项里找到它对应的那个?

思路:前缀匹配。 遍历 collector 里所有 (source_type, result_obj),看谁的完整 id 以这个短 ID 开头。 真实实现(retrieval_service.py:1248 _find_item_by_shortid):

def _find_item_by_shortid(self, sid, collector):
    for source_type, result_obj in collector.get_all_results():
        if getattr(result_obj, "id", None) is not None:
            full_id_str = str(result_obj.id)
            if full_id_str.startswith(sid):          # 前缀命中
                if source_type == "chunk":
                    return result_obj.as_dict()      # chunk 转成 dict
                else:
                    return result_obj                # 其它类型原样返回
    return None

为什么用 startswith 而不是相等? 因为发出去的是 UUID 前缀,不是完整 UUID。 短 ID 与完整 id 之间天然是「前缀」关系,前缀匹配正是这套设计的闭环点—— 3.1 用 [:7] 截前缀发出去,这里用 startswith 认前缀收回来。

还有一个更全面的兄弟方法: collector.find_by_short_id(base_utils.py:485)。 它做同样的前缀匹配,但额外处理了 dict 形态的结果、并会下钻进 document 内部的 chunks 找(base_utils.py:512-535)。 rag() 走的是精简版 _find_item_by_shortid;find_by_short_id 供其它调用点(如 agent 工具)使用。

3.7 辅助入口:completion() 与 embedding()​

同一个 RetrievalService 还暴露两个更底层的入口,给不需要完整 RAG 管线的调用方直接用:

方法作用文件:符号
completion裸调 LLM,给一组 messages 拿补全,不检索、不引用retrieval_service.py:963 completion
embedding把一段文本编码成向量retrieval_service.py:977 embedding
async def completion(self, messages, generation_config, *args, **kwargs):
    return await self.providers.llm.aget_completion(
        [message.to_dict() for message in messages], generation_config, *args, **kwargs
    )

async def embedding(self, text: str):
    return await self.providers.completion_embedding.async_get_embedding(text=text)

它们是 rag() 内部所依赖能力(LLM 补全、向量化)的对外直通版本,不参与 short-id 引用系统。

4. 巧妙之处(可借鉴的技术)​

  • 用「已有的 UUID 前缀」当引用锚点,零额外状态。 不需要单独维护一张「引用号 → chunk」的映射表—— 短 ID 就是 chunk id 的前缀,发靠 [:7](base_utils.py:25),收靠 startswith(retrieval_service.py:1259),天然闭环。

  • 正则收得紧,把「真引用」从「普通方括号」里择出来。 {7,8} 位 + 纯字母数字的约束(core/utils/init.py:45), 让 [1]、[TODO]、[abc-123] 都不会被误判为引用(test_citations.py:242-252 佐证)。

  • span 级去重,而非 ID 级。 流式下同一 ID 多次出现要各报一次位置,但每个精确位置只报一次—— (id, span) 组合做键(core/utils/init.py:129),既不漏也不重。

  • 首次带全量、后续带轻量。 is_new 标志决定是否附 payload(retrieval_service.py:1173、responses.py:53), 长答案里同一来源被引十次,大 payload 也只在 SSE 流里传一次。

  • is_new_span 一个方法「查+标记」二合一。 查询即副作用地登记,省掉「先查再写」两步竞态(core/utils/init.py:99-100 注释特意点明这一设计)。

5. 边界与局限(诚实)​

  • 短 ID 碰撞风险(理论上)。 7 位十六进制前缀在一次检索的几十个项里几乎不会撞; 但 startswith 只返回第一个命中项(retrieval_service.py:1259),万一两个 id 共享同一 7 位前缀,后者会被错指。代码未做碰撞检测。

  • 非流式不做 ID 去重。 非流式 citations 列表按 raw_sids 逐次生成(retrieval_service.py:1071), 同一引用出现多次就有多个条目、payload 也重复;去重优化只在流式路径。

  • 模型不配合就没有引用。 整套机制依赖 LLM 真的按 [ID] 格式打标记; 若模型漏打、或把 ID 写错/写成别的格式,extract_citations 自然抓不到——引用是「尽力而为」,非强保证。

  • 反查失败会得到 None payload。 _find_item_by_shortid 找不到就返回 None(retrieval_service.py:1266); 非流式仍会为该 sid 建一个 payload 为空的 Citation(模型幻觉出一个不存在的 ID 时即如此)。

6. 横向对比(同组兄弟章)​

关切本章相关章
引用锚点从哪来chunk 的 UUID 前 7 位chunk 由 01-底层三路检索 产出
上下文里有哪些源chunk/graph/web/doc 聚合聚合结构见 03-知识图谱检索与多源聚合
多路结果如何排序进上下文直接用聚合顺序排序/RRF 见 02-检索策略与两级RRF
生成是否可迭代调用工具否,单轮生成多轮见 05-Agentic RAG 与 Deep Research

全景与阅读顺序见 index。

7. 代码地图(导航索引)​

主题文件路径符号
RAG 生成总入口(流式/非流式分派)py/core/main/services/retrieval_service.pyrag
非流式:抽取+映射+组装 RAGResponsepy/core/main/services/retrieval_service.pyrag(1043-1076)
流式 SSE 生成器py/core/main/services/retrieval_service.pysse_generator(1080)
短 ID 前缀反查源对象py/core/main/services/retrieval_service.py_find_item_by_shortid
裸 LLM 补全 / 向量化入口py/core/main/services/retrieval_service.pycompletion / embedding
引用抽取(正则)py/core/utils/init.pyextract_citations / CITATION_PATTERN
带位置的引用抽取py/core/utils/init.pyextract_citation_spans
流式增量新引用检测py/core/utils/init.pyfind_new_citation_spans
引用状态跟踪(去重)py/core/utils/init.pyCitationTracker / is_new_span / is_new_citation
UUID → 7 位短 IDpy/shared/utils/base_utils.pyid_to_shorthand
检索结果格式化为带 Source ID 的上下文py/shared/utils/base_utils.pyformat_search_results_for_llm
聚合结果拍平 + 短 ID 全量反查py/shared/utils/base_utils.pySearchResultsCollector / find_by_short_id
源对象序列化为 payloadpy/shared/utils/base_utils.pydump_obj
Citation / RAGResponse 数据结构py/shared/api/models/retrieval/responses.pyCitation / RAGResponse
行为佐证(单元测试)py/tests/unit/retrieval/test_citations.pyTestCitationExtraction / TestCitationUtils