基于知识图谱的RAG问答系统——可溯源
·
引言
本项目旨在从零开始，完整地构建一个基于《本草纲目》文本的、可交互、可溯源的智能问答系统。系统能够优先从我们自建的知识图谱中查找答案并提供原文依据；当知识图谱无法覆盖用户问题时，系统能智能地回退，利用大语言模型（LLM）的通用知识库进行回答。
-
技术栈: Python, Conda, Transformers, PyTorch, NetworkX, Streamlit
-
核心流程: 环境配置 -> 知识抽取与图谱构建 -> 可溯源Web UI应用开发 -> 部署运行
第一阶段：环境配置与数据准备
这是所有工作的基础，一个干净、隔离的环境能避免许多不必要的麻烦。
1. 创建并激活Conda环境
我们使用Conda来创建一个独立的Python环境，确保项目依赖不会与其他项目冲突。
# 创建一个名为 kg_rag_tut 的环境，并指定Python版本为3.10
conda create -n kg_rag_tut python=3.10
# 激活新创建的环境
conda activate kg_rag_tut
2. 安装所有必需的Python库
在已激活的kg_rag_tut环境中，使用pip一次性安装所有核心依赖。
pip install torch transformers networkx streamlit tqdm matplotlib langchain
-
torch&transformers: 运行和加载大语言模型的核心。 -
networkx: 创建、操作和保存知识图谱。 -
streamlit: 构建最终的用户交互Web界面。 -
tqdm: 在处理长任务时显示美观的进度条。 -
matplotlib: 用于知识图谱的可视化。 -
langchain: 提供方便的文本分割工具。
3. 准备原始数据
-
在你的项目主目录下，创建一个名为
benchaogangmu_clean的文件夹。 -
利用以下脚本从网站上爬取本草纲目txt文本
import requests from bs4 import BeautifulSoup import time import os import re import json import hashlib import random import warnings from urllib.parse import urljoin # 禁用SSL警告 warnings.filterwarnings('ignore', message='Unverified HTTPS request') # 网站配置 BASE_URL = "https://www.diancang.xyz/xuanxuewushu/bencaogangmu/" OUTPUT_DIR = "benchaogangmu_clean" PROGRESS_FILE = os.path.join(OUTPUT_DIR, 'progress.json') # 创建保存目录 if not os.path.exists(OUTPUT_DIR): os.makedirs(OUTPUT_DIR) # 初始化进度文件 if not os.path.exists(PROGRESS_FILE): with open(PROGRESS_FILE, 'w', encoding='utf-8') as f: json.dump({ 'completed': [], 'failed': [], 'volume_order': [] }, f, ensure_ascii=False, indent=2) def load_progress(): """加载进度文件""" with open(PROGRESS_FILE, 'r', encoding='utf-8') as f: return json.load(f) def save_progress(progress): """保存进度文件""" with open(PROGRESS_FILE, 'w', encoding='utf-8') as f: json.dump(progress, f, ensure_ascii=False, indent=2) def get_toc_links(soup, base_url): """从目录页提取卷链接 - 针对该网站优化""" # 尝试多种可能的目录容器 toc_containers = [ soup.find('div', class_='list_box'), # 可能是这个网站使用的类名 soup.find('div', class_='list-box'), soup.find('div', class_='content'), soup.find('div', class_='article'), soup.find('div', id='content'), soup.find('div', id='article'), soup.find('ul', class_='chapter-list'), soup.find('ul', class_='toc'), ] # 过滤掉None值 toc_containers = [tc for tc in toc_containers if tc is not None] # 如果没有找到任何容器，尝试直接在整个页面查找 if not toc_containers: toc_containers = [soup] volume_links = [] # 遍历所有可能的容器 for container in toc_containers: # 查找所有链接 links = container.find_all('a', href=True) for link in links: href = link.get('href') if href and 'bencaogangmu' in href and href.endswith('.html'): # 跳过非卷页面 if 'index' in href or 'list' in href or 'catalog' in href: continue volume_url = urljoin(base_url, href) volume_title = link.text.strip() # 如果标题为空，尝试从链接提取 if not volume_title: # 从URL提取标题 volume_title = href.split('/')[-1].replace('.html', '') volume_title = re.sub(r'_\d+$', '', volume_title) # 移除末尾数字 volume_title = volume_title.replace('_', ' ').title() # 生成唯一ID volume_id = hashlib.md5(volume_title.encode('utf-8')).hexdigest()[:8] volume_links.append({ 'id': volume_id, 'title': volume_title, 'url': volume_url }) # 如果在这个容器中找到了链接，停止搜索 if volume_links: break return volume_links def scrape_volume_page(url, volume_id, volume_title): """爬取单个卷页面""" headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36', 'Referer': BASE_URL } try: # 添加随机延迟 time.sleep(random.uniform(2, 5)) # 获取页面 response = requests.get(url, headers=headers, verify=False) if response.status_code != 200: return { 'success': False, 'error': f"HTTP状态码: {response.status_code}" } # 解析HTML soup = BeautifulSoup(response.content, 'html.parser') # 查找内容区域 - 尝试多种选择器 content_selectors = [ {'name': 'div', 'class': 'content'}, {'name': 'div', 'id': 'content'}, {'name': 'article'}, {'name': 'div', 'class': 'article'}, {'name': 'div', 'class': 'text'}, {'name': 'div', 'class': 'main-content'}, {'name': 'div', 'class': 'entry-content'}, ] content_div = None for selector in content_selectors: if 'class' in selector: content_div = soup.find(selector['name'], class_=selector['class']) elif 'id' in selector: content_div = soup.find(selector['name'], id=selector['id']) else: content_div = soup.find(selector['name']) if content_div: break if not content_div: return { 'success': False, 'error': "未找到内容区域" } # 提取文本内容 text = content_div.get_text(separator='\n', strip=True) # 清理文本 text = re.sub(r'\n{3,}', '\n\n', text) # 减少多余空行 text = re.sub(r'[\xa0\u3000]+', ' ', text) # 替换特殊空格 # 移除常见页眉页脚文本 footer_phrases = [ "返回目录", "上一章", "下一章", "本章完", "全本完", "copyright", "©", "版权所有", "本文来源", "本文链接", "相关推荐", "热门推荐", "阅读更多", "分享到", "点击数", "推荐阅读", "扫一扫", "公众号", "二维码", "微信" ] for phrase in footer_phrases: text = re.sub(rf'\n.*{phrase}.*\n', '\n', text, flags=re.IGNORECASE) # 移除URL text = re.sub(r'https?://\S+', '', text) return { 'success': True, 'content': text.strip() } except Exception as e: return { 'success': False, 'error': str(e) } def main(): print("《本草纲目》爬虫开始运行...") print(f"数据将保存到: {OUTPUT_DIR}") start_time = time.time() # 加载进度 progress = load_progress() # 第一步：获取目录页 print(f"正在获取目录页: {BASE_URL}") headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8' } try: response = requests.get(BASE_URL, headers=headers, verify=False) response.encoding = 'utf-8' if response.status_code != 200: print(f"目录页请求失败，状态码: {response.status_code}") return # 解析目录页 soup = BeautifulSoup(response.text, 'html.parser') print(f"页面标题: {soup.title.text if soup.title else '无标题'}") # 提取卷链接 volume_links = get_toc_links(soup, BASE_URL) if not volume_links: print("未找到卷链接，请检查选择器或网站结构") # 保存页面用于调试 with open(os.path.join(OUTPUT_DIR, 'debug_toc.html'), 'w', encoding='utf-8') as f: f.write(response.text) print("已保存目录页为 debug_toc.html 用于分析") return print(f"找到 {len(volume_links)} 卷") # 记录卷顺序 for i, volume in enumerate(volume_links): volume['index'] = i # 检查是否已爬取 if volume['id'] in progress['completed']: volume['completed'] = True else: volume['completed'] = False # 爬取所有卷 success_count = 0 for i, volume in enumerate(volume_links): if volume['completed']: print(f"[{i+1}/{len(volume_links)}] 跳过已完成卷: {volume['title']}") continue print(f"[{i+1}/{len(volume_links)}] 正在爬取卷: {volume['title']}") result = scrape_volume_page(volume['url'], volume['id'], volume['title']) if result['success']: # 保存到文件 safe_title = re.sub(r'[\\/*?:"<>|]', '', volume['title']) filename = f"{volume['index']+1:03d}_{safe_title}.txt" filepath = os.path.join(OUTPUT_DIR, filename) with open(filepath, 'w', encoding='utf-8') as f: f.write(f"=== 《本草纲目》{volume['title']} ===\n\n") f.write(result['content']) f.write(f"\n\n来源: {volume['url']}") # 更新进度 progress['completed'].append(volume['id']) progress['volume_order'].append({ 'id': volume['id'], 'title': volume['title'], 'filename': filename, 'index': volume['index'] }) save_progress(progress) print(f" 成功保存: {filename} ({len(result['content'])} 字符)") success_count += 1 else: print(f" 爬取失败: {result['error']}") # 记录失败 progress['failed'].append({ 'id': volume['id'], 'title': volume['title'], 'url': volume['url'], 'error': result['error'], 'timestamp': time.strftime("%Y-%m-%d %H:%M:%S") }) save_progress(progress) # 创建完整版 create_full_version(progress) print("\n爬取完成!") print(f"成功爬取: {success_count}/{len(volume_links)} 卷") print(f"失败: {len(progress['failed'])} 卷") except Exception as e: print(f"爬取过程中发生错误: {str(e)}") print(f"总耗时: {time.time() - start_time:.2f} 秒") def create_full_version(progress): """创建合并的完整版《本草纲目》""" # 确保有卷被爬取 if not progress['volume_order']: print("没有可用的卷创建完整版") return # 按索引排序 sorted_volumes = sorted(progress['volume_order'], key=lambda x: x['index']) full_file = os.path.join(OUTPUT_DIR, "《本草纲目》完整版.txt") with open(full_file, 'w', encoding='utf-8') as full: full.write("《本草纲目》完整版\n\n") full.write("=" * 50 + "\n\n") for vol in sorted_volumes: vol_file = os.path.join(OUTPUT_DIR, vol['filename']) if os.path.exists(vol_file): full.write(f"=== 卷{vol['index']+1}: {vol['title']} ===\n\n") with open(vol_file, 'r', encoding='utf-8') as f: # 跳过第一行标题 content = f.readlines()[1:] full.writelines(content) full.write("\n" + "=" * 50 + "\n\n") print(f"已创建完整版: {full_file}") def retry_failed(): """重试失败的卷""" progress = load_progress() if not progress['failed']: print("没有失败的卷需要重试") return print(f"发现 {len(progress['failed'])} 个失败的卷需要重试") retry_list = progress['failed'].copy() progress['failed'] = [] # 清空失败列表 save_progress(progress) success_count = 0 for i, item in enumerate(retry_list): print(f"[{i+1}/{len(retry_list)}] 重试卷: {item['title']}") # 查找原始索引 original_index = None for vol in progress['volume_order']: if vol['id'] == item['id']: original_index = vol['index'] break if original_index is None: print(" 找不到原始索引，跳过") continue result = scrape_volume_page(item['url'], item['id'], item['title']) if result['success']: # 保存到文件 safe_title = re.sub(r'[\\/*?:"<>|]', '', item['title']) filename = f"{original_index+1:03d}_{safe_title}.txt" filepath = os.path.join(OUTPUT_DIR, filename) with open(filepath, 'w', encoding='utf-8') as f: f.write(f"=== 《本草纲目》{item['title']} ===\n\n") f.write(result['content']) f.write(f"\n\n来源: {item['url']}") # 更新进度 progress['completed'].append(item['id']) progress['volume_order'].append({ 'id': item['id'], 'title': item['title'], 'filename': filename, 'index': original_index }) save_progress(progress) print(f" 重试成功: {filename}") success_count += 1 else: print(f" 重试失败: {result['error']}") # 重新添加到失败列表 progress['failed'].append(item) save_progress(progress) # 重新创建完整版 create_full_version(progress) print(f"\n重试完成! 成功: {success_count}/{len(retry_list)}") if __name__ == "__main__": if '--retry' in os.sys.argv: print("执行重试操作...") retry_failed() else: main() -
将所有从《本草纲目》网站上爬取并经过初步清洗的
.txt文本文件放入此文件夹中。每个文件代表一个章节，例如01_序例上.txt,02_序例下.txt等。
第二阶段：知识图谱构建（含原文溯源）
目标: 编写一个脚本，该脚本能自动读取所有文本文档，利用本地部署的大语言模型进行信息抽取，并构建一个结构化的知识图谱。最关键的是，图谱中的每一条知识关系，都必须附带其原始文本出处，为后续的“答案溯源”功能打下基础。
1. 创建知识图谱构建脚本：build_graph.py
在项目主目录下创建 build_graph.py 文件，并将以下完整代码复制进去(我使用的本地模型，可以改成调用性能更强的API)。
# build_graph.py
import os
# 设置环境变量，减少不必要的警告信息，并指定使用的GPU
os.environ['TRANSFORMERS_VERBOSITY'] = 'error'
CUDA_DEVICE_IDS = "2,3,4,5" # <--- 在这里修改为你希望使用的GPU编号
os.environ['CUDA_VISIBLE_DEVICES'] = CUDA_DEVICE_IDS
import re
import json
import time
import networkx as nx
from tqdm import tqdm
import matplotlib
matplotlib.use('Agg') # 使用非交互式后端，避免在服务器上显示图形窗口
import matplotlib.pyplot as plt
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from langchain.text_splitter import RecursiveCharacterTextSplitter
# --- 1. 配置区 ---
MODEL_PATH = "xxx/Qwen2.5-14B-Instruct"
INPUT_DIR = "benchaogangmu_clean"
GRAPH_OUTPUT_FILE = "bencao_kg_local.gexf" # 输出的图谱文件名
# --- 2. 本地模型加载函数 ---
def load_local_model(model_path):
print(f"正在从本地路径加载模型: {model_path}")
model = AutoModelForCausalLM.from_pretrained(
model_path,
torch_dtype=torch.bfloat16,
device_map="auto", # 自动将模型分载到所有可见的GPU上
)
tokenizer = AutoTokenizer.from_pretrained(model_path)
print("模型和分词器加载成功。")
return model, tokenizer
# --- 3. 知识抽取函数 ---
def extract_knowledge_locally(text_chunk, model, tokenizer):
"""
使用本地LLM从单个文本块中抽取知识，包含健壮的JSON解析逻辑。
"""
prompt = build_extraction_prompt(text_chunk)
messages = [{"role": "user", "content": prompt}]
text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
model_inputs = tokenizer([text], return_tensors="pt").to(model.device)
generated_ids = model.generate(
model_inputs.input_ids,
max_new_tokens=2048,
do_sample=False
)
generated_ids = [
output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
]
response_text = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
# 健壮的JSON解析逻辑，处理不完美的模型输出
try:
json_match = re.search(r'```json\s*([\s\S]*?)\s*```', response_text)
if json_match:
clean_str = json_match.group(1).strip()
else:
first_brace_index = response_text.find('{')
if first_brace_index == -1:
raise json.JSONDecodeError("在模型输出中未找到JSON对象的起始 '{'。", response_text, 0)
clean_str = response_text[first_brace_index:]
decoder = json.JSONDecoder()
json_obj, _ = decoder.raw_decode(clean_str)
return json_obj
except json.JSONDecodeError as e:
print(f"\n[错误] JSON解析失败: {e}")
return None
# --- 4. 辅助函数 ---
def chunk_document(doc_content, chunk_size=500, chunk_overlap=100):
"""将长文档切分为重叠的小块。"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size, chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", "。", "，", "、", " "]
)
return text_splitter.split_text(doc_content)
def load_chapter_documents(directory):
"""加载目录下的所有.txt文件。"""
print(f"正在从目录 {directory} 加载章节文件...")
documents = {}
files_to_load = sorted([f for f in os.listdir(directory) if f.endswith('.txt')])
for filename in tqdm(files_to_load, desc="加载文件"):
filepath = os.path.join(directory, filename)
try:
with open(filepath, 'r', encoding='utf-8') as f:
documents[filename] = f.read()
except Exception as e:
print(f"读取文件 {filename} 失败: {e}")
print(f"成功加载 {len(documents)} 个章节文件。")
return documents
def build_extraction_prompt(text_chunk):
"""构建用于知识抽取的Prompt。"""
return f"""
你是一名资深的中医药学者和信息抽取专家。你的任务是从下面提供的《本草纲目》文本段落中，提取关于核心药材的结构化信息。
请严格按照以下要求进行：
1. **识别核心药材**：确定本文段主要介绍的药材。
2. **提取信息**：提取该药材的`释名`(别名)、`气味`(药性)和`主治`(用途)。
3. **输出格式**：必须以严格的JSON格式输出，结构如下：
{{
"herb": "核心药材名称",
"relations": [
{{"subject": "核心药材名称", "relation": "有别名", "object": "别名1"}},
{{"subject": "核心药材名称", "relation": "气味是", "object": "气味描述"}},
{{"subject": "核心药材名称", "relation": "主治", "object": "病症1"}}
]
}}
4. **内容来源**：所有提取的对象(object)都必须来源于源文本。如果某项信息不存在，则不要包含该项。
**源文本：**
---
{text_chunk}
---
**请严格按照上述要求，仅输出JSON格式的提取结果：**
"""
def visualize_subgraph(graph, center_node, radius=1):
"""可视化指定节点周围的子图。"""
if center_node not in graph:
print(f"错误：节点 '{center_node}' 不在图谱中。")
return
try:
matplotlib.rcParams['font.sans-serif'] = ['SimHei']
matplotlib.rcParams['axes.unicode_minus'] = False
except:
print("[警告] SimHei字体未找到，可视化标签可能显示为方框。")
sub_nodes = set(nx.single_source_shortest_path(graph, center_node, cutoff=radius).keys())
subgraph = graph.subgraph(sub_nodes)
plt.figure(figsize=(12, 12))
pos = nx.spring_layout(subgraph, k=0.8, iterations=50)
node_colors = ["skyblue" if data.get("type") == "药材" else "lightgreen" for node, data in subgraph.nodes(data=True)]
nx.draw_networkx(subgraph, pos, with_labels=True, node_size=2000, node_color=node_colors, font_size=10)
edge_labels = nx.get_edge_attributes(subgraph, 'label')
nx.draw_networkx_edge_labels(subgraph, pos, edge_labels=edge_labels, font_size=8)
plt.title(f"'{center_node}' 的知识子图", size=15)
safe_center_node = re.sub(r'[\\/*?:"<>|]', '', center_node)
VIS_OUTPUT_FILE = f"{safe_center_node}_subgraph.png"
plt.savefig(VIS_OUTPUT_FILE, dpi=300, bbox_inches='tight')
print(f"子图可视化已保存到 {VIS_OUTPUT_FILE}")
# --- 5. 主逻辑 ---
if __name__ == "__main__":
model, tokenizer = load_local_model(MODEL_PATH)
documents_dict = load_chapter_documents(INPUT_DIR)
KG = nx.DiGraph()
print(f"\n开始使用本地模型进行知识抽取 (包含溯源信息)...")
MAX_CHUNK_RETRIES = 3
for filename, doc_content in tqdm(documents_dict.items(), desc="文件处理进度"):
chunks = chunk_document(doc_content)
for i, chunk in enumerate(tqdm(chunks, desc=f"处理 {filename}", leave=False)):
extracted_data = None
for attempt in range(MAX_CHUNK_RETRIES):
extracted_data = extract_knowledge_locally(chunk, model, tokenizer)
if extracted_data is not None:
break
else:
if attempt < MAX_CHUNK_RETRIES - 1:
time.sleep(2) # 失败后稍作等待
if extracted_data is None:
continue
if "relations" in extracted_data:
for relation in extracted_data["relations"]:
if all(k in relation for k in ["subject", "object", "relation"]):
subj, obj, rel = relation["subject"], relation["object"], relation["relation"]
if subj and obj and rel:
KG.add_node(subj.strip(), type='药材')
KG.add_node(obj.strip(), type=rel)
# --- 核心改动：在添加边时，把原文(chunk)作为边的'source'属性存进去 ---
KG.add_edge(subj.strip(), obj.strip(), label=rel, source=chunk)
print(f"\n知识图谱构建完成！")
print(f"图谱中有 {KG.number_of_nodes()} 个节点和 {KG.number_of_edges()} 条边。")
if KG.number_of_nodes() > 0:
nx.write_gexf(KG, GRAPH_OUTPUT_FILE)
print(f"图谱已保存到 {GRAPH_OUTPUT_FILE}")
# 随机选择一个药材节点进行可视化
example_node = next((node for node, data in KG.nodes(data=True) if data.get("type") == "药材"), None)
if example_node:
visualize_subgraph(KG, example_node, radius=1)
else:
print("\n处理结束，知识图谱为空。")
2. 运行脚本并检查产出
在终端中，确保你在kg_rag_tut环境下，然后运行脚本：
python build_graph.py
脚本执行完毕后，你的项目文件夹中会生成两个核心产出：
-
bencao_kg_local.gexf: 这是我们包含所有知识和原文依据的知识图谱文件。 -
[药材名]_subgraph.png: 一个示例药材的知识子图图片，用于快速验证图谱构建是否成功。
第三阶段：构建可溯源的混合式问答UI
目标: 利用Streamlit框架，创建一个用户友好的Web聊天界面。该界面能智能查询上一步生成的知识图谱，并在图谱无相关知识时回退到LLM的通用知识库进行回答，同时始终保持答案的可溯源性，将原文出处清晰地展示给用户。
1. 创建Streamlit应用脚本：streamlit_app.py
在项目主目录下创建 streamlit_app.py 文件，并将以下完整代码复制进去。
# streamlit_app.py
import streamlit as st
import os
import re
import networkx as nx
from collections import defaultdict
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
# --- 1. 配置区 ---
os.environ['TRANSFORMERS_VERBOSITY'] = 'error'
os.environ['CUDA_VISIBLE_DEVICES'] = "2,3,4,5"
MODEL_PATH = "/mnt/2b44d1e2-569a-42b3-9610-04ea73771f19/zhangxiaopeng/Qwen2.5-14B-Instruct"
KG_FILE_PATH = "bencao_kg_local.gexf"
# --- 2. 使用Streamlit缓存来加载一次性的昂贵资源 ---
@st.cache_resource
def load_model_and_graph():
"""
使用Streamlit的缓存机制，确保模型和图谱只在应用启动时加载一次。
"""
print("--- 正在加载模型和知识图谱 (此过程只在首次运行时执行) ---")
model = AutoModelForCausalLM.from_pretrained(MODEL_PATH, torch_dtype=torch.bfloat16, device_map="auto")
tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH)
try:
kg = nx.read_gexf(KG_FILE_PATH)
print(f"知识图谱加载成功！")
except FileNotFoundError:
kg = None
st.error(f"错误：知识图谱文件 '{KG_FILE_PATH}' 未找到。请先成功运行 build_graph.py。")
print("--- 加载完成 ---")
return model, tokenizer, kg
# --- 3. 后端处理函数 ---
def extract_entity(question, model, tokenizer):
"""从用户问题中抽取核心实体。"""
prompt = f'从以下问题中，抽取出最关键的中草药名称。你只需要返回药材名。\n问题: "{question}"\n药材名:'
messages = [{"role": "user", "content": prompt}]
text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = tokenizer([text], return_tensors="pt").to(model.device)
outputs = model.generate(inputs.input_ids, max_new_tokens=50, do_sample=False)
entity = tokenizer.batch_decode(outputs[:, inputs.input_ids.shape[1]:], skip_special_tokens=True)[0].strip()
return entity
def query_knowledge_graph_with_sources(entity, kg):
"""
在知识图谱中查询给定实体的所有关系，并提取原文依据。
返回两个值：一个给LLM看的简洁上下文，一个给用户看的带原文的详细上下文。
"""
if not kg.has_node(entity):
return "NOT_FOUND", None
grouped_facts = defaultdict(list)
for neighbor in kg.neighbors(entity):
edge_data = kg.get_edge_data(entity, neighbor)
if edge_data:
relation = edge_data.get('label', '有关系')
source_text = edge_data.get('source', '未找到原文依据。')
fact_with_source = {"fact": neighbor, "source": source_text.strip()}
grouped_facts[relation].append(fact_with_source)
if not grouped_facts:
return "NOT_FOUND", None
context_for_llm = f"关于“{entity}”的已知事实如下：\n"
context_for_display = ""
for relation, facts_list in grouped_facts.items():
context_for_llm += f"【{relation}】: {', '.join(set(item['fact'] for item in facts_list))}\n"
context_for_display += f"#### 【{relation}】\n"
unique_facts = {item['fact']: item for item in facts_list}.values()
for item in unique_facts:
context_for_display += f"- **事实**: {item['fact']}\n"
context_for_display += f" - **原文依据**: “...{item['source'][:200]}...”\n"
context_for_display += "\n"
return context_for_llm.strip(), context_for_display.strip()
def generate_final_answer(question, context, model, tokenizer):
"""根据有无上下文，选择不同模板生成最终答案。"""
if context is None or "NOT_FOUND" in context:
# 模式B: 通用知识回退
prompt_template = "你是一个知识渊博、有问必答的AI助手。请根据你的知识库，清晰、准确地回答以下问题。\n\n[问题]\n{question}\n\n[你的回答]"
prompt = prompt_template.format(question=question)
else:
# 模式A: 知识图谱优先
prompt_template = """
你是一个专业、严谨的中医药问答助手。请根据下面提供的“背景知识”，用一段通顺、连贯、总结性的话来回答用户的“提问”。
你的回答必须完全基于“背景知识”，不得包含任何“背景知识”中没有提到的信息。
[背景知识]
{context}
[提问]
{question}
[回答]
"""
prompt = prompt_template.format(context=context, question=question)
messages = [{"role": "user", "content": prompt}]
text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
inputs = tokenizer([text], return_tensors="pt").to(model.device)
outputs = model.generate(inputs.input_ids, max_new_tokens=512, do_sample=False)
final_answer = tokenizer.batch_decode(outputs[:, inputs.input_ids.shape[1]:], skip_special_tokens=True)[0]
return final_answer
# --- 4. Streamlit UI 界面 ---
st.set_page_config(page_title="可溯源问答", layout="wide")
st.title("🌿 可溯源的《本草纲目》知识问答系统")
st.caption("优先查询自建知识图谱，并提供原文出处。若无相关知识则由大模型通用知识库回答。")
# 加载昂贵资源
model, tokenizer, kg = load_model_and_graph()
if "messages" not in st.session_state:
st.session_state.messages = []
# 显示历史聊天记录
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
if "display_context" in message and message["display_context"]:
with st.expander("✅ 查看本次回答的知识图谱依据"):
st.markdown(message["display_context"])
# 获取用户输入
if prompt := st.chat_input("请输入关于中草药的问题..."):
if model is None or kg is None:
st.error("模型或知识图谱未能成功加载，请检查终端日志。")
else:
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.markdown(prompt)
with st.chat_message("assistant"):
with st.spinner("思考中... (识别实体 -> 检索图谱 -> 生成答案)"):
entity = extract_entity(prompt, model, tokenizer)
llm_context, display_context = query_knowledge_graph_with_sources(entity, kg)
response = generate_final_answer(prompt, llm_context, model, tokenizer)
st.markdown(response)
if display_context:
with st.expander("✅ 查看本次回答的知识图谱依据"):
st.markdown(display_context)
else:
st.info("ℹ️ 此回答由模型的通用知识生成，非出自《本草纲目》知识图谱。")
st.session_state.messages.append({
"role": "assistant",
"content": response,
"display_context": display_context
})
第四阶段：部署与运行
这是我们项目的最后一步，将应用启动并进行交互。
1. 运行Web应用
在你的SSH终端中，确保你在项目主目录下，并且kg_rag_tut环境已激活，然后运行以下命令：
streamlit run streamlit_app.py
2. 访问与交互
运行后，终端会显示两个URL：
-
Network URL: 这是你可以在局域网内访问的地址，例如
http://172.17.27.xxx:8501。 -
External URL: 一个公网地址。
使用浏览器（如果已安装）打开 Network URL。将看到一个功能完善的聊天界面，可以开始提问并体验我们最终的、可溯源的、混合式问答系统了。
最终项目结构总结
至此，你成功的项目文件夹结构应如下所示：
qwen_bencao_rag/
├── benchaogangmu_clean/ # 存放原始.txt数据
│ ├── 01_序例上.txt
│ └── ...
├── build_graph.py # 知识图谱构建脚本
├── streamlit_app.py # 最终的Web UI应用脚本
│
└── (运行后生成)
├── bencao_kg_local.gexf # 知识图谱文件
└── [药材名]_subgraph.png # 示例可视化子图
中国智能体开发者社区，聚焦智能体与大模型开发，提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动，促进经验交流与协作，助力开发者快速构建创新智能应用。
更多推荐
24
0- 0
已为社区贡献3条内容
所有评论(0)