智能旅游问答平台：RAG与GraphRAG分流及路径规划实战

简介这份资源是一套面向RAG与GraphRAG学习者的智能旅游问答平台完整实施项目适合具备Python与前后端基础、希望深入检索增强生成与知识图谱应用的开发者。项目围绕智能分流、个性化推荐、动态路径规划、多格式文档智能解析及外部服务API集成展开并配套成都、乐山、眉山等四川多地旅游数据可用于课程设计、毕业项目或技术预研。压缩包共96个文件约10.96MB包含17个csv数据文件、15个vue前端组件、10个py脚本、9个ts文件及json、sqlite3、yaml等配置与存储文件覆盖数据清洗、向量库构建、FastAPI服务与前端聊天应用等模块。目前已有90人学习下载。通过该资源可掌握RAG与GraphRAG分流逻辑、BGE-M3向量库搭建、文档解析与API对接思路并参考完整目录结构快速复现一套可运行的旅游问答系统。1. 智能旅游问答平台RAG 与 GraphRAG 分流到底解决什么问题做过旅游问答的同行大概都有体会用户问“带老人去北京三天怎么安排”纯向量 RAG 能召回一堆攻略片段但拼出来的答案经常把故宫和长城的顺序搞反因为向量检索只看语义相似度不看景点之间的地理约束和开放时间约束。更麻烦的是同一个平台里既有“故宫门票多少钱”这种单跳事实问题也有“从杭州出发五天四晚预算五千带小孩想去海边和古镇”这种多约束规划问题用一套检索策略硬扛要么召回不够要么延迟爆炸。这个项目的核心思路就是用 RAG 处理事实型问答用 GraphRAG 处理关系型和路径型问答中间加一层智能分流器做路由。再叠上动态路径规划、多格式文档解析和外部 API 集成最终做成一个能真正回答“怎么走、怎么排、多少钱”的旅游问答平台。适合已经有基础 RAG 系统、想往图增强和工程化方向推进的团队参考也适合正在选型 RAG 框架的开发者看清边界。2. 智能分流器怎么设计从意图识别到路由决策2.1 为什么不能只用向量相似度做分流最常见的翻车做法是拿用户 query 去和两类知识库各做一次向量检索谁的分高走谁。这个方案在旅游场景下几乎必挂原因是 GraphRAG 侧的知识往往以实体关系形式存储比如“故宫—位于—北京”“北京—有景点—长城”这些三元组的文本嵌入和用户自然语言 query 的语义距离天然偏大。反过来RAG 侧的一段攻略文本可能因为包含“北京”“三天”等词在规划类 query 上拿到虚高分数。我一般会用一个轻量分类器做前置意图判断而不是靠检索分数。分类器的输入是 query 本身输出是三分类事实型、关系型、规划型。事实型走 RAG关系型走 GraphRAG规划型走 GraphRAG 加路径规划模块。分类器不需要大模型用一个小型 BERT 或者甚至 TF-IDF 加逻辑回归就能跑到 90% 以上准确率前提是标注数据覆盖够。# 意图分类器三分类路由 # 输入用户 query 文本 # 输出route 标签0事实型RAG1关系型GraphRAG2规划型GraphRAG路径 import jieba from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.linear_model import LogisticRegression from sklearn.pipeline import Pipeline # 训练数据示例实际需要至少 500 条覆盖各场景 train_texts [ 故宫门票多少钱, # 事实型 故宫开放时间, # 事实型 北京有哪些必去景点, # 关系型 故宫和长城哪个更值得去, # 关系型 带老人三天怎么安排北京, # 规划型 杭州出发五天四晚预算五千, # 规划型 ] train_labels [0, 0, 1, 1, 2, 2] # 中文需要先分词再向量化 def tokenize(text): return .join(jieba.cut(text)) pipe Pipeline([ (tfidf, TfidfVectorizer(tokenizertokenize, token_patternNone)), (clf, LogisticRegression(max_iter1000, C1.0)), ]) pipe.fit(train_texts, train_labels) # 推理 query 从上海出发去成都玩四天想吃火锅和看熊猫 route pipe.predict([query])[0] print(f路由结果: {route}) # 2 - 规划型这段代码的关键点在于分词函数必须显式传给 TfidfVectorizer否则中文会被当成一个整串。C1.0是正则化强度旅游 query 通常短文本居多C 设太大容易过拟合我一般从 0.5 到 2.0 之间调。实际部署时这个分类器可以做成一个独立微服务延迟控制在 10ms 以内不会成为瓶颈。2.2 分流后的检索策略差异分流只是第一步真正影响答案质量的是两套检索链路的参数配置。RAG 侧我一般用 chunk size 256 到 512overlap 64top_k 设 5 到 8。GraphRAG 侧则完全不同它需要先做实体链接把 query 里的“故宫”映射到图谱节点然后做多跳查询。# GraphRAG 侧实体链接与多跳查询示意 # 假设图谱用 Neo4j 存储节点为景点/城市/交通方式 from neo4j import GraphDatabase driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, password)) def graph_rag_query(entity_name, max_hops2): 从实体出发做多跳查询返回关联子图 max_hops2 覆盖大多数旅游关系链城市-景点-附近景点 with driver.session() as session: result session.run( MATCH path (n:Entity {name: $name})-[*1..$hops]-(m) RETURN path LIMIT 50 , nameentity_name, hopsmax_hops ) return [record[path] for record in result] # 调用示例 paths graph_rag_query(故宫, max_hops2) for p in paths[:3]: print(p)这里max_hops设 2 是血泪经验设 1 跳只能拿到直接邻居规划类问题不够用设 3 跳以上在旅游图谱里会引入大量噪声比如“故宫—北京—中国—亚洲”这种无意义链路。LIMIT 50是防止大图查询拖垮响应时间实际生产环境还要加缓存。提示实体链接的准确率直接决定 GraphRAG 的上限。旅游场景里“故宫”和“故宫博物院”必须归一化到同一节点否则多跳查询会断链。3. 动态路径规划模块把图谱查询结果变成可执行行程3.1 路径规划的数据结构设计GraphRAG 返回的是子图但用户要的是“第一天去哪、第二天去哪”这种有序行程。中间需要一个转换层把图结构转成带约束的路径规划问题。我一般用带时间窗和地理距离的变种 TSP 来建模。核心数据结构是一个景点列表每个景点包含名称、经纬度、建议游玩时长、开放时间、门票价格、与前一景点的交通方式。这些数据一部分来自图谱属性一部分来自外部 API 实时查询。# 行程规划核心数据结构与贪心局部搜索求解 import math from dataclasses import dataclass, field from typing import List dataclass class POI: name: str lat: float lng: float duration_hours: float # 建议游玩时长 open_hour: int # 开放时间小时 close_hour: int ticket: float 0.0 def haversine(p1: POI, p2: POI) - float: 计算两景点间球面距离单位公里 R 6371 dlat math.radians(p2.lat - p1.lat) dlon math.radians(p2.lng - p1.lng) a math.sin(dlat/2)**2 math.cos(math.radians(p1.lat)) * \ math.cos(math.radians(p2.lat)) * math.sin(dlon/2)**2 return 2 * R * math.asin(math.sqrt(a)) def plan_route(pois: List[POI], days: int, max_hours_per_day: float 8.0): 贪心构造初始解每天从剩余 POI 中选距离当前点最近且开放时间匹配的 然后做 2-opt 局部搜索优化 remaining pois.copy() itinerary [] for day in range(days): day_plan [] current_time 9.0 # 每天 9 点出发 current_poi None while remaining and current_time max_hours_per_day 9.0: # 筛选当前时间可进入的景点 candidates [p for p in remaining if p.open_hour current_time p.close_hour] if not candidates: break if current_poi is None: chosen candidates[0] else: chosen min(candidates, keylambda p: haversine(current_poi, p)) travel 0 if current_poi is None else haversine(current_poi, chosen) / 30 # 假设均速30km/h current_time travel chosen.duration_hours day_plan.append(chosen) remaining.remove(chosen) current_poi chosen itinerary.append(day_plan) return itinerarymax_hours_per_day设 8 是保守值带老人小孩的场景建议降到 6。haversine算的是直线距离实际交通时间要乘一个系数城市内我一般用 1.5 到 2.0跨城用 1.2。这个贪心解不保证最优但旅游场景用户对“最优”不敏感对“合理”敏感贪心加 2-opt 足够。3.2 外部 API 集成实时数据怎么接路径规划依赖三类外部数据天气、交通、门票库存。这些必须实时查不能缓存在图谱里。我一般用适配器模式封装每个 API 一个 adapter 类统一接口。# 外部服务适配器模式 from abc import ABC, abstractmethod import requests class TravelAPIAdapter(ABC): abstractmethod def get_weather(self, city: str, date: str) - dict: pass abstractmethod def get_traffic(self, from_poi: str, to_poi: str) - dict: pass class WeatherAPIAdapter(TravelAPIAdapter): def __init__(self, api_key: str, base_url: str): self.api_key api_key self.base_url base_url def get_weather(self, city: str, date: str) - dict: # 实际调用时加超时和重试 resp requests.get( f{self.base_url}/weather, params{city: city, date: date, key: self.api_key}, timeout3 ) resp.raise_for_status() return resp.json() def get_traffic(self, from_poi: str, to_poi: str) - dict: return {} # 天气适配器不实现交通超时设 3 秒是底线旅游问答用户等待超过 5 秒就会流失。重试策略我一般用两次间隔 0.5 秒再失败就降级到缓存数据并在答案里标注“交通信息可能不是最新”。注意外部 API 的失败不能阻塞整个问答链路。我的做法是每个 adapter 都有 fallback 返回值天气返回历史均值交通返回直线距离估算保证行程至少能生成。4. 多格式文档智能解析PDF、Word、图片怎么进知识库4.1 解析管线的分层设计旅游资料格式极其杂乱景区官网是 HTML攻略是 PDF地图是图片宣传册可能是扫描件。我一般把解析分成三层格式识别层、内容提取层、结构化层。格式识别用文件头 magic number 加扩展名双重判断防止改后缀的情况。内容提取层按格式走不同库PDF 用 pdfplumber 或 PyMuPDFWord 用 python-docxHTML 用 BeautifulSoup图片用 OCR。结构化层负责把提取出的文本切成 chunk 并打标签。# 多格式文档解析入口 import os import pdfplumber from docx import Document from bs4 import BeautifulSoup def detect_format(filepath: str) - str: 通过文件头识别真实格式 with open(filepath, rb) as f: header f.read(8) if header[:4] b%PDF: return pdf if header[:2] bPK: return docx # docx 本质是 zip if header[:5] b!DOC or header[:5] bhtml: return html return unknown def extract_text(filepath: str) - str: fmt detect_format(filepath) if fmt pdf: with pdfplumber.open(filepath) as pdf: return \n.join(page.extract_text() or for page in pdf.pages) elif fmt docx: doc Document(filepath) return \n.join(p.text for p in doc.paragraphs) elif fmt html: with open(filepath, r, encodingutf-8) as f: soup BeautifulSoup(f.read(), html.parser) return soup.get_text(separator\n) else: raise ValueError(f不支持的格式: {filepath})PDF 解析有个大坑pdfplumber 对扫描件返回空字符串。这时候要检测提取结果长度如果小于阈值就转走 OCR 流程。我一般设 50 字符为阈值低于这个值就认为需要 OCR。4.2 图片和表格的特殊处理旅游资料里图片占比很高尤其是地图和景区导览图。纯 OCR 只能拿到文字拿不到空间关系。我的做法是图片先走 OCR 提取文字同时用目标检测模型识别图例和标注框把文字和位置绑定存成“文字坐标”的结构。这样后续如果用户问“景区里洗手间在哪”可以基于坐标做空间推理。表格处理更麻烦PDF 里的表格用 pdfplumber 的 extract_tables 能拿到二维数组但合并单元格会错位。我一般加一步后处理检测空单元格并向前填充同时保留原始行列索引作为元数据。# PDF 表格提取与合并单元格修复 import pdfplumber def extract_tables_with_merge_fix(filepath: str): tables [] with pdfplumber.open(filepath) as pdf: for page in pdf.pages: for table in page.extract_tables(): # 向前填充空单元格处理合并单元格 for row in table: last_val None for i, cell in enumerate(row): if cell is None or cell.strip() : row[i] last_val else: last_val cell tables.append(table) return tables这个向前填充逻辑对横向合并有效纵向合并需要转置后再做一次。实际项目中我遇到过一张门票价格表纵向合并了“旺季”“淡季”两列不处理的话解析出来全是 None。提示多格式解析的产出必须带来源元数据文件名、页码、坐标否则后续答案溯源做不了用户问“你这个信息哪来的”会答不上。5. 避坑与排查分流、图谱、路径规划里最容易翻车的 5 个点5.1 分流器把规划问题误判成事实问题现象用户问“北京三天怎么玩”系统走了 RAG 链路返回一堆景点介绍没有行程安排。 原因训练数据里“怎么玩”这类规划意图样本太少分类器被“北京”“三天”这些词带偏到事实类。 解决在训练数据里强制加入“怎么玩”“怎么安排”“路线”等触发词样本同时把分类器的类别权重设为 balanced避免多数类主导。5.2 GraphRAG 多跳查询返回爆炸性结果现象查询“故宫附近有什么”返回上千条路径响应时间超过 10 秒。 原因图谱里“附近”关系没有距离约束所有通过“位于北京”关联的节点都被算作附近。 解决在关系边上加距离属性查询时加WHERE r.distance 5过滤同时限制max_hops2和LIMIT 50。5.3 路径规划忽略开放时间导致行程不可执行现象生成的行程把故宫排在周一但故宫周一闭馆。 原因规划算法只考虑了地理距离和游玩时长没有把开放时间作为硬约束。 解决在 POI 数据结构里加open_days字段规划前先过滤掉当天不开放的景点。这个坑我踩过两次后来直接把开放时间校验做成规划器的前置断言。5.4 外部 API 超时拖垮整个问答链路现象用户提问后 30 秒才返回体验极差。 原因天气 API 和交通 API 串行调用每个超时 10 秒累计超过 20 秒。 解决改成并行调用用concurrent.futures同时发请求总超时设 5 秒。任何 API 失败都走 fallback不阻塞主流程。5.5 文档解析把页眉页脚当正文入库现象RAG 检索出来的答案里夹杂“第 3 页”“版权所有”等噪声。 原因PDF 提取时没有过滤页眉页脚这些文本被切进 chunk。 解决在解析层加规则过滤检测每页重复出现的短文本行并剔除。更稳的做法是用版面分析模型识别正文区域但成本高规则过滤能覆盖 80% 场景。6. 进阶技巧用缓存和预计算把响应压到 2 秒内分流加图谱加路径规划链路很长不优化的话首字延迟很难看。我一般做三层缓存第一层是 query 级缓存相同问题直接返回第二层是实体级缓存图谱查询结果按实体名缓存 10 分钟第三层是路径级缓存相同景点集合的规划结果缓存 30 分钟。# 三层缓存示意 import hashlib import time from functools import lru_cache # 第一层query 级用 LRU 做进程内缓存 lru_cache(maxsize1000) def cached_qa(query: str): return pipeline_run(query) # 第二层实体级带 TTL _entity_cache {} def cached_graph_query(entity: str, ttl: int 600): now time.time() if entity in _entity_cache: result, ts _entity_cache[entity] if now - ts ttl: return result result graph_rag_query(entity) _entity_cache[entity] (result, now) return result # 第三层路径级用景点集合的 hash 做 key _route_cache {} def cached_plan(poi_names: list, days: int): key hashlib.md5(f{sorted(poi_names)}_{days}.encode()).hexdigest() if key in _route_cache: return _route_cache[key] result plan_route([poi_map[n] for n in poi_names], days) _route_cache[key] result return resultlru_cache的 maxsize 设 1000 是内存和命中率的平衡点旅游问答的热门问题集中度很高1000 条能覆盖大部分重复查询。实体级缓存的 TTL 设 10 分钟是因为景点信息变化不频繁但门票库存这类实时数据不能走这层。路径级缓存的 key 用排序后的景点名加天数保证同一组景点不同顺序能命中同一缓存。验证缓存效果的方法很简单在日志里打三个时间戳——请求进入、缓存查询返回、完整链路返回。如果缓存命中时延迟还在 1 秒以上说明缓存层本身有问题通常是序列化开销太大换成 msgpack 能明显改善。我自己的习惯是每次上线新版本前先用 100 条历史 query 跑一遍全链路看 P99 延迟和缓存命中率。如果 P99 超过 3 秒先查外部 API 调用次数再查图谱查询的 hops 设置。这个习惯帮我省了很多次线上翻车。希望帮到你。本文还有配套的精品资源点击获取