RAG检索瓶颈如何破？SAG与OpenViking的结构化查询新思路

1. 从 RAG 的检索瓶颈说起：为什么需要 SAG 这套新思路

做 RAG 项目做久了，你一定会撞上同一堵墙：向量检索召回的片段看起来语义相似度很高，但真正喂给大模型之后，回答质量就是上不去。我最早做企业知识库问答的时候，用的就是最经典的embedding + 向量库 + top-k 召回这套组合，demo 阶段效果惊艳，一上真实业务就露馅——用户问"上季度华东区的退货率是多少"，向量检索给我召回了一堆讲"退货政策""区域划分"的文档段落，唯独没有那条真正包含数字的记录。

这个问题的根子在于：向量检索擅长的是"语义相似"，而不是"精确匹配"和"结构化聚合"。当你的问题里包含具体实体、数值、时间范围、聚合条件时，纯向量召回几乎必然翻车。业界这几年围绕这个瓶颈做了很多补丁，比如混合检索（BM25 + 向量）、重排序（rerank）、查询改写（query rewriting），但这些都是"在检索层打补丁"，没有触及一个更本质的问题：知识本身该以什么形态被组织和查询。

SAG 这个思路，我的理解是它把矛头指向了 RAG 的检索范式本身。传统 RAG 把知识切成 chunk，靠相似度找回来；而 SAG 更强调结构化、可聚合、可精确过滤的检索，把一部分本该由向量库承担的职责，交还给结构化查询能力。而 OpenViking 在这个语境里，通常被理解为一套承载这种思路的开源实现或框架方向，把 Agent、SQL、结构化知识组织这几件事串起来。热词里同时出现了rag瓶颈、rag检索增强、ontology rag、sql、agent，这其实已经把话题的边界画出来了：当 RAG 遇上结构化查询和 Agent 编排，会碰撞出什么新玩法。

这篇文章适合两类人看：一类是已经跑通过基础 RAG、正在被召回质量折磨的开发者；另一类是正在做 Agent 项目、需要让 Agent 能可靠地查询结构化数据的工程师。我会从 RAG 的真实瓶颈讲起，拆解 SAG 的核心思路，聊 OpenViking 这类方案可能的落地形态，再给出一套可以自己动手复现的实操路径。全程不堆概念，讲的是我踩过的坑和验证过的做法。

2. RAG 到底卡在哪：四类检索失败的真实复盘

在讲 SAG 之前，得先把 RAG 的病根说透。不然你只会觉得"又来个新名词"。我把这几年遇到的检索失败归成四类，每一类都对应一个具体的业务场景。

2.1 语义相似但事实无关：向量召回的经典误伤

最典型的一类。用户问"我们的退款到账要几天"，向量库召回了三段文档：一段讲"退款申请流程"，一段讲"退款政策适用范围"，一段讲"客服响应时效"。三段语义上都跟"退款"高度相关，但没有一段真正回答"几天到账"。原因很简单，embedding模型把"退款"这个主题词编码得很强，把"几天"这个关键限定词编码得很弱。向量空间里，主题相近的文本天然靠得近，而限定条件、数值、时间这些"约束信息"在稠密向量里几乎被稀释掉了。

我做过一个粗糙的统计：在一个 2000 条问答对的测试集上，纯向量 top-5 召回的"事实命中率"只有 60% 出头，剩下 40% 全是语义相关但答非所问的片段。这个数字在 demo 里看不出来，因为 demo 的问题往往很"标准"，一旦用户开始用自然口语提问，命中率断崖式下跌。

2.2 聚合类问题：向量检索天生不会算数

第二类更致命。用户问"哪个产品的投诉最多""本月销售额环比增长多少""有多少订单超时未发货"。这类问题的答案不存在于任何单个 chunk 里，它需要对一批记录做聚合运算。向量检索的机制决定了它只能"找相似的文本"，不能"对一批数据做 count、sum、avg、group by"。

我见过不少团队试图用"预先把聚合结果写成文档"来绕过这个问题，比如每天定时生成一份"销售日报"塞进知识库。这招在小规模、固定维度下能凑合，但用户的问题维度是发散的——今天问按区域，明天问按渠道，后天问按时间段交叉，你不可能预生成所有组合。聚合问题的本质是查询，不是检索，这是 RAG 架构层面绕不过去的坎。

