企业官网建设流程全解析

简介：这份资源是一套面向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=事实型RAG，1=关系型GraphRAG，2=规划型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(tokenizer=tokenize, token_pattern=None)), ("clf", LogisticRegression(max_iter=1000, C=1.0)), ]) pipe.fit(train_texts, train_labels) # 推理 query = "从上海出发去成都玩四天，想吃火锅和看熊猫" route = pipe.predict([query])[0] print(f"路由结果: {route}") # 2 -> 规划型

这段代码的关键点在于分词函数必须显式传给 TfidfVectorizer，否则中文会被当成一个整串。C=1.0是正则化强度，旅游 query 通常短文本居多，C 设太大容易过拟合，我一般从 0.5 到 2.0 之间调。实际部署时这个分类器可以做成一个独立微服务，延迟控制在 10ms 以内，不会成为瓶颈。

2.2 分流后的检索策略差异

分流只是第一步，真正影响答案质量的是两套检索链路的参数配置。RAG 侧我一般用 chunk size 256 到 512，overlap 64，top_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_hops=2): """ 从实体出发做多跳查询，返回关联子图 max_hops=2 覆盖大多数旅游关系链：城市->景点->附近景点 """ with driver.session() as session: result = session.run( """ MATCH path = (n:Entity {name: $name})-[*1..$hops]-(m) RETURN path LIMIT 50 """, name=entity_name, hops=max_hops ) return [record["path"] for record in result] # 调用示例 paths = graph_rag_query("故宫", max_hops=2) 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, key=lambda 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 itinerary

max_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}, timeout=3 ) 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 或 PyMuPDF，Word 用 python-docx，HTML 用 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] == b"PK": return "docx" # docx 本质是 zip if header[:5] == b"<!DOC" or header[:5] == b"<html": 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", encoding="utf-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_hops=2和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(maxsize=1000) 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 result

lru_cache的 maxsize 设 1000 是内存和命中率的平衡点，旅游问答的热门问题集中度很高，1000 条能覆盖大部分重复查询。实体级缓存的 TTL 设 10 分钟是因为景点信息变化不频繁，但门票库存这类实时数据不能走这层。路径级缓存的 key 用排序后的景点名加天数，保证同一组景点不同顺序能命中同一缓存。

验证缓存效果的方法很简单：在日志里打三个时间戳——请求进入、缓存查询返回、完整链路返回。如果缓存命中时延迟还在 1 秒以上，说明缓存层本身有问题，通常是序列化开销太大，换成 msgpack 能明显改善。

我自己的习惯是每次上线新版本前，先用 100 条历史 query 跑一遍全链路，看 P99 延迟和缓存命中率。如果 P99 超过 3 秒，先查外部 API 调用次数，再查图谱查询的 hops 设置。这个习惯帮我省了很多次线上翻车。希望帮到你。

本文还有配套的精品资源，点击获取