RAG回答服务设计:受约束提示词、引用校验与双重拒答机制详解

导读:RAG 系统检索到证据之后,最关键的一步是把证据交给大模型生成回答。但大模型天生爱"自由发挥"——你给它三条证据,它可能用自己的知识补充一堆文档里没有的内容,甚至编造政策条款和引用编号。本文从实际政务政策 RAG 项目出发,完整拆解 RAG 回答服务的设计:如何用 dataclass 定义结构化返回、如何构造带编号的证据上下文、如何用四条规则约束大模型只依据证据回答、如何用正则表达式校验引用编号的合法性、如何实现"证据不足拒答"和"无有效引用拒答"的双重防线。文末附完整可复用的代码和验证脚本。

适合读者

  • 正在搭建 RAG 系统、需要让大模型基于检索证据生成可靠回答的开发者
  • 遇到大模型"幻觉"问题、想通过提示词工程和后校验机制解决的工程师
  • 需要在回答中实现引用标注(如 [1][2])并校验引用合法性的技术人员
  • 想了解 RAG 回答服务完整架构设计的同学

阅读收益

  • 掌握 RAG 回答服务的完整 10 步流程设计
  • 理解受约束提示词的 4 条核心规则及为什么这样设计
  • 学会用正则表达式校验 LLM 输出中的引用编号并自动清理无效引用
  • 理解双重拒答机制(证据级 + 引用级)的设计思路
  • 获得一套生产可用的 RAGAnswer 数据结构和 answer_question 完整实现

目录

  1. 问题背景:为什么需要RAG回答服务
  2. RAGAnswer:结构化回答的数据模型
  3. answer_question:10步流程全拆解
  4. 证据上下文组织:带编号的上下文构造
  5. 受约束提示词:四条规则锁死模型行为
  6. 大模型调用与temperature=0
  7. 引用校验:正则提取与无效引用清理
  8. 引用列表生成与最终拒答
  9. 双重拒答机制设计
  10. Trace调试信息设计
  11. 验证脚本:一次完整RAG回答测试
  12. 踩坑清单:回答服务中的8个关键问题
  13. 总结与延伸
  14. 文末互动

1. 问题背景:为什么需要RAG回答服务

1.1 检索到证据≠回答可靠

前面几章完成了完整的检索链路:

向量检索 → BM25检索 → RRF融合 → 重排序 → 证据过滤

系统已经能从知识库中找到相关证据。但用户最终看到的不是一堆证据片段,而是一段清晰、可靠、带引用编号的回答。这个"从证据到回答"的转换,正是 RAG 回答服务的职责。

1.2 大模型的三个致命倾向

直接把证据拼接丢给大模型让它回答,会面临三个问题:

# 问题1:模型用外部知识补充
# 证据里只说了"社保卡申领需要身份证"
# 模型回答:"社保卡申领需要身份证原件、复印件和两张一寸照片"
# → "复印件和两张照片"是模型自己补充的,证据里没有

# 问题2:模型编造引用编号
# 证据只有3条,但模型在回答里标注了[5]
# → [5]指向的引用不存在,用户点击后报错

# 问题3:证据不足时模型仍然"自信回答"
# 检索到的证据与问题相关性很低
# 但模型仍然根据这些弱证据生成一段看似合理的回答
# → 误导用户

1.3 回答服务要做什么

RAG 回答服务的完整职责:

用户问题
  → 调用检索服务获取证据
  → 没有证据就拒答(第一道防线)
  → 有证据就组织带编号的上下文
  → 构造受约束提示词
  → 调用大模型生成回答
  → 校验引用编号合法性(第二道防线)
  → 返回回答、引用、耗时、调试信息

2. RAGAnswer:结构化回答的数据模型

2.1 为什么用dataclass而不是dict

