知识图谱在智能问答系统中的应用

在人工智能领域，智能问答系统一直是研究的热点与商业落地的焦点。从早期的基于规则匹配到如今的深度学习驱动的语义理解，问答系统经历了漫长的演进历程。而**知识图谱（Knowledge Graph）**的引入，为问答系统带来了前所未有的结构化推理能力，使得机器能够真正“理解”用户问题的语义，并给出准确、可解释的答案。本文将深入探讨知识图谱在智能问答系统中的应用，从技术演进、核心原理到具体实现进行完整剖析。

一、智能问答系统的演进历程

1.1 基于规则匹配的初代问答系统

早期的问答系统主要依赖于**模式匹配（Pattern Matching）和模板填充（Template Filling）**技术。代表性的系统包括 ELIZA 和 ALICE，它们通过维护大量的规则库和问答对模板，根据用户输入的关键词匹配预定义的回复规则。

# 简单的基于规则匹配的问答示例
class RuleBasedQA:
    def __init__(self):
        self.rules = [
            (["天气", "怎么样"], "今天天气晴朗，气温25°C，适合外出。"),
            (["温度", "多少"], "当前温度为25摄氏度。"),
            (["下雨", "带伞"], "今天有小雨，建议带伞出门。"),
        ]
    
    def answer(self, question):
        for keywords, response in self.rules:
            if all(kw in question for kw in keywords):
                return response
        return "抱歉，我无法理解您的问题。"

这种方式的优点是回答可控、可解释；缺点是能力有限，无法处理开放域问题，且维护成本随规则数量指数增长。

1.2 基于检索的问答系统

随着互联网的兴起，**检索式问答（Retrieval-based QA）**成为主流。系统从大规模文档库中检索与问题相关的文本片段，再从中提取答案。典型代表包括 IBM Watson 的 DeepQA 架构，以及后来的开放域问答系统。

# 检索式问答的核心流程
def retrieval_qa_system(question, document_corpus, embedder):
    # 1. 问题编码
    question_vector = embedder.encode(question)
    
    # 2. 文档检索
    scores = []
    for doc in document_corpus:
        doc_vector = embedder.encode(doc)
        similarity = cosine_similarity(question_vector, doc_vector)
        scores.append((doc, similarity))
    
    # 3. 取Top-K相关文档
    top_docs = sorted(scores, key=lambda x: x[1], reverse=True)[:5]
    
    # 4. 答案抽取（使用阅读理解模型）
    context = "\n".join([doc for doc, _ in top_docs])
    answer = reading_comprehension_model(question, context)
    
    return answer

检索式系统解决了规则系统的局限，能够处理开放域问题，但答案质量依赖文档覆盖度，且缺乏深层推理能力。

1.3 基于知识图谱的问答系统（KBQA）

**知识图谱问答（Knowledge Base Question Answering, KBQA）**代表了问答系统的新范式。KBQA 以结构化的知识图谱作为知识源，通过语义解析将自然语言问题转换为对图谱的查询操作，直接返回精准的实体、关系或聚合答案。

KBQA 的核心优势体现在：

结构化推理：能够进行多跳推理、逻辑组合等复杂查询
答案精确：返回的是图谱中的具体实体或数值，而非文本片段
可解释性强：答案路径清晰可追溯
语义理解深：能够理解实体、关系、属性等语义要素

二、知识图谱如何赋能智能问答

2.1 知识图谱的结构化优势

知识图谱以**三元组（Triple）**作为基本单元，形式化为 Subject - Predicate - Object 或写作 (头实体, 关系, 尾实体)。这种结构天然适合表示实体间的语义关联。

例如，“张三的老板是李四”这一事实可以表示为：

1
2
3

(张三, 老板, 李四)
(李四, 任职公司, 阿里巴巴)
(阿里巴巴, 创始人, 马云)

这种结构化表示使得系统能够：

精确匹配实体和关系：避免自然语言的歧义性
进行路径推理：沿着关系边进行多跳查询
执行聚合计算：对满足条件的实体进行计数、求和等操作
发现隐含知识：通过推理发现直接查询未表达的知识

2.2 KBQA 的核心处理流程

一个完整的 KBQA 系统通常包含以下模块：

1	用户问题 → 意图识别 → 语义解析 → 图谱查询 → 答案生成 → 结果返回

