5.2 智能问答系统实现

5.2 智能问答系统实现 — GraphRAG 驱动的多轮对话问答引擎

本节导读：基于 5.1 节搭建的企业知识库，本节将实现一个支持多轮对话的智能问答系统。你将学会如何将 GraphRAG 检索结果无缝嵌入 LLM 对话链路，实现意图识别、追问澄清和答案溯源。

学习目标

• 设计支持多轮对话的 GraphRAG 问答系统架构
• 实现意图识别与追问澄清机制
• 构建 GraphRAG 检索结果到 LLM 上下文的格式化管道
• 完成答案溯源与置信度评估模块

核心概念

智能问答系统是 GraphRAG 最直接的价值出口。传统的单轮问答只接受一个独立问题，返回一段文本。而企业级问答系统需要处理上下文依赖（"它支持哪些格式？"需要知道"它"指什么）、追问（"那技术部呢？"）和歧义消解（"服务器"可能指硬件也可能指软件服务）。

上图为多轮对话问答系统的交互流程。关键设计点在于：对话管理器维护会话历史用于指代消解，GraphRAG 检索提供结构化+非结构化的混合上下文，答案溯源模块标注信息来源增强可信度。

环境准备

在 5.1 节环境基础上，额外需要：

• LangChain 或 LlamaIndex（对话链路编排）
• OpenAI API 或兼容的本地 LLM（如 Ollama + Qwen2.5）
• Redis（会话状态缓存，可选）

pip install langchain openai redis
​

分步实战

步骤 1：对话状态管理

多轮对话的核心是维护对话状态，包括历史消息、当前意图、追问上下文等。

from dataclasses import dataclass, field
from typing import List, Optional, Dict
from enum import Enum
import time

class IntentType(Enum):
    KNOWLEDGE_QUERY = "knowledge_query"      # 知识查询
    CLARIFICATION = "clarification"          # 澄清追问
    CHITCHAT = "chitchat"                    # 闲聊
    FEEDBACK = "feedback"                     # 反馈（好/不好）
    UNSUPPORTED = "unsupported"              # 不支持

@dataclass
class Message:
    role: str          # "user" 或 "assistant"
    content: str
    timestamp: float = field(default_factory=time.time)
    metadata: dict = field(default_factory=dict)

@dataclass
class ConversationState:
    """对话状态管理器"""
    session_id: str
    messages: List[Message] = field(default_factory=list)
    resolved_entities: List[str] = field(default_factory=list)  # 已解析实体
    last_intent: Optional[IntentType] = None
    confidence: float = 0.0
    turn_count: int = 0
    
    def add_message(self, role: str, content: str, metadata: dict = None):
        self.messages.append(Message(role=role, content=content, metadata=metadata or {}))
        self.turn_count += 1
    
    def get_context_window(self, max_turns: int = 5) -> str:
        """获取最近 N 轮对话作为上下文"""
        recent = self.messages[-max_turns * 2:]  # 每轮2条消息
        parts = []
        for msg in recent:
            prefix = "用户" if msg.role == "user" else "助手"
            parts.append(f"{prefix}: {msg.content}")
        return "\n".join(parts)
    
    def get_last_user_query(self) -> str:
        """获取最近一条用户消息"""
        for msg in reversed(self.messages):
            if msg.role == "user":
                return msg.content
        return ""
​

对话状态管理器的设计原则是：轻量级、无外部依赖（纯内存）、支持序列化（方便存入 Redis）。每轮对话产生两条消息（用户+助手），状态管理器自动追踪轮数和意图变化。

步骤 2：意图识别与指代消解

意图识别决定了系统走哪条处理路径。在 GraphRAG 问答场景中，核心区分是"知识查询"和"闲聊"。

import re