很多 RAG 项目直接返回一个字典,字段名靠口头约定,调用方不知道里面有什么。用 dataclass 定义结构化的返回类型,好处是:

  • 字段类型明确,IDE 自动补全
  • 调用方不需要记字段名
  • 序列化到 JSON 时结构稳定
  • 后续扩展字段不影响已有逻辑

2.2 完整定义

from dataclasses import dataclass
from typing import Any

@dataclass
class RAGAnswer:
    answer: str                              # 回答文本,可能含 [1][2] 引用标记
    refused: bool                            # 是否拒答(证据不足或无有效引用时为True)
    citations: list[dict[str, Any]]          # 引用列表,每条包含chunk_id/filename/quote/score等
    trace: dict[str, Any]                    # 调试追踪信息,含检索过程和引用校验结果
    latency_ms: int                          # 回答总耗时(毫秒)
    token_usage: dict[str, Any]              # 大模型Token使用量

2.3 字段设计思路

字段类型用途为什么这样设计
answerstr回答正文直接展示给用户,可能含 [1][2] 角标
refusedbool拒答标记前端据此显示特殊样式,提示用户换问题
citationslist[dict]引用列表每条引用含原文快照,用户可点击查看来源
tracedict调试信息开发阶段排查"为什么拒答"“引用了哪条证据”
latency_msint耗时监控回答性能,超过阈值告警
token_usagedictToken用量成本统计和用量监控

关键决策citationslist[dict] 而不是 list[Citation](ORM 模型)。原因是回答服务不应该直接依赖数据库模型——它只负责生成回答,持久化是另一层的事。


3. answer_question:10步流程全拆解

3.1 函数签名

async def answer_question(
    session: Session,                        # 数据库会话(传给检索服务)
    question: str,                           # 用户问题
    filters: RetrievalFilters | None = None, # 检索过滤条件(如文档类型、地区)
    mode: str = "full",                      # 检索模式(full=完整检索,quick=快速检索)
    top_k: int | None = None,                # 向量检索候选数
    evidence_top_k: int | None = None,       # 最终证据条数
    min_score: float | None = None,          # 最低证据分数阈值
):

参数设计要点

  • filters 默认 None 而不是空对象——避免可变默认参数陷阱
  • top_k / evidence_top_k / min_score 都设为 None——在函数内部给默认值,避免调用方必须传
  • mode 用字符串而不是枚举——保持简单,后续可升级为 Enum

3.2 完整10步流程

async def answer_question(session, question, filters=None, mode="full",
                          top_k=None, evidence_top_k=None, min_score=None):
    started = time.perf_counter()
    filters = filters or RetrievalFilters()

    # 步骤1:调用检索服务
    retrieval = await retrieve(
        question, session, filters,
        top_k=top_k or 10,
        evidence_top_k=evidence_top_k or 5,
        min_score=min_score if min_score is not None else 0.5,
        mode=mode,
    )

    # 步骤2:证据不足时拒答(第一道防线)
    if retrieval.refused:
        return RAGAnswer(
            REFUSAL_TEXT, True, [], retrieval.trace,
            int((time.perf_counter() - started) * 1000), {},
        )

    # 步骤3:构造带编号的证据上下文
    context = "\n\n".join(
        f"[{index}] 文档:{item['metadata'].get('filename', '')},"
        f"章节:{item['metadata'].get('section', '')},"
        f"页码:{item['metadata'].get('page') or '无'}\n{item['content']}"
        for index, item in enumerate(retrieval.evidence, start=1)
    )

    # 步骤4:编写受约束提示词
    prompt = f"""你是政务政策知识库助手。必须遵守:
    1. 只能依据"政策证据"回答,不得使用外部知识补充政策内容。
    2. 每个包含政策事实、数字、日期、对象或条件的句子末尾标注引用编号,如[1]。
    3. 证据没有说明的内容,明确回答"资料未说明"。
    4. 不得编造政策名称、办理条件、部门、时间或引用。

    政策证据:
    {context}

    用户问题:
    {question}
    """

    # 步骤5:调用大模型生成回答
    try:
        answer, usage = await invoke_text(
            create_chat_model(temperature=0), prompt
        )
    except Exception as exc:
        raise RuntimeError(f"模型服务调用失败:{exc}") from exc

    answer = str(answer).strip()

    # 步骤6:校验引用编号
    answer, cited_indexes = validate_citations(answer, retrieval.evidence)

    # 步骤7:生成引用列表
    citations = []
    for index in cited_indexes:
        item = retrieval.evidence[index - 1]
        citations.append({
            "chunk_id": item["chunk_id"],
            "document_id": item["metadata"].get("document_id", ""),
            "filename": item["metadata"].get("filename", ""),
            "section": item["metadata"].get("section") or None,
            "page": item["metadata"].get("page") or None,
            "quote": item["content"][:500],
            "score": item["evidence_score"],
            "label": str(index),
        })

    # 步骤8:没有有效引用时拒答(第二道防线)
    if mode == "full" and not citations:
        answer = REFUSAL_TEXT
        refused = True
    else:
        refused = False

    # 步骤9:写入引用校验trace
    trace = {
        **retrieval.trace,
        "citation_validation": {
            "valid": len(cited_indexes) > 0,
            "cited_indexes": cited_indexes,
        },
    }

    # 步骤10:返回RAGAnswer
    return RAGAnswer(
        answer, refused, citations, trace,
        int((time.perf_counter() - started) * 1000), usage,
    )