意图识别判断用户问题是否属于知识图谱可回答的类型；语义解析将问题转化为结构化的查询语句（如 SPARQL、Cypher）；图谱查询执行查询并获取结果；答案生成将结果组织成自然语言返回给用户。

# KBQA 核心流程的简化实现
class KBQASystem:
    def __init__(self, kg_client):
        self.kg_client = kg_client
        self.parser = SemanticParser()
        self.generator = AnswerGenerator()
    
    def answer(self, question):
        # 意图识别
        intent = self.recognize_intent(question)
        if intent == "entity_query":
            # 语义解析：将问题转为SPARQL
            query = self.parser.parse(question, intent)
            # 执行查询
            results = self.kg_client.execute(query)
            # 生成答案
            answer = self.generator.generate(question, results)
            return answer
        elif intent == "comparison":
            # 处理比较类问题
            pass
        elif intent == "count":
            # 处理计数类问题
            pass
        else:
            return "该问题无法通过知识图谱回答"

三、知识图谱问答 vs 向量检索问答

当前工业界主要有两种知识问答方案：基于知识图谱的结构化问答和基于向量数据库的语义检索问答（也称为 RAG，Retrieval-Augmented Generation）。理解两者的差异对于技术选型至关重要。

3.1 技术原理对比

维度	知识图谱问答 (KBQA)	向量检索问答 (RAG)
知识表示	结构化的图谱（实体、关系、三元组）	非结构化的文本块（Chunks）
查询方式	结构化查询语言（SPARQL/Cypher）	向量相似度搜索（ANN）
推理能力	支持多跳推理、逻辑组合	依赖 LLM 的推理能力
答案准确性	高（精确匹配）	中等（可能有幻觉）
可解释性	强（答案路径清晰）	中（依赖检索结果）
更新难度	需要图谱更新和重构	只需追加新文档
部署成本	较高（图数据库 + 解析引擎）	较低（向量数据库 + LLM）

3.2 典型场景对比

KBQA 更适合的场景：

需要精确答案的查询：“公司的 CEO 是谁？”
多跳推理问题：“张三的老板的老板是谁？”
聚合统计问题：“有多少员工在2024年获得晋升？”
需要强可解释性的场景：金融风控、医疗诊断

RAG 更适合的场景：

开放域知识问答：“请介绍一下量子计算的基本原理”
需要生成式答案的场景：“用通俗语言解释什么是区块链”
文档问答：“根据这份合同，违约金如何计算？”
知识频繁更新的场景：新闻问答、产品咨询

3.3 代码对比：两种查询方式的实现

# KBQA 方式：查询知识图谱
def kbqa_answer(question, kg_client):
    """知识图谱问答"""
    # 语义解析示例：谁是张三的老板？
    parsed = {
        "head_entity": "张三",
        "relation": "老板",
        "query_type": "entity_query"
    }
    
    # 构建 SPARQL 查询
    sparql = f"""
    PREFIX ex: <http://example.org/>
    SELECT ?boss WHERE {{
        ex:张三 ex:老板 ?boss .
    }}
    """
    
    result = kg_client.query(sparql)
    return result[0]['boss']  # 返回：李四


# RAG 方式：向量检索 + LLM 生成
def rag_answer(question, vector_store, llm):
    """检索增强生成问答"""
    # 向量检索
    query_embedding = embedding_model.encode(question)
    relevant_chunks = vector_store.search(query_embedding, top_k=5)
    
    # 构建 Prompt
    context = "\n".join([chunk.text for chunk in relevant_chunks])
    prompt = f"""基于以下上下文回答问题。
    
    上下文：
    {context}
    
    问题：{question}
    """
    
    # LLM 生成
    response = llm.generate(prompt)
    return response

四、KBQA 的核心技术实现

4.1 语义解析器设计

语义解析是 KBQA 的核心模块，负责将自然语言问题转换为结构化查询。当前主流方法包括：

基于序列到序列的神经语义解析：

import torch
import torch.nn as nn
from transformers import T5ForConditionalGeneration

class NeuralSemanticParser:
    def __init__(self, model_name="t5-small"):
        self.model = T5ForConditionalGeneration.from_pretrained(model_name)
        self.tokenizer = T5Tokenizer.from_pretrained(model_name)
    
    def parse(self, question, schema):
        """
        将自然语言问题解析为 SPARQL 查询
        schema: 知识图谱的 Schema（实体类型、关系列表）
        """
        # 构建输入提示
        prompt = f"Convert to SPARQL: {question}"
        
        # 编码
        inputs = self.tokenizer(prompt, return_tensors="pt", padding=True)
        
        # 生成查询
        outputs = self.model.generate(
            inputs['input_ids'],
            max_length=128,
            num_beams=5,
            early_stopping=True
        )
        
        sparql = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        return sparql