2.3 多跳推理：单次召回撑不起链式问题

第三类是"多跳"。用户问"负责 A 项目的项目经理，他之前带过哪些项目"。这需要先查到 A 项目的负责人是谁，再拿这个人的名字去查他带过的项目。传统 RAG 是"一次召回、一次生成"，它没有"中间步骤"的概念。你可以在 prompt 里让模型自己拆解，但模型拆解出来的子查询还是得回到同一个向量库去检索，而向量库对"人名精确匹配"这件事并不擅长——它可能给你召回一堆名字相似的人。

多跳问题的核心诉求是可组合的查询能力：第一步的输出能作为第二步的输入，而且每一步都要精确。这恰恰是结构化查询（SQL）的强项，也是 SAG 思路里非常关键的一环。

2.4 权限与时效：被忽视的工程约束

第四类不是算法问题，是工程问题，但杀伤力最大。企业知识库往往有权限隔离：销售看不了财务数据，A 部门看不了 B 部门的文档。纯向量检索很难在召回阶段做细粒度权限过滤，你只能在召回后再过滤，但这时候 top-k 已经被无关内容占满了，过滤完可能一条不剩。时效性同理，用户问"最新的政策"，向量库可能召回三年前的旧版本，因为新旧版本语义几乎一样。

把这四类问题摆在一起看，你会发现它们指向同一个结论：RAG 需要的不只是"更好的向量模型"，而是"更多样的检索手段"。SAG 的价值就在这里——它不否定向量检索，而是把结构化查询、精确过滤、聚合运算这些能力补进来，让整个检索层变成一个"能选工具"的系统。

3. SAG 的核心主张：把"检索"升级成"可编排的查询"

SAG 这个词目前还没有一个被广泛接受的官方定义，但从它出现的语境和热词关联来看，我倾向于把它理解为Structured Augmented Generation或类似含义——在 RAG 的基础上，显式引入结构化查询能力。它的核心主张可以拆成三层。

3.1 知识分层：非结构化、半结构化、结构化各归其位

传统 RAG 把所有知识都拍扁成文本 chunk，这是最大的浪费。SAG 的第一个主张是按知识的形态分层存储：

这个分层不是拍脑袋，而是让每种数据用它最擅长的方式被查询。文档就该用语义检索，因为它没有固定字段；而订单表就该用 SQL，因为它的价值在于精确和聚合。硬要把订单表塞进向量库，等于让一个会计去做文学鉴赏，用错地方了。

3.2 查询路由：让系统自己决定"该查哪里"

分层之后，新的问题来了：用户提问时，系统怎么知道该走向量检索还是走 SQL？这就是 SAG 的第二个主张——查询路由（query routing）。通常的做法是让一个轻量模型或规则引擎先对问题做意图分类：

• 问题里出现"多少""几个""排名""环比"这类聚合词，路由到 SQL。
• 问题里出现具体实体名、编号、精确条件，优先走结构化查询。
• 问题偏概念解释、开放讨论，走向量检索。
• 复杂问题拆成多个子查询，分别路由，最后合并。

我在实际项目里用过"规则 + 小模型"的混合路由：先用关键词规则兜底（快、可控），再用一个微调过的小分类模型处理边界情况。纯靠大模型做路由不是不行，但延迟和成本都上去了，而且大模型偶尔会"想太多"，把简单问题复杂化。

3.3 结果融合：多路召回怎么拼成一份上下文

路由出去之后，多路结果要合并。这里有个容易被忽略的细节：不同来源的结果不能简单拼接，否则模型会被互相矛盾的信息带偏。我的做法是给每路结果打上来源标签和置信度，然后在 prompt 里明确告诉模型"以下数据来自结构化查询，优先级最高""以下来自文档检索，仅供参考"。实测下来，这种显式的来源标注能显著降低模型"张冠李戴"的概率。

SAG 的这三层主张，本质上是在回答一个问题：当知识库不再只是"一堆文档"，而是一个包含多种数据形态的系统时，检索该怎么设计。这个问题在纯文档场景下不明显，但一旦你的知识库接入了业务数据库、工单系统、CRM，它就变成了必须解决的问题。

