从漏报率37%到4% 开源漏洞库Agentic RAG落地实战

问题背景：传统RAG做漏洞检索的核心痛点

我们团队维护的内部开源漏洞情报库，累计收录了120万+条CVE、CNNVD、CNVD官方漏洞条目，同时关联了POC脚本、修复方案、受影响版本范围等扩展数据，早期上线的传统RAG检索模块始终存在体验短板：用户输入模糊关键词比如「2024年Spring远程代码执行漏洞」时，如果向量库对应条目的标注关键词是「Spring Framework 5.x 反序列化RCE」，单次向量召回的Top20结果经常漏掉目标CVE-2024-38821条目。

上线初期的统计数据显示，传统RAG的漏洞检索漏报率高达37%，用户往往需要反复调整3-5次关键词才能找到目标漏洞信息，完全无法满足企业级漏洞应急响应的效率要求。我们尝试过单纯提升向量维度、扩大召回TopN数量等常规优化手段，最多只能把漏报率降到22%，同时响应时间涨到4s以上，依然达不到生产可用标准。

从根因上拆解，传统RAG在漏洞这个垂直领域会同时撞上三类语义鸿沟。第一类是命名鸿沟：普通用户习惯说「远程代码执行」「Spring漏洞」「Log4j炸了」，而官方条目使用的是 Remote Code Execution、Spring Framework RCE、CVE编号、CWE分类、CVSS向量串；中文口语、英文官方术语、厂商产品代号三者几乎不在同一个向量邻域。第二类是时间与版本鸿沟：「2024年那个Spring」「上周爆的Confluence」这类查询隐含了年份、版本区间、披露时间，而嵌入向量对数字和版本号并不敏感，5.3.0 与 5.3.31 在向量空间里可能非常接近，却对应「受影响」和「已修复」两种完全相反的结论。第三类是粒度鸿沟：一条CVE描述往往长达数百词，包含产品、组件、协议、攻击前置条件，单向量切片会把「Spring」「RCE」「2024」平均成一个模糊向量，关键信号被稀释。

我们也复盘过常见的「便宜优化」为什么不够：把向量从 768 维升到 1536 维，Recall@20 只提升 3 个百分点，QPS 却掉了一半；把 TopN 从 20 拉到 100，漏报降到 22%，但 Rerank 阶段算力上涨 5 倍，P95 延迟突破 4s，而且大量无关条目把真正相关的结果挤出前排；换成更大的 Embedding 模型同样只能再压 2–3 个点。本质问题不是「向量不够强」，而是单次检索的查询表达能力太弱——用户给的是一个模糊意图，系统却只给它一次命中机会。这也是我们最终转向「多轮、可决策、带领域约束」的 Agentic RAG 的直接原因。

方案对比：不同检索架构的实测数据

我们针对同一组1000条标注过的漏洞检索测试集，对比了三种方案的核心指标，最终选定了带领域约束的Agentic RAG架构作为落地方案：

1. 基线传统RAG：单次检索无Query改写，直接召回Top20向量结果再做Rerank，漏报率37%，平均响应时间1.2s，优点是速度快，缺点是完全无法处理用户输入关键词和库内标注词不匹配的场景。
2. 通用Agentic RAG：允许大模型自主决定检索次数，无领域规则约束，漏报率11%，平均响应时间7.8s，优点是漏报率大幅下降，缺点是大模型经常生成完全无关的检索Query，导致检索次数无限膨胀，响应时间超标。
3. 带漏洞领域约束的Agentic RAG：限制最大检索次数为3次，Query改写强制遵循漏洞领域规则，最终漏报率降到4%，平均响应时间2.1s，完全满足生产环境的性能和准确率要求。

测试集的构造尽量贴近真实应急场景：1000 条 query 中，40% 是「产品 + 漏洞类型 + 年份」的模糊口语（如「2024 年 Spring 远程代码执行」），25% 直接给 CVE/CNNVD 编号，20% 是「产品 + 版本号 + 现象」（如「Tomcat 9.0.70 PUT 方法打文件」），剩下 15% 是攻击链、组件别名、错别字与中英混杂。评测指标除了漏报率（目标 CVE 未出现在 Top50），我们还统计了 Hit@5、MRR、平均检索轮数、平均生成 query 数、P95 延迟与单次会话 token 成本，结果汇总如下：

