5.1 智能问答系统实战

5.1 智能问答系统实战

本节导读：本节将从零开始，基于LightRAG构建一个完整的智能问答系统。涵盖环境搭建、数据准备、知识图谱构建、向量索引建立、检索管道设计、问答接口开发以及多轮对话和上下文管理的完整流程，帮助读者掌握从原型到可用系统的端到端实现能力。

学习目标

• 掌握基于LightRAG搭建智能问答系统的完整流程
• 学会数据预处理和文档切分的工程化方法
• 理解图构建与向量索引的协同工作机制
• 设计高可用的检索管道和问答接口
• 实现多轮对话与上下文管理功能
• 了解系统的评估方法和优化策略

核心概念

智能问答系统架构概览

基于LightRAG的智能问答系统采用"数据→索引→检索→生成"的四阶段流水线架构，每一层都有明确的职责和可优化空间：

核心模块说明

环境准备 / 前置知识

技术依赖

# requirements.txt
lightrag-hku>=0.2.0       # LightRAG核心库
openai>=1.0.0              # LLM API（或本地模型）
sentence-transformers>=2.2.0  # 向量编码
fastapi>=0.100.0           # Web服务框架
uvicorn>=0.23.0            # ASGI服务器
redis>=5.0.0               # 缓存（可选）
pydantic>=2.0.0            # 数据验证
numpy>=1.24.0              # 数值计算
networkx>=3.0              # 图操作
​

安装与初始化

# 安装LightRAG
# pip install "lightrag-hku[all]"

import asyncio
import json
import logging
from pathlib import Path
from typing import List, Dict, Optional, AsyncIterator
from dataclasses import dataclass, field
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(name)s] %(levelname)s: %(message)s'
)
logger = logging.getLogger("LightRAG-QA")
​

分步实战

步骤 1：项目结构设计

一个结构清晰的问答系统项目应该有良好的模块划分：

lightrag-qa-system/
├── config/
│   └── settings.py          # 全局配置
├── core/
│   ├── document_loader.py   # 文档加载与预处理
│   ├── indexer.py           # 图构建与向量索引
│   ├── retriever.py         # 检索管道
│   └── generator.py         # 答案生成
├── api/
│   ├── app.py               # FastAPI应用
│   └── models.py           # 数据模型
├── data/
│   └── knowledge/           # 知识库文档
├── storage/                 # 持久化存储
└── main.py                  # 启动入口
​

步骤 2：数据准备与文档预处理

文档预处理是问答系统质量的基石，好的预处理能显著提升检索效果。

import re
from pathlib import Path
from typing import List, Dict, Optional
from dataclasses import dataclass

@dataclass
class Document:
    """统一文档数据结构"""
    doc_id: str
    title: str
    content: str
    source: str
    metadata: Dict = field(default_factory=dict)
    chunks: List[str] = field(default_factory=list)

