实战工坊：把工单拆成时间线再检索，用 400 行代码实现排障 Agent 的状态化 RAG

排障类知识库有个很尴尬的属性：内容极其丰富，但利用率极低。一家公司可能积累了上万张已关闭的工单，每张都记录了完整的调查过程，可当你面对一个正在发生的新故障时，搜索引擎给你的仍然是几十条标题相似的条目，你只能一条条点开、翻到中间、找到那个和你当前状态吻合的段落。

问题不在检索技术不够先进，而在检索单元选错了。

一、把工单当文档，是排障检索的根本错误

先看一个具体场景。某台 Windows Server 在打补丁后无法加入域，工单里的调查过程是这样推进的：

1. 报错原文是「无法联系域控制器」；
2. 工程师先查了网络连通性，能 ping 通；
3. 再查 DNS 解析，正向解析正常，反向解析缺失；
4. 修好反向解析后仍然失败；
5. 最后发现是补丁重置了 Kerberos 相关的加密策略。

如果你把这张工单整体嵌入向量库，得到的是一个混合了五个阶段语义的平均向量。它和另一张同样无法加入域的工单相似度很高——但那张工单卡在第 2 步，根因是网卡驱动。

对新故障来说，真正有用的历史案例是：已经在你的当前状态上验证过、并且走向了不同结论的那些案例。所以查询的本质是「在第 4 步之后还失败，历史上有谁遇到过、根因是什么」，这是一个状态级查询，而整篇文档级的相似度根本无法表达它。

RAFT 这篇被 EMNLP 2026 工业赛道接收的论文给的答案很直接：与其想办法改善整篇文档的表示，不如换一个检索单元。

二、核心抽象：每条历史案例是一条状态链

论文把每个已关闭案例 h_i 重新抽象成一个五元组：

• ρ_i：审阅者对这张工单可提取性的判断
• {φ_k}：按时间顺序排列的时间线条目，这是关键
• r_i：已确认的根因
• a_i：已记录的解决步骤
• e_i：排障相关实体

每个时间线条目 φ_k 提炼的是原始消息序列中连续的一段，捕获调查的一个有意义的阶段。论文明确说明，语义分段不必与处理批次边界对齐，分割原则是「每次当前理解发生实质性更新」——症状首次报告、假设被添加、假设被丢弃、假设被确认、根因确认、方案验证，各开一个新条目。确认性质的消息（「好的我再试试」）不新开条目。

这样做的直接收益是时间线保持紧凑：K_i 远小于 T_i，但每条都携带「采取了什么行动、在验证什么假设、当前怎么理解问题」。

抽取本身用 worker–reviewer 两段式：worker 按有界批次顺序处理消息，把演进中的状态带进下一批，允许后续证据修正前期误判；全部批次处理完后由 reviewer 统一审阅，输出结构化结果。论文给的结构里，实体最多 25 个、每条时间线条目上限 4800 字符、根因和解决步骤同样上限 4800 字符，并且用校验逻辑强制 extractable 与 non_extractable_reasoning 两个字段的一致性——把「无法提取」也当成一个必须给出理由的正经输出。

三、动手实现

下面用 400 行左右的 Python 把这套检索层跑起来。为便于本地验证，向量部分用任意可替换的嵌入函数占位，其余全部是真实可运行的逻辑。

3.1 数据结构与抽取

先定义结构。注意时间线条目是独立可检索的一等公民，父案例只作为它的容器。

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class TimelineEntry:
    case_id: str
    index: int                 # 条目在时间线中的序号，从 0 开始
    text: str                  # 该阶段的动作 + 假设 + 当前理解
    embedding: Optional[list[float]] = None