class IntentRecognizer:
    """意图识别与指代消解引擎"""
    
    KNOWLEDGE_PATTERNS = [
        r'(什么|哪些|怎么|如何|为什么|哪|几|多少)',
        r'(是|是不是|是否|有没有|能不能|可以)',
        r'(介绍|说明|解释|对比|区别|联系)',
        r'(谁|哪个部门|什么人|谁负责)',
    ]
    CHITCHAT_PATTERNS = [
        r'^(你好|hello|hi|嗨|在吗)',
        r'^(谢谢|感谢|thanks)',
        r'^(再见|bye|拜拜)',
    ]
    FEEDBACK_PATTERNS = [
        r'^(不对|不是|错了|不准确)',
        r'^(对的|正确|就是|没错)',
        r'^(没用|没帮助|不够)',
    ]
    
    def recognize(self, text: str, state: ConversationState = None) -> IntentType:
        """识别用户意图"""
        # 优先匹配反馈类（用户评价上一轮回答）
        for pattern in self.FEEDBACK_PATTERNS:
            if re.search(pattern, text):
                return IntentType.FEEDBACK
        
        # 匹配闲聊类
        for pattern in self.CHITCHAT_PATTERNS:
            if re.search(pattern, text):
                return IntentType.CHITCHAT
        
        # 匹配知识查询类
        for pattern in self.KNOWLEDGE_PATTERNS:
            if re.search(pattern, text):
                return IntentType.KNOWLEDGE_QUERY
        
        # 短文本可能是追问或指代
        if len(text) < 10 and state and state.last_intent == IntentType.KNOWLEDGE_QUERY:
            return IntentType.KNOWLEDGE_QUERY
        
        return IntentType.KNOWLEDGE_QUERY  # 默认按知识查询处理
    
    def resolve_reference(self, text: str, state: ConversationState) -> str:
        """指代消解：将代词替换为实际实体"""
        resolved = text
        
        if not state or not state.resolved_entities:
            return resolved
        
        # 常见指代词替换
        reference_map = {
            "它": state.resolved_entities[-1] if state.resolved_entities else "它",
            "那个": state.resolved_entities[-1] if state.resolved_entities else "那个",
        }
        
        for ref, entity in reference_map.items():
            if ref in resolved and len(resolved) < 20:
                resolved = resolved.replace(ref, entity)
        
        # "X呢？" 模式：追问其他实体
        follow_pattern = re.match(r'(.+?)(?:呢|呢？)', resolved)
        if follow_pattern and state.resolved_entities:
            previous_query = state.get_last_user_query()
            # 提取上一轮问题的模板（去掉实体部分）
            template = re.sub(r'(.+?)(?:的|用|负责|属于)', '', previous_query)
            resolved = follow_pattern.group(1) + template
        
        return resolved
​

指代消解是中文问答系统中的经典难题。上述实现采用了基于规则的轻量方案：维护已解析实体列表，对短文本中的代词进行替换。对于复杂场景（如"张三和他同事的那个项目"），需要结合句法分析或 LLM 来处理，但企业内部知识库场景下，基于规则的方法已经能覆盖 80% 的追问场景。

步骤 3：GraphRAG 上下文格式化

将 GraphRAG 检索结果格式化为 LLM 可理解的结构化上下文。

class ContextFormatter:
    """GraphRAG 检索结果格式化器"""
    
    def format_for_llm(self, graph_results: List[dict],
                       vector_results: List[dict],
                       query: str) -> str:
        """将检索结果格式化为 LLM 提示上下文"""
        parts = []
        
        # 1. 知识图谱路径结果
        if graph_results:
            parts.append("【知识图谱检索结果】")
            for i, r in enumerate(graph_results[:5]):
                path_desc = " -> ".join(r.get("path", []))
                parts.append(
                    f"  路径{i+1}: {r['source']} -> {r['target']}\n"
                    f"    关系链: {path_desc}\n"
                    f"    相关度: {r['score']:.2f}"
                )
        
        # 2. 文档检索结果
        if vector_results:
            parts.append("\n【文档检索结果】")
            for i, r in enumerate(vector_results[:3]):
                parts.append(
                    f"  文档{i+1}: {r['text'][:500]}\n"
                    f"    相关度: {r['score']:.2f}"
                )
        
        if not parts:
            return "未找到相关知识。"
        
        context = "\n".join(parts)
        
        # 添加系统指令
        system_prompt = (
            f"你是一个企业知识库问答助手。请根据以下检索结果回答用户的问题。\n"
            f"如果检索结果中没有相关信息，请诚实说明。\n"
            f"回答时请引用来源路径或文档编号。\n\n"
            f"用户问题: {query}\n\n"
            f"检索上下文:\n{context}\n\n"
            f"请用简洁专业的中文回答:"
        )
        
        return system_prompt
    
    def extract_sources(self, graph_results: List[dict],
                        vector_results: List[dict]) -> List[dict]:
        """提取答案溯源信息"""
        sources = []
        for r in graph_results[:3]:
            sources.append({
                "type": "knowledge_graph",
                "source": r.get("source", ""),
                "target": r.get("target", ""),
                "path": r.get("path", []),
                "score": r.get("score", 0)
            })
        for r in vector_results[:2]:
            sources.append({
                "type": "document",
                "text_preview": r.get("text", "")[:200],
                "score": r.get("score", 0)
            })
        return sources
