RAG 工作流设计实战:用代码决定检索、重试和停止

很多 RAG Demo 的流程都很简单:把文档切块,向量检索,拼 Prompt,调用大模型生成答案。这个链路可以解释概念,也能回答一些局部问题,但一到真实企业文档场景,问题会很快变复杂。

例如用户问:“列出合同中卖方的全部义务,并说明哪些义务引用了外部标准。”这不是一次检索就能稳定解决的问题。系统至少要做三件事:先找到义务所在章节,再把列表补全,最后沿着“外部标准”“附件”“第 X 条”等引用继续追踪。每一步都可能失败,也都可能触发下一轮检索或解析。

这时候很多团队会想到“上 Agent”,让大模型自己决定下一步调用哪个工具。这个方向不是不能做,但对企业文档问答来说,第一版生产系统更应该先回答一个工程问题:谁决定流程继续,谁决定流程停止?

我的建议是:把决策写进一个可读的 Dispatcher,把重试写成有边界的反馈循环。LLM 可以提供信号,但不要让 LLM 独自掌握控制流。

先区分三种“Agentic RAG”

现在“Agentic RAG”这个词用得很宽,讨论前最好先拆开。

第一种只是营销说法。只要不是最朴素的“embed -> retrieve -> generate”,就被叫成 Agentic RAG。这个定义没有太多工程价值。

第二种是反馈驱动的 RAG。大模型不仅生成答案,还输出结构化反馈,例如“上下文是否完整”“是否发现新关键词”“是否还有未解析引用”。代码读取这些字段,再决定是否重新检索、重新解析或停止。这类系统看起来有“自我修正”能力,但控制权仍在程序里。

第三种是自治 Agent。大模型拿到一组工具,在每一步自己选择调用什么、用什么参数、下一步是否继续。它的灵活性更高,但成本、延迟、复现和审计压力也更大。

企业文档智能大多不需要一开始就走到第三种。文档类型、问题类型、检索方式和失败模式通常是可归纳的,用代码调度更容易稳定迭代。

Anthropic 在 2024 年 12 月发布的 Building effective agents 中也区分了 workflows 和 agents:前者通过预定义代码路径编排 LLM 和工具,后者由 LLM 动态决定流程。这个区分对 RAG 架构很有用。文档问答更像工作流问题,而不是完全开放的工具探索问题。

图片1.png

企业文档 RAG 真正需要循环的地方

一次性 RAG 的问题不在于“没有 Agent”,而在于它默认第一次检索就是对的。真实文档里常见的失败通常有几类。

第一类是解析不够。PDF 里有表格、扫描页、页眉页脚、跨页段落。如果解析层只拿到乱序文本,后面的检索和生成都会被污染。

第二类是检索范围不够。用户问的是“全部条款”“所有类别”“完整清单”,但检索只拿到排名最高的几个片段。模型可能给出看似完整的答案,其实漏了一半。

第三类是引用没有跟进。合同里的“按附件 A 执行”、论文里的“结果见表 3”、标准文档里的“详见 ID.AM-08”,都要求系统进行第二跳检索。

第四类是词汇不匹配。用户问“供应链风险”,文档里写的是“第三方依赖”“外部服务提供商”“vendor risk”。LLM 在阅读初始上下文后可能发现更贴近文档的词,但一次性 RAG 没机会把这些词放回检索层。

这些问题都适合用循环处理,但不能无脑循环。每一种循环都应该有明确的信号、触发条件、修复动作和停止条件。

Dispatcher 负责选模式,不负责猜答案

一个实用的 RAG 工作流可以把系统拆成两层:底层是解析、问题理解、检索、生成这几个“砖块”;上层是组合层,负责决定这次问题要启用哪些模式。

组合层里最关键的文件就是 decide.py 或类似模块。它不需要写得神秘,本质上就是一个基于结构化输入的路由函数。

输入可以是两类对象:

ParsedQuestion:问题解析结果,包括意图、关键词、章节提示、版面提示、是否要求完整列表。

DocumentProfile:文档画像,包括页数、是否有目录、是否疑似扫描件、是否有大量表格。

输出是一组开关:

def decide_pipeline_patterns(parsed, profile):

activations = {

    "toc_retrieval": False,

    "keyword_retrieval": True,

    "dense_retrieval": False,

    "two_hop_references": False,

    "listing_aggregation": False,

    "adaptive_parsing": False,

    "iterative_feedback": True,

}

if profile.has_usable_toc:

    activations["toc_retrieval"] = True

if parsed.intent in {"open_scoped", "open_corpus_wide"}:

    activations["dense_retrieval"] = True