@dataclass
class Case:
    case_id: str
    entries: list[TimelineEntry]
    root_cause: str = ""
    resolution: str = ""
    entities: list[str] = field(default_factory=list)

    @property
    def size(self) -> int:
        """案例在检索结果里占用的 token 估算量。"""
        payload = " ".join(e.text for e in self.entries)
        payload += self.root_cause + self.resolution
        return max(1, len(payload) // 3)   # 中文粗估，可换成 tokenizer

抽取阶段关键在于让模型输出结构而不是自由文本。用 JSON schema 约束，字段上限照抄论文的设定，可以让抽取结果稳定到可以直接入库：

EXTRACT_SCHEMA = {
    "type": "object",
    "properties": {
        "entities": {
            "type": "array", "maxItems": 25,
            "items": {"type": "string", "maxLength": 120},
        },
        "timeline": {
            "type": "array", "minItems": 1,
            "items": {"type": "string", "maxLength": 4800},
        },
        "root_cause": {"type": "string", "maxLength": 4800},
        "resolution_steps": {"type": "string", "maxLength": 4800},
    },
    "required": ["timeline", "root_cause", "resolution_steps"],
}

💡 提示：抽取质量的下限很敏感。论文的消融实验里，把索引阶段的模型从 gpt-5.4 换成 gpt-5.4-nano，合成基准上的 Case Hit 从 0.842 掉到 0.772。索引是一次性成本，这里省钱的性价比很低；推理侧用便宜模型才是合理的降级点。

3.2 条目级混合评分与 RRF

检索的第一层是给每个条目打分，而不是给案例打分。论文用语义相似度加 BM25 的混合评分，两者通过 Reciprocal Rank Fusion 融合：

def rrf_merge(rank_lists: list[list[str]], k: int = 60) -> dict[str, float]:
    """多路排名列表融合。k 是 RRF 平滑常数，论文未公开取值，60 是通用默认。"""
    scores: dict[str, float] = {}
    for ranks in rank_lists:
        for pos, key in enumerate(ranks, start=1):
            scores[key] = scores.get(key, 0.0) + 1.0 / (k + pos)
    return scores

RRF 只消费排名、不消费分数，因此不需要对向量相似度和 BM25 分数做任何归一化。这一点在换嵌入模型或换分词器时特别省心——你不需要重新标定权重系数。

排序完成后进入最关键的一步：贪心父案例提升。条目排序列表里常常出现大量同源条目，如果直接截断 Top-N，很可能 N 条全部来自同一张工单。

def retrieve(cases: dict[str, Case], entries: list[TimelineEntry],
             score_entry, budget: int, max_cases: int = 5):
    ranked = sorted(entries, key=score_entry, reverse=True)
    picked: list[tuple[Case, TimelineEntry]] = []
    seen: set[str] = set()
    used = 0
    for entry in ranked:
        if entry.case_id in seen:
            continue
        case = cases[entry.case_id]
        if used + case.size > budget:
            break                      # 预算耗尽即停，不跳过继续找
        seen.add(entry.case_id)
        used += case.size
        picked.append((case, entry))   # 案例 + 锚定条目一起返回
        if len(picked) == max_cases:
            break
    return picked

返回结构是「完整案例表示 + 触发匹配的那一条条目」。这一点比看起来重要：只给条目，模型会丢失上下文；只给案例，模型要自己找是哪一步匹配上的。两者同时给出，等于告诉模型「在这一点上，历史案例和你的处境一致」。

注意 break 而不是 continue——预算耗尽就停止，这与论文算法 1 的描述一致。上下文预算在论文里合成基准取 6000 token、真实 Jira 取 5000 token，可检索案例数上限 5。

3.3 可选的案例级图

如果条目级检索的召回还不够，可以叠加一层案例图。图是可选组件，构建方式是：把根因和解决文本拼起来作为链接文本，对每一对案例算同样的混合评分，每个案例连到 Top-k 个邻居，移除嵌入相似度低于 0.6 的边，然后对称化，边权用共享最近邻。

论文实测 k=3，并且测了 k=5 和 k=10，Case Hit 波动不超过 0.17 个百分点——说明这个参数不敏感，选 3 就够了。

图扩展的用法是：在 5 个案例槽位里留 1 个给种子案例的一跳邻居。

四、效果与抗噪性

论文在两个基准上验证。合成基准用 Microsoft Learn 的 Windows Server 排障文档生成 826 个案例，覆盖 Active Directory、组策略、远程桌面、Windows 安全、备份存储、网络等七类，每个案例平均 10.4 轮消息、2767 token。真实评估集从 Apache Cassandra、Hadoop、HBase、Spark 的公开 issue 里筛出 30 组人工审计过的重复单，配 570 个干扰项。

查询构造在三个进度点上：0%（只有初始症状）、30%、60%。

真实 Jira 上的方向性证据一致：0.833 / 0.840 / 0.895，对普通 RAG 分别领先 16.7、17.3、10.5 个百分点。

比平均值更值得琢磨的是抗噪数字。给查询注入一个无关轮次后，普通 RAG 在 60% 进度点暴跌 49.31 个百分点，条目级检索只掉 9.36——差距 39.94 个百分点。这背后是结构性的原因：无关内容混进文档向量会直接污染那个平均向量，而条目级检索里，无关轮次最多变成一条低分条目，其他条目不受影响。

另一个有意思的发现是匹配条目在时间线里的深度会随进度前移：0% 进度时匹配条目平均落在 9.1% 深度，60% 进度时落到 54%。这符合直觉——进度越靠后，你越需要在案例的中后段找参照。

还有个反直觉的结果：Fast-GraphRAG 在所有进度点都不如普通 RAG（0.421 对 0.673）。论文的解释是，当任务不需要层级知识检索时，复杂的实体图提取反而是有害的。这个结论值得记下来——图结构不是免费的，它有自己的适用边界。

五、落地时的三个判断

第一，先量清楚检索单元的必要性。 如果你的知识库本来就是结构化的（比如每条记录就是一条独立的报错码加处置），那状态化检索没有意义。只有当知识以多阶段调查过程的形式存在时，换检索单元才带来收益。

第二，图扩展放到最后。 论文的数字显示它在进度靠后时会转负，而构建成本是几何级的。先把条目级检索做到位，再评估是否值得加。

第三，索引侧的成本要提前算。 抽取每个案例需要一次或多次 LLM 调用，这是相对普通 RAG 的纯增量开销。论文的应对是支持增量更新：新案例独立抽取入库，不需要重建整库。但对上万张工单的历史库，第一次全量抽取的账单要提前评估——而它换来的是后续每次检索都精准匹配到你所在的那个状态。

⚠️ 注意：论文自己也列出三条局限——主评估是合成数据、真实 Jira 只有 30 组且无置信区间、只评估检索层而未评估端到端诊断成功率。迁移到自己的数据前，先在自己的历史工单上做一次小规模对照，不要直接相信这些数字。