​

格式化的关键在于信息分层：知识图谱路径放在最前面（结构化信息更可靠），文档片段作为补充。每条结果都附带相关度分数，帮助 LLM 判断信息可信度。系统指令明确要求 LLM 在回答中引用来源，这是企业问答系统的核心要求——每个回答都必须有据可查。

步骤 4：LLM 对话引擎

将检索上下文注入 LLM，生成最终回答。

from openai import OpenAI

class DialogueEngine:
    """基于 LLM 的对话引擎"""
    
    def __init__(self, api_key: str = None, base_url: str = None,
                 model: str = "qwen2.5-72b-instruct"):
        self.client = OpenAI(api_key=api_key or "your-key",
                           base_url=base_url or "http://localhost:11434/v1")
        self.model = model
    
    def generate_answer(self, formatted_context: str) -> str:
        """调用 LLM 生成回答"""
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是企业知识库问答助手，用中文回答。"},
                    {"role": "user", "content": formatted_context}
                ],
                temperature=0.3,    # 低温度保证回答稳定
                max_tokens=1024
            )
            return response.choices[0].message.content.strip()
        except Exception as e:
            return f"回答生成失败: {str(e)}"
    
    def evaluate_confidence(self, answer: str, context: str) -> float:
        """评估回答置信度"""
        if "未找到" in answer or "无法确定" in answer:
            return 0.2
        if "根据检索结果" in answer or "知识图谱显示" in answer:
            return 0.8
        # 检查回答是否包含检索结果中的关键实体
        overlap = sum(1 for word in answer if word in context)
        return min(0.9, 0.5 + overlap * 0.01)
​

LLM 温度设为 0.3 而非默认的 0.7 或 1.0，这是因为企业知识问答需要稳定、可重复的回答，不需要创造性发散。置信度评估通过启发式规则实现：回答中包含检索来源关键词的置信度高，明确说"不知道"的置信度低。

步骤 5：问答系统集成

将以上所有模块组装为完整的问答服务。

class GraphRAGQAService:
    """GraphRAG 智能问答系统"""
    
    def __init__(self, kg_builder, engine, llm_engine=None):
        self.kg = kg_builder
        self.graphrag = engine
        self.llm = llm_engine or DialogueEngine()
        self.intent_recognizer = IntentRecognizer()
        self.formatter = ContextFormatter()
        self.sessions: Dict[str, ConversationState] = {}
    
    def get_or_create_session(self, session_id: str) -> ConversationState:
        if session_id not in self.sessions:
            self.sessions[session_id] = ConversationState(session_id=session_id)
        return self.sessions[session_id]
    
    def chat(self, session_id: str, user_input: str) -> dict:
        """处理一条用户消息"""
        state = self.get_or_create_session(session_id)
        state.add_message("user", user_input)
        
        # 1. 意图识别
        intent = self.intent_recognizer.recognize(user_input, state)
        
        # 2. 指代消解
        resolved_query = self.intent_recognizer.resolve_reference(user_input, state)
        
        # 3. 根据意图路由处理
        if intent == IntentType.CHITCHAT:
            response = "您好！我是企业知识库助手，有什么可以帮您查询的？"
        elif intent == IntentType.FEEDBACK:
            response = "感谢您的反馈，我们会持续优化知识库内容。"
        else:
            # GraphRAG 检索
            graph_results = self.graphrag._graph_search(resolved_query, top_k=5)
            vector_results = self.graphrag._vector_search(resolved_query, top_k=3)
            
            # 格式化上下文
            formatted = self.formatter.format_for_llm(
                graph_results, vector_results, resolved_query
            )
            
            # LLM 生成回答
            response = self.llm.generate_answer(formatted)
            
            # 提取溯源
            sources = self.formatter.extract_sources(graph_results, vector_results)
            state.resolved_entities = list(set(
                [r.get("source", "") for r in graph_results] +
                [r.get("target", "") for r in graph_results]
            ))
            state.confidence = self.llm.evaluate_confidence(response, formatted)
        
        state.last_intent = intent
        state.add_message("assistant", response)
        
        return {
            "answer": response,
            "intent": intent.value,
            "resolved_query": resolved_query,
            "confidence": state.confidence,
            "turn_count": state.turn_count
        }