if parsed.intent == "listing":

    activations["listing_aggregation"] = True

if parsed.retrieval.section_hint or parsed.retrieval.layout_hint:

    activations["two_hop_references"] = True

if profile.is_likely_scanned or parsed.requires_table_precision:

    activations["adaptive_parsing"] = True

return activations

这个函数的价值不是“聪明”,而是可维护。新同事读它,就能知道团队如何判断问题类型;线上某个问题答错了,也能补一条测试用例,验证该开的模式是否真的打开。

相比之下,如果把“下一步做什么”完全交给 Prompt,失败时排查会很困难。你不知道是模型没想到要追引用,还是工具描述没写清楚,还是某次随机采样选了另一条路径。

反馈循环要读结构化字段

循环不能只靠一句“请检查答案是否完整”。更稳的做法是让生成结果带上结构化反馈字段。答案是给用户看的,反馈字段是给系统看的。

一个文档问答结果可以包含这些字段:

class AnswerWithEvidence(BaseModel):

answer: str

citations: list[Citation]

complete_answer_found: bool

context_structured: bool

pending_references: list[str] = []

discovered_keywords: list[str] = []

confidence: float

caveats: list[str] = []

字段不一定要完全依赖 LLM。引用是否存在、引用页码是否合法、列表数量是否和文档里的数量提示一致,都可以做程序化检查。LLM 适合提供软信号,代码适合做硬校验。

几个常见触发规则可以这样设计:

complete_answer_found = False:扩大检索范围,或启用列表聚合。

context_structured = False:只重解析失败页,不重跑整份文档。

pending_references 不为空:按引用名、表号、条款号做第二跳检索。

discovered_keywords 不为空:保留原始关键词,再追加新词重检索。

列表数量不一致:放宽章节边界或补一次相邻页检索。

关键点是每个触发都有对应动作。不要把所有失败都丢给“再问一次模型”,否则循环只是把不确定性重复了一遍。

停止条件比重试条件更重要

做循环时,很多人只写“什么时候重试”,却没有认真写“什么时候停止”。这会带来两个问题:简单问题被过度处理,复杂问题在没有新增信息时继续烧 token。

至少需要三类边界。

第一是最大轮数。企业文档问答里,两到三轮通常已经能覆盖大部分补救动作。如果多个模式可能叠加,可以把总上限设成四轮,但不要让每个模式各自拥有无限预算。

第二是候选集稳定。当连续两轮拿到的候选片段完全相同,继续重试通常没有意义。系统应该停止,并返回当前最好答案,同时标记“已达到迭代边界”。

第三是置信度或质量下降。如果新一轮引入了更多片段,但答案开始偏离原问题,或程序化校验结果变差,就应该回退到历史最好结果。

一个简化版循环可以这样写:

def iterate_with_bound(initial_state, run_pass, adjust, max_iterations=3):

state = initial_state

history = []

best = None

for index in range(max_iterations):

    result = run_pass(state)

    history.append(build_iteration_record(index, state, result))

    if best is None or result.confidence > best.confidence:

        best = result

    if is_satisfactory(result):

        return best, history, False

    if not should_continue(history, result):

        return best, history, True

    state = adjust(state, result)

return best, history, True

这里有一个容易忽略的细节:adjust 只能改变一个明确变量。例如这轮只追加关键词,下一轮只追引用,或者只重解析某几页。一次调整太多东西,后面很难判断到底是哪一步带来了改进或退化。

防止查询漂移:新词只能追加,不能替换原问题

反馈循环常见的副作用是查询漂移。

用户问“供应链风险”,第一轮上下文里出现“第三方服务”,第二轮扩展到“vendor management”,第三轮又扩展到“outsourcing policy”。每一轮看起来都找到了新东西,但答案可能已经离原始问题越来越远。

解决办法是把原问题和原始锚点固定在状态里。新关键词只能追加,不能替换。生成阶段始终看见原始问题,而不是被扩写后的检索语句牵着走。

def expand_query_safely(parsed, new_keywords, max_keywords=15):

expanded = parsed.model_copy(deep=True)

originals = list(expanded.retrieval.anchor_keywords or [])

additions = [kw for kw in new_keywords if kw not in originals]

combined = originals + additions

if len(combined) > max_keywords:

    room = max_keywords - len(originals)

    combined = originals + additions[-max(room, 0):]

expanded.retrieval.anchor_keywords = combined[:max_keywords]

return expanded

生产环境里还可以加一层漂移检测:新候选片段至少要和原始关键词、章节提示或实体名有一定重合。如果完全没有重合,就不要继续扩大范围。