4. OpenViking 可能的落地形态：Agent 与 SQL 怎么接

OpenViking 这个名字在热词里和agent、sql、ontology rag绑在一起出现，我判断它大概率是一套面向 Agent 场景的知识检索框架，核心是把结构化查询能力封装成 Agent 可以调用的工具。下面聊聊这类方案通常怎么落地，以及我实操中验证过的关键点。

4.1 把 SQL 查询封装成 Agent 的"工具"

Agent 要查结构化数据，最直接的方式是给它一个"执行 SQL"的工具。但这里有个巨大的坑：不能让模型直接生成任意 SQL 然后无脑执行。我见过太多 demo 直接LLM 生成 SQL -> 执行 -> 返回结果，在测试环境跑得飞起，一上生产就出事——模型可能生成DELETE、UPDATE，或者写出全表扫描把数据库拖垮。

我的做法是三层防护：

1. 只读账号：Agent 用的数据库连接必须是只读权限，从根上杜绝写操作。
2. SQL 白名单校验：解析生成的 SQL，只允许SELECT，禁止DROP、DELETE、UPDATE、INSERT，禁止多语句。
3. 强制 LIMIT 和超时：所有查询自动加LIMIT，并设置查询超时，防止慢查询拖垮库。

# 一个简化的 SQL 安全校验示例 import sqlparse FORBIDDEN = {"delete", "update", "insert", "drop", "alter", "truncate", "grant"} def is_safe_sql(sql: str) -> bool: parsed = sqlparse.parse(sql) if len(parsed) != 1: # 禁止多语句 return False stmt = parsed[0] stmt_type = stmt.get_type() # 返回 'SELECT' / 'UNKNOWN' 等 if stmt_type != "SELECT": return False lowered = sql.lower() return not any(kw in lowered for kw in FORBIDDEN)

这段代码不复杂，但能挡掉绝大多数危险操作。安全这件事，永远不要指望模型"自觉"，要靠工程手段兜底。

4.2 Schema 注入：让模型知道表结构长什么样

模型要生成正确的 SQL，前提是它知道有哪些表、哪些字段、字段什么含义。这就是 schema 注入。做法是把数据库的表结构（表名、字段名、类型、注释）整理成一段文本，塞进 prompt。但表一多，schema 本身就超长，token 成本飙升。

我的优化思路是按需注入：先用向量检索或关键词匹配，从所有表里筛出和当前问题最相关的几张表，只注入这几张表的 schema。比如用户问"订单相关"，就只注入订单表、订单明细表、用户表，而不是把整个数据库 50 张表全塞进去。这个"schema 检索"的步骤，本身就可以复用 RAG 的思路，挺有意思的一个闭环。

4.3 从自然语言到 SQL 的中间层：别一步到位

直接让模型"自然语言 -> SQL"一步到位，准确率在复杂问题上很难看。我推荐加一个中间层：先让模型输出结构化的查询意图（JSON），再由代码把意图翻译成 SQL。

{ "intent": "aggregate", "table": "orders", "metric": "count", "filters": [ {"field": "status", "op": "=", "value": "timeout"}, {"field": "created_at", "op": ">=", "value": "2024-01-01"} ], "group_by": ["region"] }

这样做的好处是：意图 JSON 可以被校验、被修正、被缓存，而且比裸 SQL 更容易调试。当查询出错时，你能清楚看到是"意图理解错了"还是"SQL 翻译错了"，定位问题的效率高很多。这个中间层是我踩了无数坑之后总结出来的，强烈建议你在做 text-to-SQL 时加上。

4.4 和向量检索的协同：谁先谁后

Agent 同时有向量检索工具和 SQL 工具，那先调哪个？我的经验是先路由，再执行，必要时串行。简单问题一次路由就够；复杂问题（比如"投诉最多的产品，它的说明书里怎么写的"）需要先 SQL 查出产品名，再用产品名去向量库检索说明书。这种"SQL 出实体，向量查内容"的串行模式，在多跳场景里非常实用。

5. 动手复现：搭一套最小可用的 SAG 检索链路

光讲思路不够，得能跑起来。下面这套链路我用 Python 搭过，依赖不多，适合自己动手验证。整体分四步：准备数据、建结构化查询、建向量检索、做路由融合。