方案	漏报率	Hit@5	MRR	平均检索轮数	P95 延迟	相对 token 成本
传统 RAG（Top20 + Rerank）	37%	0.51	0.43	1	1.2s	1.0x
传统 RAG（Top100 + Rerank）	22%	0.58	0.47	1	4.1s	3.2x
通用 Agentic RAG（无约束）	11%	0.79	0.71	4.6	7.8s	6.5x
领域约束 Agentic RAG（≤3 轮）	4%	0.91	0.86	2.1	2.1s	2.4x

通用 Agentic RAG 的问题非常典型：在抽样审计的 200 个坏 case 里，38% 出现「查询漂移」——模型从「Spring RCE」一路联想到「Java 反序列化历史」「Struts2 漏洞合集」，越搜越宽；27% 出现「自我重复」，只是把同一句 query 换个说法反复检索；另有 15% 触发了无意义的元提问（「请告诉我更多关于 CVE 的信息」）。这些 query 在向量库里召回的都是噪声，却被模型当成「新证据」继续推理，形成检索膨胀。领域约束版通过三件事把它按住：结构化决策输出（只允许 JSON）、白名单改写维度（产品官方名、别名、CWE、CVE 编号、年份、版本）、硬性终止与轮数上限。代价是放弃了一部分「开放式探索」能力，但在漏洞应急这种目标明确的任务里，这个取舍非常划算。

实操步骤与可运行代码示例

我们落地的方案核心逻辑是：大模型每次拿到当前轮次的检索结果后，自主判断是否已经覆盖用户的全部查询需求，如果未覆盖则生成符合领域规则的新Query发起下一轮检索，直到满足终止条件或者达到最大检索次数，最后把多轮召回的全部结果合并去重后送入Rerank模块排序输出。

整体链路可以拆成五个阶段：① 入口归一化（抽取产品名、版本、年份、CVE 编号）；② 多轮受控检索（Agent 决策 + 混合召回）；③ 结果合并去重（以 CVE 主键 + 标题指纹双重去重）；④ Rerank（领域 Cross-Encoder，特征包含版本命中、时间衰减、POC 是否存在）；⑤ 答案生成（强制引用 CVE 编号与受影响版本区间）。下面给出关键模块的可运行示例。

1. 漏洞领域专用检索决策提示词模板

我们设计了带强约束的提示词，完全避免大模型生成无效Query：

# language: python
DECISION_PROMPT = """你是漏洞检索决策助手，只输出 JSON，不要输出任何解释。
用户原始查询：{user_query}
已识别实体：产品={product}，版本={version}，年份={year}，CVE={cve_id}
当前已检索轮次：{round_idx} / 3
当前已拿到的检索结果（按相关度排序，最多展示前 10 条）：
{current_docs}

请严格按以下规则输出决策结果：
1. 如果当前结果已经覆盖用户查询的所有漏洞信息（产品、漏洞类型、年份/版本均命中），
   输出 {{"action": "stop", "reason": "..."}}。
2. 否则输出 {{"action": "retrieve", "queries": [...]}}，queries 中最多 2 条，
   且每条必须属于以下类型之一：
   - official_name：产品官方英文名（如 Apache Tomcat、Spring Framework）
   - alias：常见别名/历史名（如 Jenkins、Spring4Shell）
   - cwe：弱点类型（如 CWE-78 OS Command Injection、CWE-502 Deserialization）
   - cve_year：CVE 编号前缀（如 CVE-2024）
   - version_range：带版本区间的短语（如 "5.3.0 - 5.3.31"）
3. 禁止生成与漏洞检索无关的 query，禁止重复已有 query，禁止提问式句子。
4. 已达到第 3 轮时，无论是否覆盖，一律输出 stop。
"""

工程上有几个避坑点必须强调：第一，务必用 response_format={"type": "json_object"} 或结构化输出（function calling / JSON Schema）锁死格式，否则线上约有 5%–8% 的响应会夹带 markdown 代码块或自然语言，导致解析失败；第二，current_docs 不要把整段描述塞进去，只传 cve_id + 标题 + 受影响版本 + 标签，单轮上下文控制在 2k token 内，决策延迟可以从 1.4s 压到 0.4s；第三，提示词里要显式写出「已识别实体」，让模型基于抽取结果而不是凭语感改写，实测这一项就能让漂移 query 比例从 17% 降到 3%。

