考研知识图谱对话系统：作业 4-洪萨配资

KaoyanGraphQASystem（考研知识对话系统）

系统功能

考研知识对话系统是一个智能问答系统，专为考研学生设计。它基于知识图谱和大语言模型，从研招网爬取数据，构建结构化知识图谱，并通过对话管理模块回答用户关于考研院校、专业、就业前景等问题。系统功能分为以下模块：

爬虫模块：
- 功能：从研招网爬取院校信息、专业信息和就业前景信息。
- 输出：原始数据存储在data/raw/目录。
- 示例：爬取院校名称、专业代码、招生人数等。
数据处理模块：
- 功能：清洗和格式化爬取的数据，移除无效值，转换为结构化格式（如JSON或CSV）。
- 输出：处理后数据存储在data/processed/目录。
知识图谱构建模块：
- 功能：从处理后的数据中抽取实体（如院校、专业）和关系（如开设、属于），构建知识图谱。
- 输出：图谱结构文件存储在data/graph/目录。
图数据库存储模块：
- 功能：将知识图谱导入图数据库（支持Neo4j和TuGraph），执行查询操作。
- 示例：查找某院校开设的专业。
大模型集成模块：
- 功能：集成Qwen-plus大模型，用于实体抽取、关系抽取或生成自然语言回答。
- 示例：调用API生成“计算机专业的就业前景”的回答。
对话系统模块：
- 功能：解析用户查询，从图谱检索信息，结合大模型生成回答，并管理对话历史。
- 示例：用户问“清华大学有哪些计算机专业？”，系统解析意图，检索图谱数据，生成回答。

系统支持多种扩展：添加新数据源、集成其他图数据库或大模型，并考虑性能（如爬虫延迟）和安全性（API密钥加密）。

搭建过程

搭建该系统需遵循项目结构，使用Python 3.8+环境。以下是核心步骤和代码示例：

步骤1: 环境设置

依赖安装：创建虚拟环境，安装依赖包。

# 核心依赖 requests>=2.31.0 beautifulsoup4>=4.12.0 lxml>=4.9.0 pandas>=2.0.0 numpy>=1.24.0 # 图数据库客户端 py2neo>=2021.2.4 neo4j>=5.14.0 # TuGraph Python客户端（如果有官方包，需要安装） # 大模型API - 使用LangChain langchain>=0.1.0 langchain-community>=0.0.20 dashscope>=1.17.0 # 阿里云DashScope SDK（Qwen API，LangChain依赖） # 配置和日志 pyyaml>=6.0 python-dotenv>=1.0.0 # Web框架（可选，用于Web接口） flask>=3.0.0 flask-cors>=4.0.0 # 工具库 tqdm>=4.66.0 jieba>=0.42.1 # 中文分词 zhconv>=1.4.2 # 中文繁简转换 # 测试 pytest>=7.4.0 pytest-cov>=4.1.0

requirements.txt包含：requests, beautifulsoup4, pandas, py2neo, dashscope, jieba。

配置文件：编辑config/config.yaml，设置API密钥、数据库凭证等。

# config/config.yaml 示例 neo4j: uri: "bolt://localhost:7687" user: "neo4j" password: "password" qwen: api_key: "your_api_key"

步骤2: 实现核心模块

基于项目结构，编写关键代码。以下是核心代码片段：

爬虫模块（crawler/yan_zhao_wang.py）：

功能：爬取研招网院校信息。

代码示例：

import requests from bs4 import BeautifulSoup import json class YanZhaoWangCrawler: def __init__(self, base_url="https://yz.chsi.com.cn"): self.base_url = base_url def crawl_schools(self): url = f"{self.base_url}/sch/" response = requests.get(url) soup = BeautifulSoup(response.text, 'html.parser') schools = [] for item in soup.select('.school-list li'): name = item.select_one('.name').text.strip() code = item.select_one('.code').text.strip() schools.append({"name": name, "code": code}) return schools def save_data(self, data, filename): with open(f"data/raw/{filename}.json", 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False) # 使用示例 if __name__ == "__main__": crawler = YanZhaoWangCrawler() schools = crawler.crawl_schools() crawler.save_data(schools, "schools")

截图：运行爬虫后，数据保存在data/raw/schools.json文件中

2025-12-13 16:38:43,033 - __main__ - INFO - ================================================== 2025-12-13 16:38:43,033 - __main__ - INFO - 开始爬取研招网数据 2025-12-13 16:38:43,033 - __main__ - INFO - ================================================== 2025-12-13 16:38:43,033 - crawler.yan_zhao_wang - INFO - 开始全面爬取研招网信息... 2025-12-13 16:38:43,033 - crawler.yan_zhao_wang - INFO - 开始爬取院校信息...

数据存放示例