5.1 数据准备：一份文档 + 一张表

先造点数据。文档部分放几段产品说明，结构化部分用 SQLite 建一张订单表（SQLite 零配置，适合练手）。

import sqlite3 conn = sqlite3.connect("demo.db") cur = conn.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY, product TEXT, region TEXT, status TEXT, amount REAL, created_at TEXT ) """) cur.executemany( "INSERT INTO orders (product, region, status, amount, created_at) VALUES (?,?,?,?,?)", [ ("A产品", "华东", "timeout", 199.0, "2024-03-01"), ("B产品", "华南", "done", 299.0, "2024-03-02"), ("A产品", "华东", "done", 199.0, "2024-03-05"), ("C产品", "华北", "timeout", 99.0, "2024-03-06"), ], ) conn.commit()

文档部分你可以用任意几段文本，塞进一个列表即可。真实项目里文档会走切分、embedding、入库，这里为了聚焦检索逻辑，先简化。

5.2 结构化查询工具：意图 JSON 转 SQL

按前面说的中间层思路，写一个把意图 JSON 翻译成 SQL 的函数。注意所有值都要参数化，防止注入。

def intent_to_sql(intent: dict) -> tuple[str, list]: table = intent["table"] metric = intent.get("metric", "count") filters = intent.get("filters", []) group_by = intent.get("group_by", []) select_part = f"{metric}(*)" if metric == "count" else f"{metric}(amount)" where_parts, params = [], [] for f in filters: where_parts.append(f"{f['field']} {f['op']} ?") params.append(f["value"]) sql = f"SELECT {select_part}" if group_by: sql += ", " + ", ".join(group_by) sql += f" FROM {table}" if where_parts: sql += " WHERE " + " AND ".join(where_parts) if group_by: sql += " GROUP BY " + ", ".join(group_by) sql += " LIMIT 100" return sql, params

这个函数只处理了最基础的聚合和过滤，真实场景要复杂得多，但骨架就是这样。关键点是参数化查询，永远不要用字符串拼接把用户输入拼进 SQL，这是 SQL 注入的经典入口。

5.3 向量检索工具：最简实现

向量部分我用一个极简的本地实现，不引入重型依赖。真实项目换成任意向量库即可。

import numpy as np def simple_embed(text: str, dim: int = 64) -> np.ndarray: # 仅用于演示，真实场景请用正经的 embedding 模型 vec = np.zeros(dim) for ch in text: vec[ord(ch) % dim] += 1 norm = np.linalg.norm(vec) return vec / norm if norm > 0 else vec def vector_search(query: str, docs: list[str], top_k: int = 2): qv = simple_embed(query) scored = [(np.dot(qv, simple_embed(d)), d) for d in docs] scored.sort(reverse=True) return [d for _, d in scored[:top_k]]

这个simple_embed纯属演示，别拿去生产用。它的作用是让你在没有 GPU、没有 API key 的情况下也能把整条链路跑通，理解数据怎么流动。

5.4 路由与融合：把两路结果拼成上下文

最后是路由。我用一个简单的关键词规则做演示，真实项目可以换成小模型。

AGG_KEYWORDS = ["多少", "几个", "排名", "最多", "环比", "统计", "总数"] def route(query: str) -> str: if any(kw in query for kw in AGG_KEYWORDS): return "sql" return "vector" def build_context(query: str, docs: list[str]) -> str: route_type = route(query) if route_type == "sql": # 这里应由模型生成 intent，演示中直接构造 intent = { "intent": "aggregate", "table": "orders", "metric": "count", "filters": [{"field": "status", "op": "=", "value": "timeout"}], "group_by": ["region"], } sql, params = intent_to_sql(intent) cur.execute(sql, params) rows = cur.fetchall() return f"[结构化查询结果] {rows}" else: hits = vector_search(query, docs) return "[文档检索结果] " + " | ".join(hits)

跑一遍build_context("华东区超时订单有多少")，你会看到它走了 SQL 分支，返回了聚合结果；而build_context("退款政策怎么规定的")会走向量分支。这就是 SAG 检索链路的最小闭环。先把这个跑通，再往上加复杂度，比一上来就搭大框架靠谱得多。

6. 实操中真正会咬人的几个坑

链路跑通只是开始，下面这些坑是我在真实项目里被咬过的，提前告诉你，能省不少时间。

6.1 意图识别的边界比你想的模糊

"最近卖得怎么样"这种问题，既像聚合又像开放讨论，路由规则很容易判错。我的处理方式是给路由加一个"不确定"分支：当规则和小模型都拿不准时，两路都查，把结果都塞给模型，让它自己判断用哪个。多花一点 token，换准确率，值。另外，路由错误要能被观测到——我会记录每次路由的决策和最终回答质量，定期回看，持续优化规则。

6.2 SQL 生成的时间字段是重灾区

模型生成 SQL 时，对时间条件的处理经常出错。用户说"上个月"，模型可能生成created_at > '2024-03-01'但忘了上界，或者把日期格式写错。我的做法是在 prompt 里显式给出当前日期和常用时间表达式的换算规则，并且在意图 JSON 里把时间条件规范化成明确的起止日期，而不是让模型直接写进 SQL。时间这东西，交给代码算，别交给模型猜。

6.3 向量库和数据库的数据一致性

如果你的结构化数据是实时更新的，而向量库里的文档是定期同步的，两者会出现时间差。用户可能从 SQL 查到"订单 X 已取消"，但从向量库检索到的文档还说"订单 X 处理中"。这种矛盾会让模型输出自相矛盾的回答。我的经验是在 prompt 里明确标注数据来源和时间戳，并告诉模型"结构化数据实时性更高，冲突时以它为准"。同时，向量库的同步频率要尽量跟上业务节奏。

6.4 别忽视慢查询和并发

Agent 场景下，一个用户问题可能触发多次 SQL 查询，并发一上来，数据库压力陡增。热词里慢sql优化、并行sql优化、ai agent 怎么扛并发这几个词其实都在提示这个痛点。我的做法是：给所有 Agent 发起的查询加缓存（相同意图 JSON 直接命中缓存），给数据库连接池设上限，对慢查询做监控告警。Agent 的并发问题，本质上是它把原本人工触发的低频查询变成了机器触发的高频查询，这个量级变化必须提前设计。

6.5 权限过滤要在查询层做，不能只在生成层做

前面提过权限问题，这里再强调一次。如果权限过滤只放在"生成回答"这一步，那模型其实已经看到了它不该看的数据，存在泄露风险。正确做法是在 SQL 查询和向量检索阶段就带上权限条件，比如 SQL 自动加WHERE dept_id = ?，向量检索按权限标签过滤。数据不出库，才是真安全。

7. 这套思路还能往哪延伸

把 SAG 这套"分层 + 路由 + 融合"的骨架搭起来之后，你会发现它能接的东西比想象中多。

一个是ontology rag的方向。热词里出现了这个词，我的理解是给知识库加一层本体（ontology），把实体、关系、属性显式建模出来。这样多跳查询就不用靠模型猜，而是沿着本体里的关系走。它和 SAG 是天然互补的：SAG 解决"查哪里"，ontology 解决"怎么连"。

另一个是Agent 工具生态。当你的检索层已经能提供"向量检索""SQL 查询""精确过滤"这些原子能力，把它们封装成标准工具，Agent 就能像搭积木一样组合使用。这时候检索层不再是 RAG 的一个内部模块，而是 Agent 的一个能力底座。热词里agent架构、agent项目、ai agent搭建这些词的热度，说明大家正在往这个方向走。

还有一个容易被忽略的点：评估。RAG 的评估本来就难，加了 SQL 之后更难——你不仅要评"回答对不对"，还要评"路由选对没""SQL 生成对没""融合排序合理没"。我的建议是分层评估，每一层单独建测试集，别指望一个端到端指标能定位问题。这块我还在摸索，有成熟做法了再单独写一篇。

最后分享一个我自己的体会：别一上来就追求"全自动"。我早期总想让模型端到端搞定一切，结果处处不可控。后来改成"规则兜底 + 模型增强"，把确定性的部分交给代码，把模糊的部分交给模型，系统反而稳了。SAG 这套思路的内核其实也是这个——该结构化的结构化，该语义的语义，各司其职。这个原则，比任何具体框架都更值得记住。