一个更贴近生产的例子

假设合规团队上传了一份安全框架文档,问题是:“列出 GOVERN 下的全部类别,并说明哪一项覆盖供应链风险。”

这个问题看起来只是一句话,但 Dispatcher 应该打开多个模式:

文档有目录,启用 TOC 检索,先定位 GOVERN。

问题要求“全部类别”,启用列表聚合,不能只取向量排名靠前的片段。

问题还问“哪一项覆盖供应链风险”,启用综合判断,对候选类别逐项比对。

如果某个类别描述引用其他章节,再启用二跳引用检索。

第一轮可能拿到 GOVERN 的开头和几个类别,但列表不完整。生成结果里的 complete_answer_found 为 False,列表校验也发现数量不足。系统不直接返回,而是扩大到 GOVERN 章节的相邻页和下级标题。

第二轮拿到完整类别列表,但供应链风险的证据只在一个引用说明里。pending_references 触发二跳检索,系统按引用编号或关键词继续找。

第三轮拿到引用页,答案闭合。此时 complete_answer_found = True,pending_references = [],循环停止。

这套流程不需要模型自由选择工具。模型只负责解析问题、生成答案和给出反馈信号;真正的模式选择和停止判断都在代码里。

审计记录不是附属品

企业 RAG 的答案不能只给一句“根据文档,结论如下”。至少要保留每轮迭代记录,方便用户、审核人员和工程团队追踪。

一条迭代记录可以很简单:

{

"iteration": 2,

"trigger": "pending_reference",

"action": "retrieve Table 3 and referenced pages",

"candidate_pages_before": [7, 8],

"candidate_pages_after": [7, 8, 9],

"confidence_before": 0.62,

"confidence_after": 0.84,

"stop_reason": "complete_answer_found"

}

这类记录有两个用途。业务侧能看到系统为什么多跑了一轮;工程侧能通过失败样本反推 Dispatcher 规则是不是漏开了某个模式。

如果后续要把系统部署成内部 API,建议把迭代历史、激活模式、候选文档 ID 和引用页一起写入日志。涉及长期运行的 RAG 服务时,可以部署在 Hostease 的服务器,按实际文档规模、并发、存储和网络条件确认配置,不要把测试环境的资源假设直接搬到生产。

哪些场景适合自治 Agent

Dispatcher 工作流不是要否定 Agent。自治 Agent 适合的是开放问题空间。

例如研究助理需要同时使用网页搜索、代码执行、计算器、数据库、文件系统和外部 API,而且每个任务的路径都不同。这时让模型动态规划工具调用有价值。

再比如旅行规划、数据探索、复杂故障排查,问题本身没有固定文档边界,也没有明确的“先目录、再检索、再生成”路径。自治 Agent 可以用更高成本换灵活性。

但企业文档问答通常不是这种问题。文档结构相对稳定,问题类型可以归类,错误模式能被测试覆盖。此时工作流的优势更明显:可复现、可审计、成本可控、团队能持续维护。

落地时可以按这份清单检查

如果你正在做企业文档 RAG,可以先检查以下问题:

1.是否有 ParsedQuestion,而不是把用户原句直接丢给检索?

2.是否有 DocumentProfile,能识别目录、扫描页、表格密度和页数?

3.Dispatcher 的每条激活规则是否有测试样例?

4.每个循环是否写清楚信号、触发、动作和最大轮数?

5.是否记录每轮候选集、触发原因和停止原因?

6.新增关键词是否保留原始锚点,避免查询漂移?

7.程序化校验是否覆盖引用、列表数量、页码和输出格式?

8.循环耗尽时,答案是否带有“不完整”或“已达到边界”的标记?

这些检查项比“是否使用 Agent 框架”更关键。RAG 系统真正进入生产后,稳定性通常来自这些朴素的工程边界。

结语

企业文档 RAG 的核心不是让大模型“更自由”,而是让系统在需要时能多走一步,在没有新增信息时能停下来。

Dispatcher 负责把团队经验写成规则:什么问题开目录检索,什么问题开列表聚合,什么问题追引用。反馈循环负责把生成结果变成下一步动作:补关键词、重解析、二跳检索,或者停止。LLM 仍然很重要,但它提供的是语义能力和反馈信号,不是无限制的流程控制权。

当这些边界写清楚以后,再讨论是否引入自治 Agent,才是一个可评估的架构选择。

©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成,浏览时请结合常识与多方信息审慎甄别。
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

相关阅读更多精彩内容

友情链接更多精彩内容