class DocumentPreprocessor:
    """文档预处理器：清洗、分段、元数据提取"""

    def __init__(self, 
                 chunk_size: int = 800, 
                 chunk_overlap: int = 100,
                 min_chunk_size: int = 50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.min_chunk_size = min_chunk_size

    def load_markdown(self, file_path: str) -> Document:
        """加载Markdown文档"""
        path = Path(file_path)
        content = path.read_text(encoding='utf-8')

        # 提取标题
        title = self._extract_title(content)

        return Document(
            doc_id=path.stem,
            title=title,
            content=content,
            source=str(path),
            metadata={
                'format': 'markdown',
                'size': len(content),
                'loaded_at': datetime.now().isoformat()
            }
        )

    def load_txt(self, file_path: str) -> Document:
        """加载纯文本文档"""
        path = Path(file_path)
        content = path.read_text(encoding='utf-8')
        title = path.stem

        return Document(
            doc_id=path.stem,
            title=title,
            content=content,
            source=str(path),
            metadata={'format': 'text'}
        )

    def clean_text(self, text: str) -> str:
        """文本清洗"""
        # 去除多余空白
        text = re.sub(r'\n{3,}', '\n\n', text)
        text = re.sub(r' {2,}', ' ', text)

        # 去除特殊字符（保留中文、英文、标点）
        text = re.sub(r'[^\u4e00-\u9fff\w\s.,;:!?，。；：！？、""''（）\[\]【】\-]', '', text)

        return text.strip()

    def chunk_document(self, doc: Document) -> Document:
        """文档分段：按语义切分为适合检索的文本块"""
        content = self.clean_text(doc.content)

        # 策略1：按段落分割
        paragraphs = [p.strip() for p in content.split('\n\n') if p.strip()]

        # 策略2：合并短段落，拆分长段落
        chunks = []
        current_chunk = ""

        for para in paragraphs:
            if len(current_chunk) + len(para) + 1 < self.chunk_size:
                current_chunk = f"{current_chunk}\n{para}".strip() if current_chunk else para
            else:
                if current_chunk:
                    chunks.append(current_chunk)
                if len(para) > self.chunk_size:
                    # 长段落需要二次切分
                    chunks.extend(self._split_long_text(para))
                else:
                    current_chunk = para

        if current_chunk:
            chunks.append(current_chunk)

        # 过滤过短的chunk
        chunks = [c for c in chunks if len(c) >= self.min_chunk_size]

        doc.chunks = chunks
        logger.info(f"文档 '{doc.title}' 切分为 {len(chunks)} 个文本块")
        return doc

    def _split_long_text(self, text: str) -> List[str]:
        """拆分超长文本（滑动窗口）"""
        chunks = []
        start = 0
        while start < len(text):
            end = start + self.chunk_size
            chunk = text[start:end]

            # 尝试在句号处断句
            if end < len(text):
                last_period = chunk.rfind('。')
                last_period = max(last_period, chunk.rfind('！'))
                last_period = max(last_period, chunk.rfind('？'))
                if last_period > self.chunk_size * 0.5:
                    chunk = text[start:last_period + 1]
                    end = last_period + 1

            chunks.append(chunk)
            start = end - self.chunk_overlap

        return chunks

    def batch_load(self, directory: str) -> List[Document]:
        """批量加载目录下的所有文档"""
        docs = []
        dir_path = Path(directory)

        for file_path in dir_path.rglob("*.md"):
            doc = self.load_markdown(str(file_path))
            doc = self.chunk_document(doc)
            docs.append(doc)

        for file_path in dir_path.rglob("*.txt"):
            doc = self.load_txt(str(file_path))
            doc = self.chunk_document(doc)
            docs.append(doc)

        logger.info(f"共加载 {len(docs)} 个文档，{sum(len(d.chunks) for d in docs)} 个文本块")
        return docs

    def _extract_title(self, content: str) -> str:
        """从Markdown中提取标题"""
        first_line = content.strip().split('\n')[0]
        if first_line.startswith('#'):
            return first_line.lstrip('#').strip()
        return "Untitled"
​

步骤 3：LightRAG初始化与知识库构建

from lightrag import LightRAG, QueryParam
from lightrag.llm import openai_complete_if_cache
from lightrag.utils import EmbeddingFunc
import asyncio

class KnowledgeBaseBuilder:
    """基于LightRAG的知识库构建器"""

    def __init__(self, 
                 working_dir: str = "./storage/lightrag",
                 llm_model: str = "gpt-4o-mini",
                 embedding_model: str = "text-embedding-3-small"):
        self.working_dir = Path(working_dir)
        self.working_dir.mkdir(parents=True, exist_ok=True)

        self.llm_model = llm_model
        self.embedding_model = embedding_model
        self.rag: Optional[LightRAG] = None

    async def init_rag(self):
        """初始化LightRAG实例"""
        self.rag = LightRAG(
            working_dir=str(self.working_dir),
            llm_model_func=openai_complete_if_cache,
            llm_model_name=self.llm_model,
            embedding_func=EmbeddingFunc(
                embedding_func=self._embedding_wrapper,
                embedding_dim=1536
            )
        )
        logger.info("LightRAG实例初始化完成")

    async def _embedding_wrapper(self, texts: List[str]) -> List[List[float]]:
        """Embedding函数包装"""
        import openai
        response = await openai.embeddings.create(
            model=self.embedding_model,
            input=texts
        )
        return [item.embedding for item in response.data]

    async def build_from_documents(self, documents: List[Document]):
        """从文档列表构建知识库"""
        if not self.rag:
            await self.init_rag()

        # 收集所有文本块
        all_texts = []
        for doc in documents:
            for chunk in doc.chunks:
                all_texts.append(chunk)

        logger.info(f"开始构建知识库，共 {len(all_texts)} 个文本块...")

        # 批量插入
        batch_size = 10
        for i in range(0, len(all_texts), batch_size):
            batch = all_texts[i:i + batch_size]
            batch_text = "\n\n".join(batch)

            await self.rag.ainsert(batch_text)

            progress = min(i + batch_size, len(all_texts))
            if progress % 50 == 0 or progress == len(all_texts):
                logger.info(f"进度: {progress}/{len(all_texts)}")

        logger.info("知识库构建完成")

    async def build_from_file(self, file_path: str):
        """从单个文件构建"""
        content = Path(file_path).read_text(encoding='utf-8')
        if not self.rag:
            await self.init_rag()
        await self.rag.ainsert(content)
        logger.info(f"文件 {file_path} 已入库")

    async def query(self, 
                    question: str,
                    mode: str = "hybrid",
                    top_k: int = 10,
                    max_tokens: int = 2048) -> str:
        """执行查询"""
        if not self.rag:
            await self.init_rag()

        param = QueryParam(
            mode=mode,          # local/global/hybrid/mix
            top_k=top_k,
            max_tokens_for_text_chunk=max_tokens
        )

        result = await self.rag.aquery(question, param=param)
        return result
​

步骤 4：检索管道设计

检索管道负责将用户问题转化为高质量的检索上下文：

import hashlib
from typing import List, Dict, Tuple
from dataclasses import dataclass

@dataclass
class RetrievalResult:
    """检索结果数据结构"""
    content: str
    source: str
    score: float
    chunk_id: str
    entities: List[str] = field(default_factory=list)
    relations: List[Dict] = field(default_factory=list)

class QueryProcessor:
    """查询预处理器：改写、扩展、实体提取"""

    def __init__(self):
        self.history_queries: List[str] = []

    def rewrite_query(self, query: str, history: List[str] = None) -> str:
        """查询改写：基于历史上下文补全代词"""
        if not history:
            return query

        last_queries = history[-3:]
        # 简单策略：检测代词并替换
        pronouns = {'它': None, '他': None, '她': None, '这个': None, '那个': None}

        # 提取前一轮查询中的核心实体
        if last_queries:
            previous = last_queries[-1]
            # 基于规则的代词消解（生产环境建议用LLM）
            for pronoun in pronouns:
                if pronoun in query and len(previous) > 5:
                    # 用前一轮查询的前20字作为指代对象
                    query = query.replace(pronoun, previous[:20] + '的相关内容')

        return query

    def extract_query_entities(self, query: str) -> List[str]:
        """从查询中提取关键实体（简化版）"""
        # 实际项目应使用NER模型
        entities = []
        import re
        # 提取中文实体（连续的中文字符序列作为候选）
        candidates = re.findall(r'[\u4e00-\u9fff]{2,10}', query)
        # 过滤常见停用词
        stopwords = {'什么', '怎么', '如何', '为什么', '哪些', '哪个', '是否', '可以'}
        entities = [c for c in candidates if c not in stopwords and len(c) >= 2]
        return entities

    def expand_query(self, query: str) -> List[str]:
        """查询扩展：生成多个变体以提高召回率"""
        expansions = [query]

        # 同义词扩展（示例）
        synonym_map = {
            '如何': '怎么',
            '哪些': '什么',
            '介绍': '说明',
            '区别': '差异',
        }

        expanded = query
        for word, synonym in synonym_map.items():
            expanded = expanded.replace(word, synonym)
        if expanded != query:
            expansions.append(expanded)

        return expansions

class RetrievalPipeline:
    """检索管道：整合多步检索流程"""

    def __init__(self, builder: KnowledgeBaseBuilder):
        self.builder = builder
        self.query_processor = QueryProcessor()
        self.cache: Dict[str, List[RetrievalResult]] = {}

    async def retrieve(self, 
                       query: str, 
                       mode: str = "hybrid",
                       top_k: int = 10,
                       history: List[str] = None) -> Tuple[str, List[RetrievalResult]]:
        """执行完整的检索流程"""

        # 1. 查询预处理
        processed_query = self.query_processor.rewrite_query(query, history)
        logger.info(f"查询改写: '{query}' -> '{processed_query}'")

        # 2. 检查缓存
        cache_key = hashlib.md5(processed_query.encode()).hexdigest()
        if cache_key in self.cache:
            logger.info("命中缓存")
            return processed_query, self.cache[cache_key]

        # 3. 执行LightRAG检索
        raw_answer = await self.builder.query(
            question=processed_query,
            mode=mode,
            top_k=top_k
        )

        # 4. 构建结果
        results = [RetrievalResult(
            content=raw_answer,
            source="lightrag",
            score=1.0,
            chunk_id=cache_key,
        )]

        # 5. 缓存结果
        self.cache[cache_key] = results

        return processed_query, results
​

步骤 5：问答接口与多轮对话管理

from typing import List, Dict, Optional
from dataclasses import dataclass, field
from collections import deque

@dataclass
class Message:
    """对话消息"""
    role: str          # user / assistant / system
    content: str
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

@dataclass
class Conversation:
    """对话会话"""
    session_id: str
    messages: deque = field(default_factory=deque)
    max_history: int = 10
    metadata: Dict = field(default_factory=dict)

    def add_message(self, role: str, content: str):
        self.messages.append(Message(role=role, content=content))
        # 维护消息窗口
        while len(self.messages) > self.max_history * 2:
            self.messages.popleft()

    def get_history(self) -> List[str]:
        return [msg.content for msg in self.messages]

    def get_context_for_llm(self) -> str:
        """构建LLM上下文"""
        context_parts = []
        for msg in self.messages:
            if msg.role == "user":
                context_parts.append(f"用户: {msg.content}")
            elif msg.role == "assistant":
                context_parts.append(f"助手: {msg.content}")

        return "\n".join(context_parts[-self.max_history * 2:])

    def clear(self):
        self.messages.clear()

class ConversationManager:
    """多轮对话管理器"""

    def __init__(self, max_sessions: int = 1000):
        self.sessions: Dict[str, Conversation] = {}
        self.max_sessions = max_sessions

    def get_or_create(self, session_id: str) -> Conversation:
        if session_id not in self.sessions:
            # 如果超过上限，清理最旧的会话
            if len(self.sessions) >= self.max_sessions:
                oldest = next(iter(self.sessions))
                del self.sessions[oldest]
                logger.info(f"清理过期会话: {oldest}")

            self.sessions[session_id] = Conversation(session_id=session_id)
            logger.info(f"创建新会话: {session_id}")

        return self.sessions[session_id]

    def get_session(self, session_id: str) -> Optional[Conversation]:
        return self.sessions.get(session_id)

class QASystem:
    """完整的智能问答系统"""

    def __init__(self, 
                 llm_model: str = "gpt-4o-mini",
                 storage_dir: str = "./storage/lightrag"):
        self.builder = KnowledgeBaseBuilder(
            working_dir=storage_dir,
            llm_model=llm_model
        )
        self.pipeline: Optional[RetrievalPipeline] = None
        self.conversation_mgr = ConversationManager()

    async def initialize(self, documents: List[Document] = None):
        """初始化系统"""
        await self.builder.init_rag()

        if documents:
            await self.builder.build_from_documents(documents)

        self.pipeline = RetrievalPipeline(self.builder)
        logger.info("问答系统初始化完成")

    async def chat(self, 
                   question: str,
                   session_id: str = "default",
                   mode: str = "hybrid") -> str:
        """处理用户提问，返回答案"""

        # 1. 获取或创建会话
        conversation = self.conversation_mgr.get_or_create(session_id)
        conversation.add_message("user", question)

        # 2. 检索相关内容
        history = conversation.get_history()
        processed_query, results = await self.pipeline.retrieve(
            query=question,
            mode=mode,
            history=history
        )

        # 3. 构建Prompt
        retrieval_context = ""
        for r in results:
            retrieval_context += f"\n{r.content}"

        dialogue_context = conversation.get_context_for_llm()

        prompt = f"""你是一个专业的知识问答助手。请根据检索到的知识内容回答用户的问题。

## 检索到的知识
{retrieval_context}

## 对话历史
{dialogue_context}

## 当前问题
{question}

## 回答要求
1. 基于检索到的知识进行回答，不要编造信息
2. 如果检索结果不足，请诚实说明
3. 回答要准确、简洁、有条理
4. 如果涉及专业概念，请给出清晰解释

请回答："""

        # 4. 调用LLM生成答案
        import openai
        response = await openai.chat.completions.create(
            model=self.builder.llm_model,
            messages=[
                {"role": "system", "content": "你是LightRAG智能问答系统的助手。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3,
            max_tokens=2048
        )

        answer = response.choices[0].message.content
        conversation.add_message("assistant", answer)

        return answer

    async def stream_chat(self,
                          question: str,
                          session_id: str = "default",
                          mode: str = "hybrid") -> AsyncIterator[str]:
        """流式回答"""
        conversation = self.conversation_mgr.get_or_create(session_id)
        conversation.add_message("user", question)

        processed_query, results = await self.pipeline.retrieve(
            query=question, mode=mode, history=conversation.get_history()
        )

        retrieval_context = "\n".join(r.content for r in results)
        dialogue_context = conversation.get_context_for_llm()

        prompt = f"""基于以下知识回答问题。检索知识：{retrieval_context}
对话历史：{dialogue_context}
问题：{question}
请回答："""

        import openai
        stream = await openai.chat.completions.create(
            model=self.builder.llm_model,
            messages=[
                {"role": "system", "content": "你是专业的知识问答助手。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3,
            max_tokens=2048,
            stream=True
        )

        full_answer = ""
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                full_answer += delta
                yield delta

        conversation.add_message("assistant", full_answer)
​

步骤 6：FastAPI服务接口

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import uvicorn

app = FastAPI(
    title="LightRAG 智能问答系统",
    description="基于LightRAG的图+向量双层检索智能问答API",
    version="1.0.0"
)

# 全局QA系统实例
qa_system: Optional[QASystem] = None

class QueryRequest(BaseModel):
    question: str
    session_id: str = "default"
    mode: str = "hybrid"
    stream: bool = False

class QueryResponse(BaseModel):
    answer: str
    session_id: str
    mode: str

class SessionResponse(BaseModel):
    session_id: str
    message_count: int

@app.on_event("startup")
async def startup_event():
    """应用启动时初始化QA系统"""
    global qa_system

    # 加载知识库文档
    preprocessor = DocumentPreprocessor(chunk_size=800, chunk_overlap=100)

    # 演示数据
    sample_docs = [
        Document(
            doc_id="demo1",
            title="LightRAG简介",
            content="LightRAG是HKUDS（香港大学数据科学研究所）开发的一个轻量级检索增强生成框架。"
                    "它采用图+向量双层检索架构，通过构建语义知识图谱和向量索引的有机结合，"
                    "实现了高效的知识检索和生成能力。LightRAG支持local、global、hybrid和mix四种检索模式，"
                    "能够适应不同类型的查询需求。",
            source="demo",
        ),
        Document(
            doc_id="demo2",
            title="双层检索架构",
            content="LightRAG的双层检索架构由图检索层和向量检索层组成。"
                    "图检索层基于知识图谱进行结构化推理，能够发现实体间的隐含关系。"
                    "向量检索层基于语义相似度进行快速匹配，能够定位相关内容片段。"
                    "两层通过融合策略结合，提供了比单一检索更丰富的上下文。",
            source="demo",
        ),
    ]

    for doc in sample_docs:
        doc = preprocessor.chunk_document(doc)

    qa_system = QASystem()
    await qa_system.initialize(sample_docs)
    logger.info("QA系统启动完成")

@app.post("/api/chat", response_model=QueryResponse)
async def chat(request: QueryRequest):
    """问答接口"""
    try:
        answer = await qa_system.chat(
            question=request.question,
            session_id=request.session_id,
            mode=request.mode
        )
        return QueryResponse(
            answer=answer,
            session_id=request.session_id,
            mode=request.mode
        )
    except Exception as e:
        logger.error(f"问答失败: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/api/chat/stream")
async def chat_stream(request: QueryRequest):
    """流式问答接口"""
    async def generate():
        async for chunk in qa_system.stream_chat(
            question=request.question,
            session_id=request.session_id,
            mode=request.mode
        ):
            yield f"data: {json.dumps({'content': chunk}, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

@app.delete("/api/session/{session_id}")
async def clear_session(session_id: str):
    """清除会话历史"""
    conv = qa_system.conversation_mgr.get_session(session_id)
    if conv:
        conv.clear()
    return {"status": "ok"}

@app.get("/health")
async def health_check():
    """健康检查"""
    return {"status": "healthy", "service": "LightRAG-QA"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)
​

步骤 7：完整运行示例

import asyncio

async def main():
    """完整运行示例"""

    # 1. 准备文档
    preprocessor = DocumentPreprocessor(chunk_size=500, chunk_overlap=50)

    # 示例知识文档
    sample_text = """
    # LightRAG技术架构

    LightRAG采用独特的双层检索架构，由图检索层和向量检索层组成。

    ## 图检索层
    图检索层使用LLM从文档中提取实体和关系，构建语义知识图谱。
    支持local模式（关注局部实体关系）和global模式（关注全局社区发现）。

    ## 向量检索层
    向量检索层使用embedding模型将文档编码为向量，通过相似度搜索定位相关内容。
    支持高效的近似最近邻（ANN）搜索算法。

    ## 融合策略
    hybrid模式同时使用两种检索方式，通过加权融合获得最佳结果。
    mix模式则在local和global之间动态切换。

    ## 典型应用场景
    - 技术文档问答
    - 企业知识库检索
    - 科研论文分析
    - 法律文书查询
    """

    doc = Document(
        doc_id="lightrag_guide",
        title="LightRAG技术架构",
        content=sample_text,
        source="demo"
    )
    doc = preprocessor.chunk_document(doc)

    # 2. 初始化系统
    qa = QASystem(llm_model="gpt-4o-mini", storage_dir="./demo_storage")
    await qa.initialize([doc])

    # 3. 单轮问答
    print("=" * 60)
    print("单轮问答示例")
    print("=" * 60)

    questions = [
        "LightRAG的双层检索架构包含哪两层？",
        "什么是hybrid检索模式？",
        "LightRAG有哪些应用场景？",
    ]

    for q in questions:
        answer = await qa.chat(q, session_id="demo_single")
        print(f"\n❓ {q}")
        print(f"💬 {answer[:200]}...")

    # 4. 多轮对话
    print("\n" + "=" * 60)
    print("多轮对话示例")
    print("=" * 60)

    multi_turn_questions = [
        "LightRAG的检索层有哪些模式？",
        "local模式和global模式有什么区别？",  # 上下文追问
        "在什么场景下应该使用local模式？",   # 继续追问
        "能详细解释一下图检索层的工作原理吗？",  # 切换话题但保持上下文
    ]

    for q in multi_turn_questions:
        answer = await qa.chat(q, session_id="demo_multi")
        print(f"\n❓ {q}")
        print(f"💬 {answer[:150]}...")

    # 5. 清理会话
    conv = qa.conversation_mgr.get_session("demo_multi")
    print(f"\n会话消息数: {len(conv.messages)}")

if __name__ == "__main__":
    asyncio.run(main())
​

常见问题 FAQ

Q1：LightRAG的检索模式应该如何选择？

A：LightRAG提供四种检索模式，选择建议如下：

Q2：如何提升问答系统的回答质量？

A：提升回答质量的关键策略：

1. 数据质量：确保知识库文档准确、完整、结构化
2. 文档切分：合理的chunk_size和overlap提升检索相关性
3. Prompt工程：精心设计系统提示词，明确回答规范
4. 检索调优：调整top_k和mode参数适应具体场景
5. 上下文管理：有效的多轮对话上下文传递

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

A：上下文窗口大小的选择取决于：

• LLM上下文限制：通常不超过总token的60%留给历史
• 查询相关度：保留最近5-10轮通常足够
• 内存开销：每条历史消息都占用token，影响成本和延迟

建议从 max_history=5 开始，根据效果调整。

Q4：生产环境如何处理高并发？

A：高并发优化的核心措施：

• 异步架构：使用FastAPI + async/await
• 连接池：复用LLM API连接和数据库连接
• 缓存：对高频查询启用结果缓存
• 限流：实现请求限流防止过载
• 队列：使用Redis/MQ削峰填谷

最佳实践与避坑

实践 1：文档切分参数调优

chunk_size的选择需要平衡检索精度和上下文完整性：

• 短chunk（300-500字）：检索精度高，但可能丢失上下文
• 中等chunk（500-1000字）：通用推荐值
• 长chunk（1000-2000字）：上下文丰富，但检索噪声大

坑点 1：忽略文档元数据

仅靠内容检索可能无法区分来源不同的同名概念。建议在检索结果中保留文档元数据（来源、时间、作者），帮助LLM判断可信度。

实践 2：分层缓存策略

实现"查询缓存→向量缓存→图缓存"三级缓存，能将平均响应时间降低60%以上。

坑点 2：多轮对话中的"遗忘"问题

简单的滑动窗口会丢失远距离的重要上下文。对关键信息可以实施"摘要压缩"——将早期对话压缩为摘要附加到上下文中。

本节小结

本节从零搭建了一个基于LightRAG的完整智能问答系统，涵盖了以下核心内容：

1. 文档预处理：清洗、切分、批量加载的工程化方法
2. 知识库构建：基于LightRAG的图+向量双层索引建立
3. 检索管道：查询改写、多路检索、结果融合的完整流程
4. 问答接口：同步/流式API设计与实现
5. 多轮对话：会话管理、上下文窗口、历史维护

下一节将深入探讨企业级知识库的构建实战，涵盖多数据源整合和增量更新策略。

延伸阅读

• LightRAG GitHub仓库
• FastAPI官方文档
• RAG评估框架RAGAS

关键词：LightRAG, 智能问答, 多轮对话, 检索管道, FastAPI, 上下文管理
难度：实战
预计阅读：50 分钟