↓
  1. SecurityClaw 学习笔记/

02 RAG 引擎核心流程

·1774 字·4 分钟
安全研究 Security Securityclaw Agent Rag Retrieval Context

结论先说：RAG 在这里不是“长期记忆”，而是“证据注入系统” #

很多文章把 RAG 写成一句空话：给大模型外挂知识库。这个说法太粗了。

在 SecurityClaw 这类 Agent 系统里，RAG 真正承担的是三件事：

  1. 把历史证据和外部知识引入当前决策
  2. 在模型不知道事实时提供检索支撑
  3. 在向量检索失效时，保证系统还能降级工作

所以这篇不只是讲 store / retrieve / build_context 三个函数，而是要讲清楚：RAG 为什么是知识层，而不是状态层。

一、先看核心接口：RAG 引擎只有三件事 #

class RAGEngine:
    def store(self, text: str, category: str, source: str) -> str:
        ...

    def retrieve(self, query: str, k=5, category=None) -> list[dict]:
        ...

    def build_context(self, query: str, k=5) -> str:
        ...

这三个函数分别对应三个问题：

  1. store：什么知识值得留下
  2. retrieve：当前问题该找哪些证据
  3. build_context：找到的证据如何变成模型能消费的输入

二、store 解决的不是“保存文本”，而是“保存可检索证据” #

{
    "_id": sha256(text)[:32],
    "text": "...",
    "embedding": [0.1, 0.2, ...],
    "category": "anomaly",
    "source": "threat_analysis_skill",
    "timestamp": "ISO8601"
}

2.1 为什么 _id 用内容哈希 #

因为这一步真正想做的是去重后的证据库，不是原始日志仓库。

好处 #

  • 相同内容不会重复写入
  • 检索空间更干净
  • 降低向量存储冗余

代价 #

  • 同一文本在不同语境下可能被压成同一条
  • 去重粒度依赖文本规范化方式

这里的 tradeoff 很清楚：SecurityClaw 更在意“不要把重复事实无限灌进知识层”，而不是“保留每一次重复记录”。

三、retrieve 的重点不是向量检索，而是失败后的行为 #

flowchart LR
    Q[Query] --> E[Embed query]
    E --> K{kNN search}
    K -->|成功| R1[top-k semantic hits]
    K -->|失败| F[keyword fallback]
    F --> R2[text search hits]
    R1 --> C[merge / format]
    R2 --> C

很多 RAG 文章只讲“向量检索多么语义化”，但真实系统里更重要的问题是：

如果向量检索坏了，系统会不会直接瘫掉？

SecurityClaw 在这里做对的一点，就是给 retrieve() 留了 fallback：

  • 先做 kNN 搜索
  • kNN 异常时退回关键词检索

为什么这个降级比精度更重要 #

因为生产系统真正致命的问题不是“召回差一点”，而是“整个知识层失效”。

一个能降级的检索系统，比一个只在 happy path 下表现漂亮的检索系统更值钱。

四、build_context 真正解决的是上下文预算问题 #

很多人把 RAG 的最后一步写成“把结果拼起来就行”。其实不是。

def build_context(query: str, k=5) -> str:
    hits = retrieve(query, k)
    lines = ["Relevant evidence:"]
    for i, hit in enumerate(hits, 1):
        lines.append(f"{i}. [{hit['category']}/{hit['source']}] {hit['text']}")
    return "\n".join(lines)

这个函数表面上只是在格式化文本，但它实际上承担了三个控制任务：

  1. 限制召回数量：不是所有相关内容都能塞进 prompt
  2. 定义证据格式：模型看到的是摘要化证据，不是原始数据湖
  3. 压缩上下文预算：让检索参与决策，而不是淹没决策

这也是为什么我不喜欢把 RAG 叫“长期记忆” #

因为真正的长期记忆应该可恢复、可索引、可演化；而 build_context 的输出只是一次性的 prompt 证据片段。

更准确的理解是：

RAG 提供的是“临时注入的知识证据”，不是运行中的状态。

五、嵌入模型和聊天模型为什么必须分离 #

llm:
  ollama_model: qwen2.5:7b-instruct-q4_K_M
  ollama_embed_model: nomic-embed-text:latest

这不是“为了架构好看”，而是并发稳定性的需要。

如果不分离，会发生什么 #

  • 聊天和嵌入抢同一模型资源
  • 高并发下 embedding 请求卡死
  • Agent 看起来像是“推理慢”，其实是知识层阻塞了

分离之后的好处 #

  • 聊天模型负责 reasoning
  • 轻量 embedding 模型负责召回
  • 两者吞吐和成本模型不同，运行时更稳定

这个设计的本质，是把“推理成本”和“检索成本”从同一个瓶颈上拆开。

六、RAG 什么时候不该参与 #

这点非常重要，而且很多项目完全没讲。

RAG 不该在这些场景里强行参与：

6.1 问题是纯当前状态问题 #

例如：

  • 当前这轮 plan 到哪一步了
  • 当前工具刚返回了什么
  • 当前权限是什么

这些属于 state，不属于 RAG。

6.2 问题不需要历史证据 #

例如：

  • 一个确定性的 schema 校验
  • 一个固定的权限判断
  • 一个预算判断

这些应该由代码直接回答，而不是去检索。

6.3 检索成本高于收益 #

如果一次检索本身非常慢，或者返回内容高度噪音，强行注入只会污染 prompt。

所以更准确的原则是：

RAG 应该在“当前状态不足以支撑判断，但外部证据可能改变决策”时介入。

七、RAG 与 state / checkpoint 的边界 #

这三者最容易混：

层作用生命周期
State当前这轮运行中的结构化状态单次执行
Checkpoint可恢复的会话状态跨请求 / 跨重启
RAG长期可检索证据库长期积累

如果你把 RAG 结果直接塞进 checkpoint，问题会很快出现：

  • checkpoint 变大
  • 恢复变慢
  • 旧检索证据污染新问题
  • prompt 不断膨胀

SecurityClaw 把它们拆开，这一点是对的。RAG 应该保持为外部知识层，而不是运行状态的一部分。

八、这一篇真正要记住的框架 #

你可以把 RAG 压缩成这条链：

保存证据
  -> 检索证据
  -> 过滤证据
  -> 格式化证据
  -> 注入当前决策

其中最关键的不是“有没有向量数据库”，而是两点：

  1. 检索失败时能否降级
  2. 上下文注入是否受预算控制

这两点决定了 RAG 是系统能力，还是系统负担。

九、验证命令 #

pytest tests/test_rag.py tests/test_rag_fallback.py -v

如果要手动验证语义检索与降级：

python run_mock.py