# 使用示例
if __name__ == "__main__":
    kg = KnowledgeGraphBuilder("bolt://localhost:7687", "neo4j", "password123")
    engine = EnterpriseGraphRAGEngine(kg_builder=kg)
    service = GraphRAGQAService(kg_builder=kg, engine=engine)
    
    # 多轮对话演示
    session_id = "demo_session_001"
    
    r1 = service.chat(session_id, "技术部使用什么数据库？")
    print(f"助手: {r1['answer']}")
    
    r2 = service.chat(session_id, "运维中心呢？")
    print(f"助手: {r2['answer']}（指代消解后查询: {r2['resolved_query']}）")
    
    r3 = service.chat(session_id, "谢谢")
    print(f"助手: {r3['answer']}")
​

常见问题 FAQ

Q1：多轮对话中的上下文窗口应该设多大？

A：建议保留最近 5-8 轮对话（10-16 条消息）。窗口太大会导致 LLM 提示过长影响响应速度，窗口太小则丢失上下文。企业场景下，用户通常在 3-5 轮内解决问题，因此 5 轮窗口足够。如果需要更长上下文（如复杂技术讨论），可以配合会话摘要机制压缩早期对话。

Q2：指代消解准确率不够怎么办？

A：规则方案的准确率天花板在 70-80%。要进一步提升，有三个方向：第一是用 LLM 做指代消解（在正式检索前加一轮 LLM 调用，让 LLM 解释代词含义），第二是在对话管理器中维护实体上下文栈（每轮追踪当前讨论的实体），第三是对高频追问模式做模板化（如"X呢"总是追问同类型的其他实体）。三种方案可以组合使用。

Q3：如何处理 LLM 幻觉问题？

A：在 GraphRAG 问答场景中，LLM 幻觉主要表现为"编造检索结果中不存在的信息"。核心对策是限制 LLM 的自由度：在提示词中明确要求"仅根据检索结果回答，不得补充外部信息"，并设置低温度参数（0.2-0.3）。此外，答案溯源标注让用户可以验证回答依据。

Q4：问答系统的响应延迟如何优化？

A：优化方向有三个：第一，GraphRAG 检索并行化（图检索和向量检索同时发起）；第二，LLM 推理使用流式输出（FastAPI 的 StreamingResponse）；第三，高频问题加缓存（Redis 缓存检索结果，相似问题命中缓存直接返回）。实测优化后，端到端延迟可从 3-5 秒降至 1-2 秒。

Q5：如何评估问答系统的质量？

A：建议从三个维度评估：准确率（答案是否正确）、覆盖率（能回答的比例）、满意度（用户反馈）。准确率需要标注测试集（50-100 组问答对），覆盖率通过统计"未找到相关信息"的频率，满意度通过用户反馈按钮收集。三维度综合得分 = 0.4准确率 + 0.3覆盖率 + 0.3*满意度。

最佳实践与避坑

实践 1：答案必须溯源。企业问答系统的回答如果没有信息来源，可信度为零。每条回答都必须标注来源路径或文档编号。这不仅是技术要求，更是合规要求——企业内部信息查询需要可审计。

实践 2：优雅降级。当 GraphRAG 检索不到结果时，不要返回冷冰冰的"未找到"，而是提供替代建议：推荐相近问题、引导用户换关键词、建议联系人工客服。降级体验决定了用户对系统的信任度。

坑点 1：会话状态内存泄漏。长期运行的问答服务如果不清理过期会话，内存会持续增长。建议设置会话过期时间（30 分钟无操作自动清理）和最大会话数上限。

坑点 2：LLM 调用超时。大语言模型推理可能需要数秒甚至超时。必须在 LLM 调用处设置合理超时（建议 15 秒），超时后返回"思考超时，请稍后重试"的友好提示，同时记录超时日志用于优化。

坑点 3：并发会话冲突。如果同一用户在不同窗口打开问答界面，两个会话共享同一个 session_id 会导致对话上下文混乱。建议用窗口级唯一 ID（如 UUID）作为 session_id，而非用户级 ID。

本节小结

本节实现了一个完整的 GraphRAG 驱动的多轮对话问答系统。核心设计包括：对话状态管理器维护多轮上下文，意图识别器区分知识查询与闲聊，指代消解引擎处理中文代词和追问，GraphRAG 检索结果格式化为 LLM 可消化的结构化提示，答案溯源模块标注信息来源增强可信度。

5.3 节将介绍 GraphRAG 系统的高级优化技巧，包括多层次上下文扩展、自适应检索权重和性能监控体系。

关键词：GraphRAG知识图谱增强, 智能问答, 多轮对话, 意图识别, 指代消解, 答案溯源, LLM对话, 问答引擎

难度：进阶

预计阅读：20 分钟