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 的第一个主张是按知识的形态分层存储知识形态典型内容适合的检索方式非结构化制度文档、产品说明、FAQ向量检索 关键词半结构化带字段的工单、带标签的案例字段过滤 向量结构化订单、用户、交易记录SQL 精确查询 聚合这个分层不是拍脑袋而是让每种数据用它最擅长的方式被查询。文档就该用语义检索因为它没有固定字段而订单表就该用 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或者写出全表扫描把数据库拖垮。我的做法是三层防护只读账号Agent 用的数据库连接必须是只读权限从根上杜绝写操作。SQL 白名单校验解析生成的 SQL只允许SELECT禁止DROP、DELETE、UPDATE、INSERT禁止多语句。强制 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 fSELECT {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(reverseTrue) 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 这套思路的内核其实也是这个——该结构化的结构化该语义的语义各司其职。这个原则比任何具体框架都更值得记住。