基于模板的语义解析（更可控、可解释）：

class TemplateBasedParser:
    def __init__(self):
        # 定义问题模板
        self.templates = [
            {
                "pattern": r"(.*)的老板是谁",
                "sparql": "PREFIX ex: <http://example.org/> SELECT ?boss WHERE {{ ex:{head} ex:boss ?boss }}",
                "slot": "head"
            },
            {
                "pattern": r"(.*)有多少(.*)",
                "sparql": "PREFIX ex: <http://example.org/> SELECT COUNT(?item) WHERE {{ ex:{head} ex:{relation} ?item }}",
                "slots": ["head", "relation"]
            },
            {
                "pattern": r"(.*和.*)都是(.*)",
                "sparql": "PREFIX ex: <http://example.org/> SELECT ?item WHERE {{ ?item ex:{relation} ?value . FILTER(?value IN (ex:{val1}, ex:{val2})) }}",
                "slots": ["val1", "val2", "relation"]
            }
        ]
        
        # 实体链接词典
        self.entity词典 = self.load_entity词典()
    
    def parse(self, question):
        for template in self.templates:
            match = re.match(template["pattern"], question)
            if match:
                groups = match.groups()
                # 实体链接：将问句中的实体映射到图谱实体
                linked_entities = [self.link_entity(g) for g in groups]
                
                # 填充模板
                sparql = template["sparql"].format(**{
                    slot: entity 
                    for slot, entity in zip(template.get("slots", ["head"]), linked_entities)
                })
                return sparql
        
        return None
    
    def link_entity(self, mention):
        """实体链接：将文本mention映射到知识图谱实体"""
        # 简化实现，实际使用ELQ或BLINK等模型
        return self.entity词典.get(mention, f"Unknown:{mention}")

4.2 知识图谱查询执行

将语义解析的结果转化为图数据库查询：

from typing import List, Dict, Any

class GraphDBClient:
    """图数据库客户端封装"""
    
    def __init__(self, db_type="neo4j"):
        self.db_type = db_type
        if db_type == "neo4j":
            from neo4j import GraphDatabase
            self.driver = GraphDatabase.driver(
                "bolt://localhost:7687", 
                auth=("neo4j", "password")
            )
        elif db_type == "rdf":
            from rdflib import Graph
            self.graph = Graph()
    
    def execute(self, query: str) -> List[Dict[str, Any]]:
        """执行查询并返回结果"""
        if self.db_type == "neo4j":
            with self.driver.session() as session:
                result = session.run(query)
                return [dict(record) for record in result]
        elif self.db_type == "rdf":
            return list(self.graph.query(query))
    
    def close(self):
        if self.db_type == "neo4j":
            self.driver.close()


# SPARQL 查询构建器
class SPARQLBuilder:
    @staticmethod
    def build_entity_query(entity: str, relation: str) -> str:
        """构建单跳查询"""
        return f"""
        PREFIX ex: <http://example.org/>
        SELECT ?target 
        WHERE {{ 
            ex:{entity} ex:{relation} ?target 
        }}
        """
    
    @staticmethod
    def build_multi_hop_query(entities: List[str], relations: List[str]) -> str:
        """构建多跳查询"""
        assert len(entities) == len(relations) + 1
        
        triples = []
        for i, relation in enumerate(relations):
            triples.append(f"ex:{entities[i]} ex:{relation} ?v{i}")
        
        where_clause = " .\n        ".join(triples)
        return f"""
        PREFIX ex: <http://example.org/>
        SELECT {" ".join([f"?v{i}" for i in range(len(relations))])}
        WHERE {{
            {where_clause}
        }}
        """
    
    @staticmethod
    def build_aggregation_query(
        entity: str, 
        relation: str, 
        agg_func: str = "COUNT"
    ) -> str:
        """构建聚合查询"""
        return f"""
        PREFIX ex: <http://example.org/>
        SELECT ({agg_func}(?target) as ?result)
        WHERE {{
            ex:{entity} ex:{relation} ?target
        }}
        """