[ { "name": "计算机科学与技术", "code": "0812", "school": "清华大学", "degree_type": "学术型", "research_direction": "人工智能", "exam_subjects": [ "数学一", "英语一", "408计算机学科专业基础" ], "enrollment_count": 50 }, { "name": "计算机科学与技术", "code": "0812", "school": "北京大学", "degree_type": "学术型", "research_direction": "机器学习", "exam_subjects": [ "数学一", "英语一", "408计算机学科专业基础" ], "enrollment_count": 40 },

知识图谱构建模块（knowledge_graph/entity_extraction.py）：

功能：从数据抽取实体。

代码示例：

class EntityExtractor: """实体抽取器""" def __init__(self, entity_types: List[str]): """ 初始化实体抽取器 Args: entity_types: 实体类型列表 """ self.entity_types = entity_types def extract_from_schools(self, schools: List[Dict]) -> List[Dict]: """ 从院校数据中抽取实体 Args: schools: 院校数据列表 Returns: 实体列表，每个实体包含：id, type, name, properties """ entities = [] entity_id = 0 for school in schools: entity = { 'id': f"school_{entity_id}", 'type': '院校', 'name': school.get('name', ''), 'properties': { 'code': school.get('code', ''), 'location': school.get('location', ''), 'type': school.get('type', ''), 'website': school.get('website', '') } } entities.append(entity) entity_id += 1 logger.info(f"从院校数据中抽取了 {len(entities)} 个实体") return entities

截图：实体抽取后的 JSON 文件

{ "entities": [ { "id": "major_0", "type": "专业", "name": "计算机科学与技术", "properties": { "code": "0812", "degree_type": "学术型" } }, { "id": "direction_1", "type": "研究方向", "name": "人工智能", "properties": {} }, { "id": "subject_2", "type": "考试科目", "name": "数学一", "properties": {}

图数据库存储模块（storage/neo4j_client.py）：

功能：将实体导入Neo4j。

代码示例：

class Neo4jClient: """Neo4j客户端""" def __init__(self, uri: str, user: str, password: str, database: str = "neo4j"): """ 初始化Neo4j客户端 Args: uri: Neo4j数据库URI user: 用户名 password: 密码 database: 数据库名称 """ try: self.graph = Graph(uri, auth=(user, password), name=database) self.matcher = NodeMatcher(self.graph) logger.info(f"成功连接到Neo4j: {uri}") except Exception as e: logger.error(f"连接Neo4j失败: {e}") raise def create_entity(self, entity: Dict) -> Node: """ 创建实体节点 Args: entity: 实体字典（包含id, type, name, properties） Returns: 创建的节点 """ node = Node( entity['type'], id=entity['id'], name=entity['name'], **entity.get('properties', {}) ) self.graph.merge(node, entity['type'], 'id') return node def create_relation(self, relation: Dict, source_entity: Dict, target_entity: Dict): """ 创建关系 Args: relation: 关系字典（包含source, target, type, properties） source_entity: 源实体字典 target_entity: 目标实体字典 """ try: source_node = self.matcher.match(source_entity['type'], id=source_entity['id']).first() target_node = self.matcher.match(target_entity['type'], id=target_entity['id']).first() if source_node and target_node: rel = Relationship(source_node, relation['type'], target_node, **relation.get('properties', {})) self.graph.merge(rel) else: logger.warning(f"无法创建关系: 找不到源节点或目标节点") except Exception as e: logger.error(f"创建关系失败: {e}")

截图：运行后Neo4j浏览器界面，导入的节点。

图数据库存储模块（storage/tugraph_client.py）：

功能：将实体导入Tugraph。

代码示例：

class TuGraphClient: """TuGraph客户端""" def __init__(self, host: str, port: int, user: str, password: str, graph_name: str): """ 初始化TuGraph客户端 Args: host: TuGraph服务器地址 port: 端口号 user: 用户名 password: 密码 graph_name: 图名称 """ self.base_url = f"http://{host}:{port}" self.user = user self.password = password self.graph_name = graph_name self.token = None self._login() def _get_headers(self) -> Dict: """获取请求头""" return { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json" } def create_entity(self, entity: Dict): """ 创建实体节点 Args: entity: 实体字典（包含id, type, name, properties） """ try: # TuGraph创建节点的Cypher语句 properties = {k: v for k, v in entity.get('properties', {}).items() if v} props_str = ", ".join([f"{k}: '{v}'" if isinstance(v, str) else f"{k}: {v}" for k, v in properties.items()]) cypher = f""" MERGE (n:{entity['type']} {{id: '{entity['id']}', name: '{entity['name']}'{', ' + props_str if props_str else ''}}}) RETURN n """ self.execute_cypher(cypher) except Exception as e: logger.error(f"创建实体失败 {entity.get('name')}: {e}")

截图：运行后Tugraph浏览器界面，导入的节点。

对话系统模块（chatbot/query_parser.py）：

功能：解析用户查询，处理输入对话中关键词传递给大模型，基于图谱进行问答。

代码示例：

class GraphQA: """图谱问答器""" def _answer_school_question(self, query: str, entities: List[Dict]) -> Dict: """回答院校相关问题""" # 提取院校名称 school_names = [e['name'] for e in entities if e.get('type') == '院校'] if not school_names: # 尝试从查询中提取 if '大学' in query or '学院' in query: school_names = [s for s in query.split() if '大学' in s or '学院' in s] if school_names: school_name = school_names[0] # 查询院校信息 nodes = self.graph_client.find_entity_by_name(school_name, '院校') if nodes: related = self.graph_client.get_related_entities(school_name) answer = f"关于{school_name}的信息：\n" answer += f"已找到相关院校信息。\n" # 添加相关专业 majors = [r for r in related if '专业' in str(r.get('types', []))] if majors: answer += f"该院校开设的专业包括：{', '.join([m['name'] for m in majors[:10]])}\n" evidence = { 'entity': school_name, 'related_entities': related } return {'answer': answer, 'evidence': evidence} return {'answer': "抱歉，我没有找到相关信息。", 'evidence': {}} def _answer_major_question(self, query: str, entities: List[Dict]) -> Dict: """回答专业相关问题""" # 提取专业名称 major_names = [e['name'] for e in entities if e.get('type') == '专业'] if major_names: major_name = major_names[0] # 查询专业信息 nodes = self.graph_client.find_entity_by_name(major_name, '专业') if nodes: related = self.graph_client.get_related_entities(major_name) answer = f"关于{major_name}的信息：\n" # 添加相关院校 schools = [r for r in related if '院校' in str(r.get('types', []))] if schools: answer += f"开设该专业的院校包括：{', '.join([s['name'] for s in schools[:10]])}\n" # 添加研究方向 directions = [r for r in related if '研究方向' in str(r.get('types', []))] if directions: answer += f"研究方向包括：{', '.join([d['name'] for d in directions[:5]])}\n" evidence = { 'entity': major_name, 'related_entities': related } return {'answer': answer, 'evidence': evidence} return {'answer': "抱歉，我没有找到相关信息。", 'evidence': {}}

class QueryParser: """查询解析器""" def __init__(self): """初始化查询解析器""" # 定义实体类型关键词 self.entity_keywords = { '院校': ['大学', '学院', '学校', '高校', '院校'], '专业': ['专业', '学科', '方向', '学硕', '专硕'], '研究方向': ['方向', '研究方向', '研究领域'], '考试科目': ['科目', '考试', '专业课', '公共课'], '就业前景': ['就业', '前景', '出路', '薪资', '工资'] } # 定义关系类型关键词 self.relation_keywords = { '开设': ['开设', '有', '提供', '招收'], '属于': ['属于', '是', '在'], '包含': ['包含', '包括', '有'], '研究方向': ['研究方向', '方向是', '研究'], '考试科目': ['考', '考试', '科目'], '就业方向': ['就业', '去向', '从事'] } def parse(self, query: str) -> Dict: """ 解析用户查询 Args: query: 用户查询文本 Returns: 解析结果字典，包含： - entities: 识别的实体列表 - intent: 查询意图 - keywords: 关键词列表 """ # 分词 words = jieba.cut(query) keywords = [w for w in words if len(w) > 1] # 过滤单字 # 识别实体 entities = self._extract_entities(query, keywords) # 识别意图 intent = self._extract_intent(query, keywords) return { 'original_query': query, 'keywords': keywords, 'entities': entities, 'intent': intent }

截图：运行解析器后，截图输出结果。

步骤3: 集成和测试

主程序（main.py）：集成所有模块。

def main(): """主函数""" parser = argparse.ArgumentParser(description='考研知识对话系统') parser.add_argument( '--mode', type=str, choices=['crawl', 'process', 'build_graph', 'import_neo4j', 'import_tugraph', 'chat', 'all', 'generate_sample'], default='chat', help='运行模式' ) parser.add_argument( '--use_sample', action='store_true', help='使用示例数据（在crawl或all模式时）' ) parser.add_argument( '--config', type=str, default='config/config.yaml', help='配置文件路径' )

测试：运行tests/中的单元测试，确保各模块功能正常。
```
pytest tests/
```
截图：截图用户输入问题和系统回答。

步骤4: 部署和扩展

安装Python依赖：使用命令pip install -r requirements.txt可以安装项目所需的所有Python包。
配置数据库：
- Neo4j数据库：如果您选择使用Neo4j，需要先安装并启动它。配置信息（如数据库URL、用户名和密码）在config/config.yaml文件中设置。
- TuGraph数据库：同样，安装并启动TuGraph后，在config/config.yaml中配置相关参数。
  这两个数据库是可选的，如果您不使用它们，可以跳过这一步。
配置API：在config/config.yaml文件中，添加您的Qwen-plus API密钥。这确保项目能正确调用API服务。
三种操作模式：
- 爬取数据：运行python main.py --mode crawl启动数据爬取功能。
- 构建知识图谱：运行python main.py --mode build_graph根据爬取的数据构建知识图谱。
- 启动对话系统：运行python main.py --mode chat进入对话模式，与系统交互。