下面逐步拆解每个关键步骤。


4. 证据上下文组织:带编号的上下文构造

4.1 为什么要带编号

大模型生成的回答需要标注引用来源。如果上下文里没有编号,模型可能自己编编号,或者引用的内容和证据对不上。给每条证据打上 [1][2][3] 的编号,模型就能在回答中用 [1] 指代第一条证据。

4.2 上下文格式

context = "\n\n".join(
    f"[{index}] 文档:{item['metadata'].get('filename', '')},"
    f"章节:{item['metadata'].get('section', '')},"
    f"页码:{item['metadata'].get('page') or '无'}\n{item['content']}"
    for index, item in enumerate(retrieval.evidence, start=1)
)

生成的上下文长这样:

[1] 文档:公共数据管理办法.pdf,章节:第三章 数据共享,页码:12
公共数据共享应当遵循依法依规、安全可控、统筹协调的原则...

[2] 文档:数据治理实施方案.docx,章节:第二章 治理体系,页码:5
建立跨部门数据治理协调机制,明确数据管理职责分工...

[3] 文档:公共数据管理办法.pdf,章节:第五章 监督管理,页码:28
对公共数据共享和治理情况进行定期评估...

4.3 设计要点

设计决策原因
编号从1开始(start=1引用编号是给用户看的,从1开始符合直觉
\n\n 分隔让模型明确区分不同证据,避免混淆
包含文件名/章节/页码模型在回答时可以引用具体来源位置
page or '无'页码可能为None(如Word文档),用"无"兜底
content原文不变不截断、不改写,保证模型看到完整证据

容易踩的坑:有些教程教你在上下文里加"请根据以下证据回答"这类引导语。不要加——引导语应该在提示词的规则部分写,上下文只放纯证据。这样换提示词时不需要动上下文构造逻辑。


5. 受约束提示词:四条规则锁死模型行为

5.1 完整提示词

prompt = f"""你是政务政策知识库助手。必须遵守:
1. 只能依据"政策证据"回答,不得使用外部知识补充政策内容。
2. 每个包含政策事实、数字、日期、对象或条件的句子末尾标注引用编号,如[1]。
3. 证据没有说明的内容,明确回答"资料未说明"。
4. 不得编造政策名称、办理条件、部门、时间或引用。

政策证据:
{context}

用户问题:
{question}
"""

5.2 四条规则逐条解析

规则1:只能依据证据回答

这是最核心的约束。没有这条规则,模型会用自己的训练知识"补充"证据中没有的内容。比如证据只说了"需要身份证",模型可能补充"还需要户口本和照片"——这些是模型从训练数据中学到的,可能过时、可能不适用于当前地区。

规则2:关键事实必须标注引用

要求模型在包含政策事实、数字、日期、对象或条件的句子末尾标注 [1] 这样的引用编号。这样用户可以点击引用查看原文,验证回答的可靠性。

注意"关键事实"的限定——不是每句话都要标引用,只标包含具体政策内容的句子。否则回答里全是 [1][2][3],可读性很差。

规则3:未说明的内容明确回答"资料未说明"

这条规则防止模型用"推测"来填补证据空白。如果用户问"社保卡补办需要多少钱",但证据里只说了补办流程没说费用,模型应该回答"资料未说明补办费用",而不是自己猜一个数字。

规则4:不得编造

最后一条是兜底——即使前面三条没管住,这条再强调一次:不得编造政策名称、办理条件、部门、时间或引用编号。

5.3 为什么不用system message

有些实现把约束规则放在 system message 里,把证据和问题放在 user message 里。这确实更规范,但实际测试中发现:

  • DeepSeek 等模型对 system message 和 user message 的约束力差异不大
  • 用单一 prompt 字符串更简单,调试时直接打印就能看到完整提示词
  • 换模型时不需要适配不同的 message 格式

如果你的模型对 system message 有特殊处理(如 Claude),可以拆成两条消息。


6. 大模型调用与temperature=0

6.1 为什么temperature设为0

answer, usage = await invoke_text(
    create_chat_model(temperature=0), prompt
)

RAG 回答服务要求确定性输出

  • 引用一致性:同一个问题+同一批证据,每次回答应该引用相同的证据编号。如果 temperature > 0,模型可能这次引用 [1][3],下次引用 [2][4],引用校验结果不稳定
  • 事实稳定性:政务政策回答不能有"创造性",0 度温度让模型选概率最高的 token,最大程度减少"自由发挥"
  • 测试可复现:验证脚本跑出来的结果应该稳定,方便对比改前改后的效果

6.2 invoke_text封装

async def invoke_text(
    model: BaseChatModel,
    prompt: str,
) -> tuple[Any, dict[str, Any]]:
    response = await model.ainvoke(prompt)
    if not isinstance(response, AIMessage):
        raise TypeError(f"聊天模型返回了不支持的消息类型:{type(response).__name__}")
    return response.content, dict(response.usage_metadata or {})

关键设计

  • 返回值是 (content, usage_metadata) 元组——文本内容和 Token 用量一起返回
  • usage_metadata or {} 防止某些模型不返回用量信息时报错
  • 类型检查 isinstance(response, AIMessage) ——防御性编程,有些模型可能返回非标准类型

6.3 create_chat_model模型工厂

def create_chat_model(temperature: float = 0, settings: Settings | None = None) -> BaseChatModel:
    settings = settings or get_settings()

    # Ollama 本地模型分支
    if settings.model_provider == "ollama":
        return ChatOllama(
            model=settings.ollama_model,
            base_url=settings.ollama_base_url,
            temperature=temperature,
        )

    # DeepSeek 云端模型分支
    if not settings.deepseek_api_key:
        raise RuntimeError("DEEPSEEK_API_KEY 未配置")

    return init_chat_model(
        model=settings.deepseek_model,
        model_provider="openai",
        api_key=settings.deepseek_api_key,
        base_url=settings.deepseek_base_url,
        temperature=temperature,
    )

为什么用工厂模式

  • 开发时用 Ollama 跑本地模型(免费、无网络依赖)
  • 生产环境切到 DeepSeek API(更强的推理能力)
  • 切换只需要改环境变量,代码不用动

6.4 异常处理

try:
    answer, usage = await invoke_text(create_chat_model(temperature=0), prompt)
except Exception as exc:
    raise RuntimeError(f"模型服务调用失败:{exc}") from exc

模型调用是最容易出错的环节——网络超时、API 限流、密钥过期、模型服务维护……用 try-except 包裹,把底层异常转成统一的 RuntimeError,带上原始异常信息(from exc 保留异常链),方便上层统一捕获和日志记录。


7. 引用校验:正则提取与无效引用清理

7.1 问题:模型可能标注不存在的引用

即使提示词明确要求"不得编造引用",大模型仍然可能:

  • 标注 [5],但证据只有 3 条
  • 标注 [0],但编号从 1 开始
  • 标注 [1][2][3],但实际只引用了第 1 条的内容

7.2 validate_citations完整实现

import re

# 正则:匹配文本里的引用标记,例如 [1] [2] [10],捕获括号里面的数字
CITATION_PATTERN = re.compile(r"\[(\d+)\]")

def validate_citations(answer: str, evidence: list[dict]):
    """
    校验回答里的角标引用标记 [数字],过滤无效引用;
    返回清理后的回答 + 合法引用编号列表

    :param answer: LLM输出的回答文本,里面带有 [1][2] 这类引用标记
    :param evidence: RAG检索出来的证据片段列表,长度N,合法引用编号是 1~N
    :return: (清理后的回答文本, 真正有效的引用id有序列表)
    """
    # 合法引用集合:evidence一共多少条,合法编号就是 {1,2,3,...,len(evidence)}
    valid = {index for index in range(1, len(evidence) + 1)}

    # 从answer文本中提取所有中括号里面的数字,转成整数集合
    cited = {int(value) for value in CITATION_PATTERN.findall(answer)}

    # 算出无效引用:回答中标注了,但不在合法编号范围内的数字
    invalid = cited - valid

    # 如果存在无效引用,把文本里对应的 [x] 全部删掉
    if invalid:
        for value in invalid:
            answer = answer.replace(f"[{value}]", "")

    # 返回:清理完的回答;同时取「既被引用、又合法」的编号,排序后返回list
    return answer, sorted(cited & valid)

7.3 校验逻辑图解

假设 evidence 有 3 条,合法编号 = {1, 2, 3}

模型回答:"...根据规定[1],需要提交申请[3],参考附件[5],编号[0]"

提取 cited = {0, 1, 3, 5}

invalid = cited - valid = {0, 5}    ← 不在 1~3 范围内

清理后回答:"...根据规定[1],需要提交申请[3],参考附件,编号"

有效引用 = cited & valid = {1, 3}   ← 既被标注又在合法范围内

返回:("...根据规定[1],需要提交申请[3],参考附件,编号", [1, 3])

7.4 设计要点

设计决策原因
用集合运算(-&集合运算天然去重且高效,O(n)
无效引用直接删除而非报错用户体验优先——回答仍然可用,只是少了错误标注
返回排序后的 list引用列表按编号排序,前端展示一致
正则 \[(\d+)\]只匹配纯数字,不误匹配 [注][a] 等非引用标记

容易踩的坑answer.replace(f"[{value}]", "") 会把文本里所有匹配的标记都删掉。如果回答里同时有 [5] 出现两次(如"参见[5],另外[5]也提到了"),两次都会被删除。这是期望行为——无效引用就应该全部清除。


8. 引用列表生成与最终拒答

8.1 引用列表生成

校验完成后,根据有效引用编号从证据列表中提取详细信息:

citations = []
for index in cited_indexes:
    item = retrieval.evidence[index - 1]    # 编号从1开始,索引从0开始
    citations.append({
        "chunk_id": item["chunk_id"],
        "document_id": item["metadata"].get("document_id", ""),
        "filename": item["metadata"].get("filename", ""),
        "section": item["metadata"].get("section") or None,
        "page": item["metadata"].get("page") or None,
        "quote": item["content"][:500],
        "score": item["evidence_score"],
        "label": str(index),
    })

8.2 每条引用包含什么

字段来源用途
chunk_idevidence[“chunk_id”]关联数据库中的文档分块
document_idevidence metadata关联文档级别信息
filenameevidence metadata展示给用户的文件名
sectionevidence metadata章节标题,定位引用位置
pageevidence metadata页码,PDF文档可跳转
quoteevidence content[:500]引用原文快照,截取前500字
scoreevidence_score证据检索分数,表示相关性
labelstr(index)引用编号,对应回答中的[1]标记

8.3 quote为什么截取500字

"quote": item["content"][:500]
  • 太短(如100字):用户看不到完整上下文,无法判断引用是否准确
  • 太长(如全文):引用列表过大,前端渲染慢,API 响应大
  • 500字:兼顾可读性和性能,足够展示一个完整的政策段落

8.4 最终拒答

if mode == "full" and not citations:
    answer = REFUSAL_TEXT
    refused = True
else:
    refused = False

这是第二道防线:如果大模型生成了回答,但经过引用校验后发现没有任何有效引用,说明模型的回答没有证据支撑——直接替换为拒答文本。

为什么只在 full 模式下拒答quick 模式是快速检索,可能证据质量不高但回答仍有参考价值。生产环境中 full 模式要求严格——没有引用的回答不能展示给用户。


9. 双重拒答机制设计

9.1 两道防线

用户提问
  │
  ▼
步骤2:证据不足拒答(第一道防线)
  │ retrieval.refused == True?
  │   → 返回拒答文本,refused=True
  │
  ▼ 步骤2通过
步骤3-7:生成回答 + 校验引用
  │
  ▼
步骤8:无有效引用拒答(第二道防线)
  │ citations 为空且 mode=="full"?
  │   → 替换为拒答文本,refused=True
  │
  ▼ 步骤8通过
返回正常回答

9.2 统一拒答文本

REFUSAL_TEXT = (
    "根据当前知识库资料,无法找到足够可靠的依据回答该问题。"
    "请补充政策名称、地区或办理事项。"
)

设计要点

  • 不说"我不知道"——这会让用户以为系统坏了
  • 明确告知"知识库资料不足"——引导用户理解是数据问题
  • 建议"补充政策名称、地区或办理事项"——给用户可操作的下一步

9.3 两种拒答场景的区别

场景触发条件trace 内容citations
证据不足拒答检索阶段 evidence 为空或全低于阈值检索 trace空列表
无有效引用拒答模型回答了但引用全被校验掉检索 trace + citation_validation空列表

第二种场景更值得警惕——检索找到了证据,但模型没有正确引用它们。可能的原因:

  • 提示词约束力不够,模型忽略了引用要求
  • 证据与问题相关性实际上不高,模型选择不用它们
  • 模型能力不足,无法正确标注引用

遇到这种情况应该检查 trace 中的 citation_validation.cited_indexes,看模型标注了哪些编号,再对照回答文本分析。


10. Trace调试信息设计

10.1 trace结构

trace = {
    **retrieval.trace,                      # 检索阶段的trace(查询改写、向量分数、BM25分数等)
    "citation_validation": {                # 新增引用校验结果
        "valid": len(cited_indexes) > 0,
        "cited_indexes": cited_indexes,
    },
}

10.2 trace的用途

# 在验证脚本中打印trace
print("检索调试:", {
    "original_query": result.trace.get("original_query"),
    "evidence_count": len(result.trace.get("final_evidence", [])),
    "citation_validation": result.trace.get("citation_validation"),
})

输出示例:

检索调试:{
    'original_query': '如何加强公共数据共享和治理',
    'evidence_count': 3,
    'citation_validation': {
        'valid': True,
        'cited_indexes': [1, 2, 3]
    }
}

10.3 设计要点

  • **retrieval.trace 展开检索阶段的 trace——回答服务的 trace 是检索 trace 的超集
  • 新增 citation_validation 字段——只记录回答阶段特有的信息
  • cited_indexes 保留原始编号列表——方便排查"模型引用了哪些证据"

为什么 trace 里不存完整回答文本:trace 是调试信息,回答文本已经在 RAGAnswer.answer 里了。重复存储浪费空间,而且 trace 可能被序列化到数据库 JSON 字段,太大会影响性能。


11. 验证脚本:一次完整RAG回答测试

11.1 check_rag_answer.py

import asyncio
import sys
from pathlib import Path

# 添加 backend 到路径
backend_dir = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(backend_dir))

from app.database import SessionLocal, init_database
from app.schemas import RetrievalFilters
from app.services.rag_service import answer_question


async def main() -> None:
    init_database()
    session = SessionLocal()

    question = "如何加强公共数据共享和治理"
    filters = RetrievalFilters()

    try:
        result = await answer_question(
            session=session,
            question=question,
            filters=filters,
            mode="full",
            top_k=6,
            evidence_top_k=3,
            min_score=0.2,
        )
    finally:
        session.close()

    print("=" * 60)
    print("问题:", question)
    print("是否拒答:", result.refused)
    print("耗时毫秒:", result.latency_ms)
    print("Token 使用:", result.token_usage)
    print("引用数量:", len(result.citations))
    print("-" * 60)
    print("回答:")
    print(result.answer)
    print()
    print("引用:")
    for citation in result.citations:
        print({
            "label": citation["label"],
            "filename": citation["filename"],
            "section": citation["section"],
            "score": round(float(citation["score"]), 4),
            "quote_preview": citation["quote"][:80],
        })
    print()
    print("检索调试:")
    print({
        "original_query": result.trace.get("original_query"),
        "evidence_count": len(result.trace.get("final_evidence", [])),
        "citation_validation": result.trace.get("citation_validation"),
    })


if __name__ == "__main__":
    asyncio.run(main())

11.2 运行结果示例

============================================================
问题: 如何加强公共数据共享和治理
是否拒答: False
耗时毫秒: 3420
Token 使用: {'input_tokens': 1850, 'output_tokens': 320, 'total_tokens': 2170}
引用数量: 3
------------------------------------------------------------
回答:
加强公共数据共享和治理需要从以下几个方面着手:

一、完善数据共享机制。建立跨部门数据共享协调机制,明确各部门数据管理职责分工[1]。

二、遵循共享原则。公共数据共享应当遵循依法依规、安全可控、统筹协调的原则[2]。

三、建立评估机制。对公共数据共享和治理情况进行定期评估[3]。

引用:
{'label': '1', 'filename': '数据治理实施方案.docx', 'section': '第二章 治理体系', 'score': 0.8723, 'quote_preview': '建立跨部门数据治理协调机制,明确数据管理职责分工...'}
{'label': '2', 'filename': '公共数据管理办法.pdf', 'section': '第三章 数据共享', 'score': 0.8156, 'quote_preview': '公共数据共享应当遵循依法依规、安全可控...'}
{'label': '3', 'filename': '公共数据管理办法.pdf', 'section': '第五章 监督管理', 'score': 0.7689, 'quote_preview': '对公共数据共享和治理情况进行定期评估...'}

检索调试:
{'original_query': '如何加强公共数据共享和治理', 'evidence_count': 3, 'citation_validation': {'valid': True, 'cited_indexes': [1, 2, 3]}}

11.3 验证要点

检查项期望结果异常说明
refusedFalse如果 True,说明证据不足或引用全被校验掉
citations 非空至少1条引用空说明模型没标注引用或标注全无效
回答含 [1] 等标记有引用编号没有说明提示词约束力不够
cited_indexes与citations的label对应不对应说明校验逻辑有bug
latency_ms合理范围超过10秒需检查模型服务状态
token_usage 非空有input/output tokens空说明模型没返回用量信息

12. 踩坑清单:回答服务中的8个关键问题

序号问题原因解决方案
1模型回答不含引用编号提示词约束力不够或模型能力不足检查提示词第2条规则;换更强的模型;降低temperature
2模型标注了不存在的编号模型"幻觉"validate_citations 自动清理无效引用
3同一问题每次引用编号不同temperature > 0 导致输出不确定设置 temperature=0
4回答正确但 citations 为空模型用文字描述了引用但没标编号提示词中强调"必须标注[数字]"
5证据充分但模型拒答式回答提示词第3条"资料未说明"被过度使用调整提示词措辞,限定"仅当证据完全未涉及时"
6answer类型不是strinvoke_text返回Any类型str(answer).strip() 强转
7模型服务超时网络问题或API限流try-except捕获并转为RuntimeError;上层加重试
8quote过长导致响应慢未截取引用原文item["content"][:500] 限制最大长度

13. 总结与延伸

13.1 核心设计回顾

用户问题
  → 检索服务(获取证据)
  → 第一道防线:证据不足拒答
  → 证据上下文组织([1][2][3] 编号)
  → 受约束提示词(4条规则)
  → 大模型调用(temperature=0)
  → 引用校验(正则提取 + 无效清理)
  → 引用列表生成(quote快照 + score + label)
  → 第二道防线:无有效引用拒答
  → 返回 RAGAnswer(answer + citations + trace)

13.2 架构全景

           answer_question()
                 │
    ┌────────────┼────────────┐
    │            │            │
  retrieve()   prompt     validate_citations()
    │            │            │
  检索链路     4条规则      正则提取
  向量+BM25    证据上下文    集合运算
  RRF+Rerank   temperature=0  无效清理
    │            │            │
    └────────────┼────────────┘
                 │
            RAGAnswer
         ┌───────┼───────┐
         │       │       │
      answer  citations  trace
      (含[1])  (quote)   (调试)

13.3 延伸方向

  • 流式输出:当前实现是一次性返回完整回答。可以改成流式输出(streaming),让用户看到打字效果。但流式模式下引用校验需要等回答生成完毕后才能做
  • 多轮对话:当前是单轮问答。加入对话历史后,需要把之前的问答也放进上下文,但要注意防止历史回答"污染"当前证据
  • 回答缓存:对于相同问题+相同证据的情况,可以缓存回答结果。但要注意文档更新后缓存失效
  • 回答质量评估:可以引入第二个模型评估回答质量(如 faithfulness、relevance),形成自动质量监控
  • 引用粒度细化:当前引用是分块级别。可以进一步定位到具体段落或句子,让引用更精准

14. 文末互动

思考题

  1. 当前实现中,validate_citations 用正则 \[(\d+)\] 提取引用编号。但如果回答文本中本来就有 [2024] 这样的年份信息,会被误识别为引用编号 2024。你会怎么修改正则或校验逻辑来避免这种误匹配?

  2. 第二道防线(无有效引用拒答)只在 mode=="full" 时触发。如果你是系统设计者,quick 模式下发现模型回答没有任何有效引用,你会选择直接展示回答、还是也拒答?两种选择各有什么利弊?

  3. temperature=0 保证了输出的确定性,但也意味着同一个问题永远得到同一个回答。如果用户连续问三次完全相同的问题,是否应该给三次不同的回答?如果要,怎么在保证引用一致性的前提下实现回答多样性?

如果本文对你有帮助

  • 点赞支持,让更多 RAG 开发者看到这篇内容
  • 收藏备用,实现回答服务时翻出来参考
  • 评论交流,说说你在 RAG 回答生成中遇到的最大坑是什么

作者的话:RAG 回答服务是整个 RAG 系统中"最后一公里"——前面十几章搭的检索链路再精巧,最终用户看到的还是这段回答。如果回答不可靠、引用对不上、该拒答时不拒答,用户对系统的信任就会崩塌。本文从受约束提示词、引用校验、双重拒答三个核心机制出发,把回答服务的设计拆成了 10 个步骤,每一步都有明确的决策依据。核心思路很简单:不信任大模型的输出,一切都要校验

Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