4.3 多跳推理的实现

多跳问答是 KBQA 的难点，例如“张三的老板的老板是谁？”需要执行两步查询：

def multi_hop_reasoning(question: str, kg_client: GraphDBClient) -> str:
    """多跳推理问答"""
    # 解析多跳问题
    # 简化实现：假设问题格式固定
    parts = question.replace("？", "?").split("的")
    
    if len(parts) == 3 and "的" in question:
        # 例如：["张三", "老板", "老板是谁"]
        entity = parts[0]
        # 确定跳数（有几个"的"）
        hops = question.count("的")
        
        if hops == 2:
            # 两跳查询
            query = f"""
            PREFIX ex: <http://example.org/>
            SELECT ?boss_of_boss WHERE {{
                ex:{entity} ex:boss ?boss .
                ?boss ex:boss ?boss_of_boss .
            }}
            """
            results = kg_client.execute(query)
            return results[0]['boss_of_boss'] if results else "未找到答案"
    
    return "无法解析该问题"

五、混合架构：KBQA + RAG 的融合

5.1 为什么要混合架构

在实际生产环境中，KBQA 和 RAG 各有优劣，混合架构能够充分发挥两者的优势：

能力	KBQA	RAG	混合架构
精确问答	✅	❌	✅
开放域生成	❌	✅	✅
多跳推理	✅	⚠️	✅
可解释性	✅	⚠️	✅
工程复杂度	高	低	中

5.2 混合架构设计

class HybridQASystem:
    def __init__(self, kg_client, vector_store, llm):
        self.kg_client = kg_client
        self.vector_store = vector_store
        self.llm = llm
        self.classifier = QuestionClassifier()
    
    def answer(self, question: str) -> Dict[str, Any]:
        """混合问答主流程"""
        
        # 1. 问题分类
        question_type = self.classifier.classify(question)
        
        if question_type == "structured":
            # 结构化查询类问题 → KBQA
            return self._kbqa_route(question)
        elif question_type == "comparison":
            # 比较类问题 → 可能需要两者结合
            return self._hybrid_comparison(question)
        elif question_type == "open_domain":
            # 开放域问题 → RAG
            return self._rag_route(question)
        elif question_type == "hybrid":
            # 需要图谱+文档的混合问题
            return self._full_hybrid(question)
        else:
            return {"answer": "无法回答该问题", "source": "none"}
    
    def _kbqa_route(self, question: str) -> Dict[str, Any]:
        """KBQA 路由"""
        parser = SemanticParser()
        query = parser.parse(question)
        
        if query:
            results = self.kg_client.execute(query)
            return {
                "answer": self._format_kg_answer(results),
                "source": "knowledge_graph",
                "query": query
            }
        return {"answer": "无法从知识图谱获取答案", "source": "knowledge_graph"}
    
    def _rag_route(self, question: str) -> Dict[str, Any]:
        """RAG 路由"""
        embeddings = self.embedding_model.encode(question)
        chunks = self.vector_store.search(embeddings, top_k=5)
        
        context = "\n".join([c.text for c in chunks])
        prompt = f"基于上下文回答：{context}\n\n问题：{question}"
        answer = self.llm.generate(prompt)
        
        return {
            "answer": answer,
            "source": "retrieval",
            "chunks": chunks
        }
    
    def _hybrid_comparison(self, question: str) -> Dict[str, Any]:
        """混合比较类问题"""
        # 例如："比较华为和阿里巴巴的营收规模"
        # 图谱提供结构化数据，文档提供详细分析
        company_entities = self._extract_companies(question)
        
        kg_data = {}
        for company in company_entities:
            query = f"SELECT ?revenue WHERE {{ ex:{company} ex:revenue ?revenue }}"
            result = self.kg_client.execute(query)
            if result:
                kg_data[company] = result[0]['revenue']
        
        return {
            "answer": self._generate_comparison(kg_data),
            "source": "hybrid",
            "structured_data": kg_data
        }

5.3 意图分类器实现

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.classifier import LogisticRegression