下面是一个最小可运行的 Agent 主循环（依赖 OpenAI 兼容 SDK，向量库与 Rerank 以函数占位）：

# language: python
import json, hashlib
from openai import OpenAI

client = OpenAI(base_url="http://localhost:8000/v1", api_key="sk-xxx")
MAX_ROUNDS = 3

def llm_decision(user_query, entities, docs, round_idx):
    prompt = DECISION_PROMPT.format(
        user_query=user_query,
        product=entities.get("product"),
        version=entities.get("version"),
        year=entities.get("year"),
        cve_id=entities.get("cve_id"),
        round_idx=round_idx,
        current_docs="\n".join(
            f"- {d['cve_id']} | {d['title']} | affected={d['affected']}"
            for d in docs[:10]
        ) or "（暂无）",
    )
    resp = client.chat.completions.create(
        model="qwen2.5-72b",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        temperature=0,
    )
    return json.loads(resp.choices[0].message.content)

def hybrid_search(query, top_k=20):
    # 向量召回 + BM25 召回，按 RRF 融合；生产环境替换为真实实现
    vec_hits = vector_store.search(query, top_k=top_k)
    bm25_hits = bm25_store.search(query, top_k=top_k)
    return rrf_fuse(vec_hits, bm25_hits, k=60)

def dedup(docs):
    seen, out = set(), []
    for d in docs:
        key = d.get("cve_id") or hashlib.md5(d["title"].encode()).hexdigest()
        if key not in seen:
            seen.add(key)
            out.append(d)
    return out

def agentic_retrieve(user_query, entities):
    merged, tried_queries = [], set()
    # 第 0 轮：直接用原始 query 检索，保证召回基线
    merged.extend(hybrid_search(user_query))
    for r in range(1, MAX_ROUNDS + 1):
        decision = llm_decision(user_query, entities, merged, r)
        if decision.get("action") == "stop" or r == MAX_ROUNDS:
            break
        for q in decision.get("queries", [])[:2]:
            if q in tried_queries:
                continue
            tried_queries.add(q)
            merged.extend(hybrid_search(q))
        merged = dedup(merged)
    return rerank(merged, user_query, entities)

几个量化经验供参考：把第 0 轮固定为「原始 query 直检」非常关键，纯 Agent 冷启动在编号类 query（如 CVE-2024-38821）上会多花 1 轮、延迟增加约 0.6s；RRF 融合相比「向量 + 向量 TopN 扩量」，在编号精确匹配场景下 Hit@5 提升约 9 个点；去重必须放在每一轮而不是最后统一做，否则 prompt 里会被重复条目占满，模型会误判「证据已充分」而过早 stop。

2. Query 改写的领域规则与白名单

光有提示词还不够，我们在模型输出之后再加了一层确定性后处理，对 query 做白名单校验和归一化，任何不合法的改写直接丢弃，而不是带着噪声去检索：

# language: python
import re

ALLOWED_TYPES = {"official_name", "alias", "cwe", "cve_year", "version_range"}
CVE_RE = re.compile(r"CVE-\d{4}-\d{4,7}", re.I)
CWE_RE = re.compile(r"CWE-\d{1,4}")

def sanitize_query(raw: str, entities: dict):
    q = raw.strip()
    if len(q) < 3 or len(q) > 80:
        return None
    if CVE_RE.fullmatch(q):                       # 纯编号：直接走精确字段
        return ("exact_cve", q.upper())
    if CWE_RE.search(q):                          # CWE 类型走标签字段
        return ("tag_cwe", CWE_RE.search(q).group(0).upper())
    if re.fullmatch(r"CVE-\d{4}", q, re.I):
        return ("tag_year", q.upper())
    if entities.get("version") and entities["version"] in q:
        return ("version_boost", q)
    if q.endswith("?") or "什么" in q or "如何" in q:
        return None                               # 拒绝提问式 query
    return ("vector", q)

