KaoyanGraphQASystem（考研知识对话系统）

系统功能

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

1. 爬虫模块：

   • 功能：从研招网爬取院校信息、专业信息和就业前景信息。
   • 输出：原始数据存储在 data/raw/ 目录。
   • 示例：爬取院校名称、专业代码、招生人数等。

2. 数据处理模块：

   • 功能：清洗和格式化爬取的数据，移除无效值，转换为结构化格式（如JSON或CSV）。
   • 输出：处理后数据存储在 data/processed/ 目录。

3. 知识图谱构建模块：

   • 功能：从处理后的数据中抽取实体（如院校、专业）和关系（如开设、属于），构建知识图谱。
   • 输出：图谱结构文件存储在 data/graph/ 目录。

4. 图数据库存储模块：

   • 功能：将知识图谱导入图数据库（支持Neo4j和TuGraph），执行查询操作。
   • 示例：查找某院校开设的专业。

5. 大模型集成模块：

   • 功能：集成Qwen-plus大模型，用于实体抽取、关系抽取或生成自然语言回答。
   • 示例：调用API生成“计算机专业的就业前景”的回答。

6. 对话系统模块：

   • 功能：解析用户查询，从图谱检索信息，结合大模型生成回答，并管理对话历史。
   • 示例：用户问“清华大学有哪些计算机专业？”，系统解析意图，检索图谱数据，生成回答。

系统支持多种扩展：添加新数据源、集成其他图数据库或大模型，并考虑性能（如爬虫延迟）和安全性（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: 实现核心模块

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

1. 爬虫模块（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
       },
     

2. 知识图谱构建模块（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": {}

3. 图数据库存储模块（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浏览器界面，导入的节点。

4. 图数据库存储模块（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浏览器界面，导入的节点。

6. 对话系统模块（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 进入对话模式，与系统交互。

总结

该系统从数据爬取到对话生成，实现了全流程自动化。搭建过程涉及环境配置、模块编码和测试。