class QuestionClassifier:
    def __init__(self):
        self.vectorizer = TfidfVectorizer()
        self.classifier = LogisticRegression()
        
        # 训练数据
        self.train_data = [
            ("谁是张三的老板", "structured"),
            ("华为的 CEO 是谁", "structured"),
            ("2024年公司营收是多少", "structured"),
            ("张三的老板的老板是谁", "structured"),
            ("解释一下什么是机器学习", "open_domain"),
            ("量子计算的应用场景有哪些", "open_domain"),
            ("比较华为和阿里巴巴的市值", "comparison"),
            ("腾讯和阿里巴巴哪个更大", "comparison"),
            ("阿里巴巴的员工有多少", "structured"),
            ("根据这份文档，违约金如何计算", "open_domain"),
        ]
        
        self._train()
    
    def _train(self):
        X, y = zip(*self.train_data)
        X_vec = self.vectorizer.fit_transform(X)
        self.classifier.fit(X_vec, y)
    
    def classify(self, question: str) -> str:
        X = self.vectorizer.transform([question])
        return self.classifier.predict(X)[0]

六、实战：构建一个完整的 KBQA 系统

6.1 系统架构

┌─────────────────────────────────────────────────────────┐
│                      用户问题输入                        │
└─────────────────────┬───────────────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────────────┐
│                    输入预处理模块                         │
│         (分词、拼写检查、标准化、实体标注)                 │
└─────────────────────┬───────────────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────────────┐
│                    问题分类模块                           │
│         (意图识别 + 主题分类 + 复杂度评估)                │
└─────────────────────┬───────────────────────────────────┘
                      │
          ┌───────────┴───────────┐
          │                       │
          ▼                       ▼
┌─────────────────┐     ┌─────────────────┐
│   KBQA 路由      │     │   RAG 路由       │
│ (结构化查询)     │     │ (语义检索)       │
└────────┬────────┘     └────────┬────────┘
         │                       │
         ▼                       ▼
┌─────────────────┐     ┌─────────────────┐
│  语义解析器      │     │  检索引擎        │
│ (NL to SPARQL)  │     │  (向量相似度)    │
└────────┬────────┘     └────────┬────────┘
         │                       │
         ▼                       ▼
┌─────────────────┐     ┌─────────────────┐
│  图数据库查询    │     │  LLM 生成        │
│   (Neo4j)       │     │                 │
└────────┬────────┘     └────────┬────────┘
         │                       │
         └───────────┬───────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────────────┐
│                   答案融合与生成模块                      │
└─────────────────────┬───────────────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────────────┐
│                      返回答案                            │
└─────────────────────────────────────────────────────────┘

6.2 完整代码实现

from dataclasses import dataclass
from typing import List, Dict, Optional, Tuple
from enum import Enum
import re

class QuestionType(Enum):
    ENTITY_QUERY = "entity_query"           # 实体查询
    RELATION_QUERY = "relation_query"       # 关系查询
    MULTI_HOP = "multi_hop"                 # 多跳查询
    AGGREGATION = "aggregation"             # 聚合查询
    COMPARISON = "comparison"               # 比较查询
    OPEN_DOMAIN = "open_domain"             # 开放域
    HYBRID = "hybrid"                       # 混合


@dataclass
class QAResult:
    answer: str
    source: str
    confidence: float
    query: Optional[str] = None
    supporting_data: Optional[Dict] = None


