【RAG技术从小白到深入理解】一文搞懂 RAG索引、检索、生成与评估全流程-CSDN博客【RAG技术从小白到深入理解】一文搞懂 RAG索引、检索、生成与评估全流程-CSDN博客如果说前两章讨论的查询优化策略解决的是如何问得更好那么本章要解决的是问谁、怎么问——这是决定 RAG 系统能否在复杂生产环境中落地的关键一步。3.1 Routing路由智能流量控制什么是 RAG 路由在实际应用中用户的查询千差万别不同类型的问题需要走不同的检索路径。RAG 路由的核心思想是在收到用户查询后先由路由器Router分析查询特征然后将其分发到最合适的检索通道。一句话总结路由就是 RAG 系统的智能调度中心——不是所有问题都适合用向量检索把问题分发给最懂它的检索器才能获得最佳效果。路由的两种核心方式方式一基于 LLM 的语义路由利用 LLM 的理解能力对查询进行分类选择最合适的检索通道。class SemanticRouter: def __init__(self, llm): self.llm llm self.routes { vector_search: 适合知识库文档查询的事实和概念性问题, sql_query: 适合查询结构化数据、统计指标、聚合分析的问题, graph_query: 适合查询实体间关系、关联路径的问题, web_search: 适合需要实时信息、最新动态的问题 } def route(self, query: str) - str: prompt f分析以下用户查询选择最合适的检索路径。 ​ 可用路径 {self._format_routes()} ​ 查询: {query} ​ 请返回 JSON: {{path: xxx, reason: xxx}} response self.llm.invoke(prompt) return self._parse_route(response)方式二关键词 规则路由class KeywordRouter: def __init__(self): self.route_rules { sql_query: [ r(统计|汇总|平均|总和|排名|TOP\d), r(多少|数量|占比|增长率), r(销售额|用户量|指标|KPI) ], graph_query: [ r(关系|关联|路径|链路|上下游), r(依赖|影响范围|追溯到|归属于) ], web_search: [ r(最新|今天|最近|202[0-9]年|当前), r(新闻|动态|发布会|刚刚) ] } def route(self, query: str) - tuple: for path, patterns in self.route_rules.items(): for pattern in patterns: if re.search(pattern, query): return path, keyword_match return vector_search, default两种方式对比维度LLM 语义路由关键词规则路由准确率高90%中等70-85%延迟高0.5-2s极低 1ms维护成本低无需维护规则高需持续更新规则边界情况好LLM 可推理差只能匹配已知模式最佳实践采用两级路由——先用规则做快速初筛命中则直接路由未命中再走 LLM 语义路由兜底。完整的路由系统架构路由决策树可视化最佳实践建议先规则后 LLM规则路由零延迟能覆盖 60-70% 的常见场景LLM 路由兜底处理长尾和模糊查询路由日志监控记录每次路由决策和最终用户反馈点赞/点踩持续优化路由规则# 路由日志示例 router_log { query: 上季度销售额TOP3产品, rule_route: sql_query, # 关键词规则命中 llm_route: None, # LLM 未触发 latency_ms: 0.3, user_feedback: helpful # 用户点赞 }提供回退策略路由失败时的优雅降级方案——默认走向量检索保证系统可用性路由权重动态调优根据各通道的检索质量指标命中率、用户满意度动态调整路由倾向3.2 Query Construction查询构建核心挑战路由决定了去哪个数据源查但下一个问题随之而来怎么查这就是查询构建Query Construction要解决的核心问题。核心挑战总结语言鸿沟自然语言是非结构化的数据库需要严格的查询语法多源异构不同数据源SQL、图数据库、搜索引擎有各自的查询语言容错要求一次语法错误的查询会导致整个检索链路中断方法 1Text-to-SQL自然语言转 SQL 查询这是最成熟的查询构建方向让 LLM 将自然语言问题直接转换为可执行的 SQL 语句。完整实现示例import sqlite3 from typing import List, Dict, Optional class TextToSQLQueryBuilder: def __init__(self, llm, db_connection, table_schemas: Dict[str, str]): self.llm llm self.db db_connection self.schemas table_schemas def build_sql( self, query: str, max_retries: int 3 ) - Optional[str]: 将自然语言查询转换为 SQL prompt f你是一个 SQL 专家。请将以下自然语言问题转换为 SQLite 查询。 数据库 Schema: {self._format_schemas()} 规则 1. 只返回 SELECT 语句只读查询 2. 使用精确的列名和表名 3. 适当使用 LIMIT 限制返回行数最多 100 行 4. 只返回 SQL不要任何解释 问题: {query} SQL: for attempt in range(max_retries): sql self.llm.invoke(prompt).strip() sql self._clean_sql(sql) # 语法验证 if self._validate_sql(sql): return sql # 如果失败附上错误信息重试 prompt f\n\n上次 SQL 执行失败请修正\n{sql}\n错误{self._last_error} return None def query(self, natural_language: str) - List[Dict]: 端到端自然语言 → SQL → 结果 sql self.build_sql(natural_language) if not sql: raise QueryBuildError(无法生成有效的 SQL 查询) cursor self.db.execute(sql) columns [desc[0] for desc in cursor.description] return [dict(zip(columns, row)) for row in cursor.fetchall()] def _validate_sql(self, sql: str) - bool: 使用 EXPLAIN 或实际执行来验证 SQL try: self.db.execute(fEXPLAIN {sql}) return True except sqlite3.Error as e: self._last_error str(e) return False方法 2Text-to-Cypher图数据库查询构建当数据以图结构存储时需要将自然语言转为 CypherNeo4j 的查询语言。Text-to-Cypher 关键 Prompt 设计你是一个 Neo4j Cypher 查询专家。 图数据库 Schema - 节点标签Employee员工、Department部门、Project项目 - 关系类型BELONGS_TO属于部门、WORKS_ON参与项目、MANAGES管理 - 关键属性 Employee: name, hire_date, title Department: dept_name, floor Project: project_name, deadline 查询规则 1. 使用 MATCH 指定路径模式 2. 在 WHERE 中做过滤 3. 用 RETURN 指定返回字段 4. 适当使用 LIMIT 将以下问题转为 Cypher {query}方法 3混合查询构建在真实场景中用户问题往往需要同时查询多种数据源才能完整回答。混合查询实现class HybridQueryBuilder: def __init__(self, vector_store, sql_db, graph_db, llm): self.vector_store vector_store self.sql_db sql_db self.graph_db graph_db self.llm llm async def query(self, natural_query: str) - dict: 混合查询跨向量库、SQL、图数据库 # 1. 分析查询需要哪些数据源 sub_queries await self._decompose_query(natural_query) # 2. 并行构建各源查询 tasks [] if sub_queries.get(vector_query): tasks.append( (vector, self.vector_store.search( sub_queries[vector_query] )) ) if sub_queries.get(sql_query): sql self._build_sql(sub_queries[sql_query]) tasks.append( (sql, self._execute_sql(sql)) ) if sub_queries.get(graph_query): cypher self._build_cypher(sub_queries[graph_query]) tasks.append( (graph, self._execute_cypher(cypher)) ) # 3. 并行执行 results await asyncio.gather(*[t[1] for t in tasks]) # 4. 融合结果 return self._merge_results( {tasks[i][0]: results[i] for i in range(len(tasks))}, natural_query )查询构建的错误处理查询构建最常见的失败场景及应对策略核心防护策略防护层措施说明预检查SQL/语法校验在执行前拦截语法错误运行时兜底try-catch 重试捕获所有运行时异常空结果处理查询宽松化空结果时自动放宽条件超时保护查询超时 截断防止慢查询阻塞全链路最终兜底降级为向量检索结构化查询失败时回退Query Construction 性能优化关键优化维度查询计划缓存相同自然语言查询 → 缓存其生成的 SQL/Cypher表结构变更时自动刷新缓存查询简化策略复杂 JOIN 拆为多步简单查询优先执行高选择性过滤条件超时与熔断单次查询超时 3s → 取消并尝试简化连续失败 5 次 → 熔断 30s降级为备用路径小结本章覆盖了 RAG 系统中两个紧密关联但常被忽视的关键环节 关键洞察策略核心价值一句话总结Routing解决查哪里不是所有问题都该去向量库——把查询导向最懂它的检索器Query Construction解决怎么查自然语言和数据查询语言之间有鸿沟LLM 可以填平它两者组合端到端精准检索路由 查询构建 从用户问题到数据库结果的完整自动化链路分层思维是 RAG 系统工程化的关键路由层负责调度宏观决策查询构建层负责翻译微观执行两层配合才能让系统既聪明又可靠。下一篇将探讨RAG的索引生成优化