这一层在我们的灰度日志里拦截了约 6.2% 的非法改写，主要是模型自己脑补的产品名（把「Spring」扩成「Spring Boot」导致召回偏移）、提问句和超长自然语言。另一个实战经验是维护一份产品别名表（约 8000 条），例如 Spring Framework ↔ spring-framework ↔ spring-core、Apache Log4j ↔ log4j-core ↔ Log4Shell，改写时优先从表里取官方名，命中率比让模型自由发挥稳定得多，别名相关 query 的漏报从 12% 降到 3%。

3. 领域感知 Rerank 与终止判断

多轮召回合并后通常有 40–80 条候选，通用 Cross-Encoder 只看文本相似度，会把「提到 Spring 但其实是其他年份的 CVE」排到很前。我们在 Rerank 阶段加入了三类领域特征做加权：版本区间是否真实命中（用 packaging.version 做区间判定，而不是字符串包含）、披露时间与 query 中年份的距离、以及是否存在公开 POC（应急场景下用户通常更关心有 EXP 的条目）。

# language: python
from packaging.version import Version

def version_hit(user_ver: str, affected: str) -> float:
    # affected 形如 ">=5.3.0, <5.3.32"，简化示例
    if not user_ver or not affected:
        return 0.0
    try:
        v = Version(user_ver)
    except Exception:
        return 0.0
    ok = True
    for clause in affected.split(","):
        c = clause.strip()
        for op in (">=", "<=", ">", "<", "=="):
            if c.startswith(op):
                target = Version(c[len(op):].strip())
                ok &= {">=": v >= target, "<=": v <= target,
                       ">": v > target, "<": v < target, "==": v == target}[op]
    return 1.0 if ok else 0.0

def domain_rerank(docs, query, entities, base_scores):
    for d, s in zip(docs, base_scores):
        boost = 0.0
        boost += 0.25 * version_hit(entities.get("version", ""), d.get("affected", ""))
        if entities.get("year") and str(entities["year"]) in d.get("published", ""):
            boost += 0.10
        if d.get("has_poc"):
            boost += 0.05
        d["final_score"] = s + boost
    return sorted(docs, key=lambda x: x["final_score"], reverse=True)

终止条件建议同时满足「模型决策 stop」和「覆盖度阈值」两道闸：覆盖度用规则计算（产品名命中 + 漏洞类型/CWE 命中 + 年份或版本至少一项命中即视为覆盖），当模型说 stop 但规则覆盖度不足时，强制再检索一轮，直到 3 轮上限。灰度数据显示，双闸策略相比纯模型决策，再压低漏报约 2 个百分点，而平均轮数只增加 0.15。

4. 上线前的避坑清单

• 编号类 query 不要走向量：CVE-xxxx-xxxx、CNNVD-xxxx-xxxxx 必须走精确字段检索，Embedding 对编号几乎无区分度，误走向量会让这类本应 100% 命中的 query 出现约 8% 的漏报。
• 版本号不要交给语义相似度：一律用结构化版本区间判定，字符串包含会把 5.3.1 错误匹配进 5.3.10–5.3.19。
• 限制每轮新增 query 数与总轮数：建议每轮 ≤2 条、总轮数 ≤3；放开到 5 轮时漏报只再降 0.8%，延迟却翻倍。
• 决策模型与生成模型解耦：决策用 7B/14B 小模型 + JSON 模式即可，延迟和成本远低于 70B 模型，准确率差距小于 1%。
• 缓存改写结果：以归一化后的 (product, version, year) 为 key 缓存 Agent 决策，热点应急 query（如 Log4Shell、Spring4Shell）的二次访问延迟可降到 300ms 以内。
• 监控检索漂移率：埋点记录每轮 query 与原始 query 的 Embedding 余弦相似度，低于 0.5 视为漂移并告警，这是发现 prompt 退化和模型版本漂移最灵敏的指标。
• 答案必须带证据引用：最终回答强制附上 CVE 编号、受影响版本、修复版本与数据来源，避免模型在多轮证据上「串台」，把不同年份的修复版本张冠李戴。

按上述方案全量上线后，我们在生产流量上连续观察了 30 天：整体漏报率稳定在 4% 左右，Top5 命中率 91%，平均检索轮数 2.1，P95 延迟 2.1s，单次会话相对成本约为传统 RAG 的 2.4 倍，但考虑到应急响应中「少一次漏报、少一轮人工翻库」的收益，这笔投入在企业安全场景下是完全值得的。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。