class KnowledgeGraphQA:
    """完整的知识图谱问答系统"""
    
    def __init__(self, kg_client, embedding_model, llm):
        self.kg_client = kg_client
        self.embedding_model = embedding_model
        self.llm = llm
        
        # 子模块
        self.entity_linker = EntityLinker()
        self.relation_extractor = RelationExtractor()
        self.semantic_parser = SemanticParser()
        self.query_executor = QueryExecutor(kg_client)
        self.answer_generator = AnswerGenerator(llm)
    
    def answer(self, question: str) -> QAResult:
        """主问答流程"""
        
        # Step 1: 预处理
        processed = self._preprocess(question)
        
        # Step 2: 实体链接
        entities = self.entity_linker.link(processed['text'])
        
        # Step 3: 关系提取
        relations = self.relation_extractor.extract(processed['text'])
        
        # Step 4: 问题分类
        question_type = self._classify_question(question, entities, relations)
        
        # Step 5: 根据类型选择策略
        if question_type in [QuestionType.ENTITY_QUERY, 
                            QuestionType.RELATION_QUERY,
                            QuestionType.MULTI_HOP,
                            QuestionType.AGGREGATION]:
            return self._handle_structured_query(
                question, entities, relations, question_type
            )
        elif question_type == QuestionType.COMPARISON:
            return self._handle_comparison(question, entities, relations)
        else:
            return self._handle_open_domain(question)
    
    def _preprocess(self, question: str) -> Dict:
        """问题预处理"""
        return {
            'text': question.strip(),
            'original': question,
            'tokens': question.split()
        }
    
    def _classify_question(
        self, 
        question: str, 
        entities: List, 
        relations: List
    ) -> QuestionType:
        """问题分类"""
        question_lower = question.lower()
        
        # 多跳检测
        if question.count("的") >= 2:
            return QuestionType.MULTI_HOP
        
        # 聚合检测
        if any(kw in question_lower for kw in ["多少", "几个", "计数", "总数"]):
            return QuestionType.AGGREGATION
        
        # 比较检测
        if any(kw in question_lower for kw in ["比较", "哪个更大", "哪个小", "差异"]):
            return QuestionType.COMPARISON
        
        # 实体+关系 → 结构化查询
        if entities and relations:
            return QuestionType.RELATION_QUERY
        
        # 纯实体查询
        if entities and not relations:
            return QuestionType.ENTITY_QUERY
        
        return QuestionType.OPEN_DOMAIN
    
    def _handle_structured_query(
        self,
        question: str,
        entities: List,
        relations: List,
        question_type: QuestionType
    ) -> QAResult:
        """处理结构化查询"""
        
        # 语义解析
        query = self.semantic_parser.parse(
            question, entities, relations, question_type
        )
        
        if not query:
            return QAResult(
                answer="无法解析该问题",
                source="kg",
                confidence=0.0
            )
        
        # 执行查询
        results = self.query_executor.execute(query)
        
        # 生成答案
        answer = self.answer_generator.generate_from_kg(
            question, results, question_type
        )
        
        return QAResult(
            answer=answer,
            source="knowledge_graph",
            confidence=0.95,
            query=str(query),
            supporting_data={"entities": entities, "relations": relations}
        )
    
    def _handle_comparison(
        self,
        question: str,
        entities: List,
        relations: List
    ) -> QAResult:
        """处理比较类问题"""
        
        # 提取比较的实体
        comparison_entities = self._extract_comparison_entities(question)
        
        # 从图谱获取数据
        comparison_data = {}
        for entity in comparison_entities:
            query = self._build_simple_query(entity, relations[0] if relations else None)
            results = self.query_executor.execute(query)
            if results:
                comparison_data[entity] = results[0]
        
        # 生成比较答案
        answer = self.answer_generator.generate_comparison(
            question, comparison_data
        )
        
        return QAResult(
            answer=answer,
            source="hybrid",
            confidence=0.85,
            supporting_data=comparison_data
        )
    
    def _handle_open_domain(self, question: str) -> QAResult:
        """处理开放域问题（委托给RAG）"""
        # 此处简化处理，实际应调用RAG模块
        return QAResult(
            answer="该问题适合使用检索增强方式回答",
            source="rag_required",
            confidence=0.0
        )


class EntityLinker:
    """实体链接"""
    
    def __init__(self):
        # 简化的实体词典
        self.entity_dict = {
            "张三": "http://example.org/张三",
            "李四": "http://example.org/李四",
            "华为": "http://example.org/华为",
            "阿里巴巴": "http://example.org/阿里巴巴",
        }
    
    def link(self, text: str) -> List[Dict]:
        """将文本中的实体mention映射到图谱实体"""
        entities = []
        for mention, uri in self.entity_dict.items():
            if mention in text:
                entities.append({
                    "mention": mention,
                    "uri": uri,
                    "type": self._infer_type(uri)
                })
        return entities
    
    def _infer_type(self, uri: str) -> str:
        if "公司" in uri or uri in ["华为", "阿里巴巴"]:
            return "Company"
        elif "人" in uri or uri in ["张三", "李四"]:
            return "Person"
        return "Unknown"


class RelationExtractor:
    """关系提取"""
    
    def __init__(self):
        self.relation_patterns = {
            r"(.*)的老板": "boss",
            r"(.*)的(.*)的老板": "boss",  # 多跳
            r"(.*)的(.*)": "related_to",
            r"有多少": "count",
        }
    
    def extract(self, text: str) -> List[str]:
        relations = []
        for pattern, relation in self.relation_patterns.items():
            if re.search(pattern, text):
                relations.append(relation)
        return list(set(relations))


class SemanticParser:
    """语义解析器"""
    
    def parse(
        self, 
        question: str, 
        entities: List, 
        relations: List,
        question_type: QuestionType
    ) -> Optional[str]:
        """将问题解析为 SPARQL"""
        
        if not entities:
            return None
        
        entity_uri = entities[0]['uri']
        entity_name = entities[0]['mention']
        
        if question_type == QuestionType.MULTI_HOP:
            # 多跳查询
            hops = question.count("的")
            if hops == 2:
                return f"""
                PREFIX ex: <http://example.org/>
                SELECT ?final WHERE {{
                    ex:{entity_name} ex:boss ?v1 .
                    ?v1 ex:boss ?final .
                }}
                """
        
        elif question_type == QuestionType.RELATION_QUERY:
            relation = relations[0] if relations else None
            if relation == "boss":
                return f"""
                PREFIX ex: <http://example.org/>
                SELECT ?boss WHERE {{
                    ex:{entity_name} ex:boss ?boss .
                }}
                """
        
        elif question_type == QuestionType.AGGREGATION:
            return f"""
            PREFIX ex: <http://example.org/>
            SELECT COUNT(?item) WHERE {{
                ex:{entity_name} ex:员工 ?item .
            }}
            """
        
        return None


class QueryExecutor:
    """查询执行器"""
    
    def __init__(self, kg_client):
        self.kg_client = kg_client
    
    def execute(self, sparql: str) -> List[Dict]:
        """执行 SPARQL 查询"""
        try:
            return self.kg_client.execute(sparql)
        except Exception as e:
            print(f"Query execution error: {e}")
            return []


class AnswerGenerator:
    """答案生成器"""
    
    def __init__(self, llm):
        self.llm = llm
    
    def generate_from_kg(
        self, 
        question: str, 
        results: List[Dict],
        question_type: QuestionType
    ) -> str:
        """从图谱结果生成答案"""
        
        if not results:
            return "抱歉，未能找到相关信息。"
        
        if question_type == QuestionType.ENTITY_QUERY:
            return f"答案是：{results[0].get('target', '未知')}"
        elif question_type == QuestionType.MULTI_HOP:
            return f"经过推理，答案是：{results[0].get('final', '未知')}"
        elif question_type == QuestionType.AGGREGATION:
            return f"共有 {results[0].get('result', 0)} 个结果。"
        
        return f"找到 {len(results)} 个相关结果"
    
    def generate_comparison(
        self, 
        question: str, 
        data: Dict
    ) -> str:
        """生成比较答案"""
        if not data:
            return "无法获取比较数据"
        
        parts = []
        for entity, value in data.items():
            parts.append(f"{entity}: {value}")
        
        return "；".join(parts)

6.3 使用示例

# 初始化系统（示例）
def main():
    # 假设已初始化各组件
    # kg_client = GraphDBClient("neo4j")
    # embedding_model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
    # llm = OpenAIModel('gpt-4')
    
    # qa_system = KnowledgeGraphQA(kg_client, embedding_model, llm)
    
    # 示例问题
    questions = [
        "谁是张三的老板？",
        "张三的老板的老板是谁？",
        "华为有多少员工？",
        "比较华为和阿里巴巴的市值",
    ]
    
    # for q in questions:
    #     result = qa_system.answer(q)
    #     print(f"Q: {q}")
    #     print(f"A: {result.answer}")
    #     print(f"Source: {result.source}")
    #     print()
    
    print("KBQA 系统示例运行完成")


if __name__ == "__main__":
    main()

七、总结与展望

知识图谱问答（KBQA）代表了智能问答系统的重要发展方向，它通过结构化的知识表示和精确的查询能力，解决了传统检索式问答的诸多局限。通过语义解析将自然语言转化为图谱查询，KBQA 能够进行多跳推理、逻辑组合等复杂查询，返回精准、可解释的答案。

然而，KBQA 也面临挑战：知识图谱的构建和维护成本高，难以覆盖开放域知识；复杂问题的语义解析仍是难题。因此，KBQA + RAG 的混合架构成为当前的主流方案，能够根据问题类型智能选择最优策略。

未来，随着大语言模型（LLM）技术的发展，KBQA 有望与 LLM 更深度融合：

使用 LLM 增强语义解析能力
利用图谱结构提升 LLM 的推理可解释性
实现动态知识图谱的实时更新与问答

相关标签：知识图谱、智能问答、KBQA、语义搜索、AI

推荐阅读：