第5章：模块1总结与项目¶

恭喜你完成了模块1的学习！现在让我们通过一个完整的实战项目，综合运用所学知识，构建一个真正的智能文档助手。

📚 学习目标¶

学完本章后，你将能够：

回顾和巩固模块1的核心概念
综合运用所学知识构建完整项目
实现一个可用的智能文档问答系统
建立项目评估和优化流程
为模块2的学习做好准备

预计学习时间：1.5小时 难度等级：⭐⭐☆☆☆

5.1 知识回顾¶

核心概念复习¶

让我们快速回顾模块1的核心知识点：

1. RAG是什么？¶

定义：检索增强生成，结合信息检索和LLM生成

核心价值： - ✅ 减少幻觉 - ✅ 知识实时更新 - ✅ 可解释性强 - ✅ 成本相对较低

工作流程：

用户问题 → 检索相关文档 → LLM生成答案 → 返回结果

2. RAG的5大组件¶

组件	作用	关键技术
文档加载器	加载各种格式文档	SimpleDirectoryReader
文本分块器	切分长文档	SentenceSplitter
嵌入模型	文本向量化	OpenAI embeddings, BGE
向量数据库	高效检索	Chroma, Qdrant, Milvus
LLM	生成答案	GPT-3.5, GPT-4, Qwen

3. 关键技术决策¶

分块策略： - chunk_size: 500-1000（通用） - chunk_overlap: 10-20% of chunk_size - 优先按段落分块

嵌入模型选择： - 快速/便宜: OpenAI text-embedding-3-small - 高质量: OpenAI text-embedding-3-large - 私有部署: BGE-small-zh

向量数据库选择： - 学习/原型: Chroma - 生产环境: Qdrant, Milvus

4. 评估指标¶

检索质量： - Hit Rate: 命中率（>0.85优秀） - MRR: 平均倒数排名（>0.7优秀） - Precision@K: 前K个结果的精确率

生成质量： - Faithfulness: 忠实度（>0.8优秀） - Relevancy: 相关性（>0.8优秀）

常见问题解答¶

Q1: chunk_size如何选择？¶

原则：根据内容类型和查询特点

短文档/精准查询：
  chunk_size: 300-500
  → 信息密度高，检索精准

长文档/广泛查询：
  chunk_size: 1000-1500
  → 保持上下文完整

代码文档：
  chunk_size: 500-800
  → 保持代码完整性

Q2: 检索效果不好怎么办？¶

排查步骤：

检查数据质量

# 查看分块效果
for i, node in enumerate(nodes[:5]):
    print(f"块{i+1}: {node.text[:100]}...")

调整检索参数

# 增加检索数量
response = index.query(question, similarity_top_k=5)

优化查询

# 查询重写
rewritten_query = rewrite_query(user_query)

尝试不同的嵌入模型

# 从OpenAI切换到BGE
embed_model = OpenAIEmbedding(model="text-embedding-3-large")

Q3: 如何降低API成本？¶

优化策略：

缓存常见查询

import functools

@functools.lru_cache(maxsize=100)
def cached_query(question):
    return rag.query(question)

使用更便宜的模型

# GPT-3.5代替GPT-4
llm = OpenAILLM(model="gpt-3.5-turbo")

批量处理

# 批量生成嵌入
embeddings = get_embeddings(texts)  # 一次调用

使用开源模型

# 本地部署BGE + Qwen
embed_model = BGEEmbedding()
llm = QwenLLM()

5.2 实战项目：智能文档助手¶

项目概述¶

项目名称：InteliKB-Lite（智能知识库轻量版）

项目目标：构建一个能够回答技术文档问题的智能助手

项目需求¶

功能需求¶

性能要求¶

响应时间：< 3秒
Hit Rate：> 0.6
用户满意度：> 70%

项目架构¶

InteliKB-Lite架构

┌─────────────────────────────────────┐
│         Web界面 (Streamlit)          │
├─────────────────────────────────────┤
│                                     │
│  ┌─────────────┐  ┌──────────────┐ │
│  │ 文档上传    │  │  问答界面    │ │
│  └──────┬──────┘  └──────┬───────┘ │
│         │                │          │
│         └────────┬───────┘          │
│                  │                  │
├──────────────────┼──────────────────┤
│         RAG引擎 (LlamaIndex)        │
│  ┌────────────────────────────────┐ │
│  │ 1. 文档处理                   │ │
│  │ 2. 分块                       │ │
│  │ 3. 向量化                     │ │
│  │ 4. 检索                       │ │
│  │ 5. 生成                       │ │
│  └────────────────────────────────┘ │
├─────────────────────────────────────┤
│         数据层                       │
│  ┌──────────┐  ┌──────────────────┐ │
│  │ Chroma   │  │  文档存储        │ │
│  └──────────┘  └──────────────────┘ │
└─────────────────────────────────────┘

完整实现¶

步骤1：项目结构¶

# 创建项目目录
mkdir intelikb_lite
cd intelikb_lite

# 创建目录结构
mkdir -p data/{raw,processed,eval}
mkdir -p notebooks
mkdir -p src
mkdir -p outputs

# 创建文件
touch README.md
touch requirements.txt
touch src/__init__.py
touch src/rag_engine.py
touch src/evaluator.py
touch app.py  # Streamlit应用

步骤2：RAG引擎实现¶

# 文件名：src/rag_engine.py
"""
InteliKB-Lite RAG引擎
"""

import os
from dotenv import load_dotenv
from pathlib import Path
from typing import List, Dict, Tuple

from llama_index.core import SimpleDirectoryReader, Document, VectorStoreIndex, StorageContext
from llama_index.core.node_parser import SentenceSplitter
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI as OpenAILLM
import chromadb

load_dotenv()

class RAGEngine:
    """RAG引擎"""

    def __init__(self, persist_dir="./chroma_db"):
        """
        初始化RAG引擎

        Args:
            persist_dir: 向量库持久化目录
        """
        self.persist_dir = persist_dir
        self.index = None
        self.query_engine = None

    def load_documents(self, data_path: str) -> List[Document]:
        """
        加载文档

        Args:
            data_path: 文档路径

        Returns:
            文档列表
        """
        reader = SimpleDirectoryReader(data_path)
        documents = reader.load_data()
        return documents

    def build_index(self, documents: List[Document]):
        """
        构建向量索引

        Args:
            documents: 文档列表
        """
        # 分块
        splitter = SentenceSplitter(
            chunk_size=800,
            chunk_overlap=100
        )
        nodes = splitter.get_nodes_from_documents(documents)

        # 创建向量库
        chroma_client = chromadb.PersistentClient(path=self.persist_dir)
        collection = chroma_client.get_or_create_collection("intelikb")
        vector_store = ChromaVectorStore(chroma_collection=collection)

        storage_context = StorageContext.from_defaults(vector_store=vector_store)

        # 构建索引
        self.index = VectorStoreIndex(
            nodes=nodes,
            storage_context=storage_context,
            embed_model=OpenAIEmbedding(model="text-embedding-3-small")
        )

        # 创建查询引擎
        self.query_engine = self.index.as_query_engine(
            llm=OpenAILLM(model="gpt-3.5-turbo", temperature=0.7),
            similarity_top_k=3,
            vector_store_query_mode="default"
        )

    def query(self, question: str) -> Tuple[str, List[Dict]]:
        """
        查询

        Args:
            question: 用户问题

        Returns:
            (答案, 来源文档列表)
        """
        if self.query_engine is None:
            raise ValueError("引擎未初始化，请先调用build_index()")

        response = self.query_engine.query(question)

        # 提取来源
        sources = []
        if hasattr(response, 'source_nodes'):
            for node in response.source_nodes:
                sources.append({
                    "content": node.node.text[:200] + "...",
                    "metadata": node.node.metadata,
                    "score": node.score if hasattr(node, 'score') else 0.0
                })

        return str(response), sources

    def add_documents(self, new_documents: List[Document]):
        """
        添加新文档

        Args:
            new_documents: 新文档列表
        """
        if self.index is None:
            self.build_index(new_documents)
        else:
            for doc in new_documents:
                self.index.insert(doc)

    def save_index(self):
        """保存索引"""
        if self.index:
            self.index.storage_context.persist()

    @classmethod
    def load_index(cls, persist_dir: str = "./chroma_db"):
        """
        加载已保存的索引

        Args:
            persist_dir: 持久化目录

        Returns:
            RAG引擎实例
        """
        engine = cls(persist_dir)

        # 加载向量库
        chroma_client = chromadb.PersistentClient(path=persist_dir)
        collection = chroma_client.get_collection("intelikb")
        vector_store = ChromaVectorStore(chroma_collection=collection)

        storage_context = StorageContext.from_defaults(vector_store=vector_store)

        # 加载索引
        engine.index = VectorStoreIndex.from_documents(
            [],
            storage_context=storage_context,
            embed_model=OpenAIEmbedding(model="text-embedding-3-small")
        )

        # 创建查询引擎
        engine.query_engine = engine.index.as_query_engine(
            llm=OpenAILLM(model="gpt-3.5-turbo", temperature=0.7),
            similarity_top_k=3
        )

        return engine

步骤3：评估器实现¶

# 文件名：src/evaluator.py
"""
评估器
"""

import json
from typing import List, Dict
from openai import OpenAI

class RAGEvaluator:
    """RAG评估器"""

    def __init__(self):
        self.client = OpenAI()

    def evaluate_faithfulness(self, answer: str, contexts: List[str]) -> float:
        """
        评估忠实度

        Args:
            answer: 答案
            contexts: 上下文列表

        Returns:
            忠实度分数
        """
        context_text = "\n\n".join([f"文档{i+1}: {c}" for i, c in enumerate(contexts)])

        prompt = f"""
评估答案是否基于参考文档。

参考文档：
{context_text}

答案：
{answer}

评分（0-1）：
"""

        try:
            response = self.client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}],
                temperature=0
            )
            score = float(response.choices[0].message.content.strip())
            return max(0, min(1, score))
        except:
            return 0.5

    def evaluate_relevancy(self, question: str, answer: str) -> float:
        """
        评估相关性

        Args:
            question: 问题
            answer: 答案

        Returns:
            相关性分数
        """
        prompt = f"""
评估答案是否回答了问题。

问题：{question}
答案：{answer}

评分（0-1）：
"""

        try:
            response = self.client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}],
                temperature=0
            )
            score = float(response.choices[0].message.content.strip())
            return max(0, min(1, score))
        except:
            return 0.5

    def load_golden_dataset(self, path: str) -> List[Dict]:
        """加载黄金数据集"""
        with open(path, 'r', encoding='utf-8') as f:
            return json.load(f)

    def run_evaluation(self, rag_engine, golden_dataset_path: str) -> Dict:
        """
        运行评估

        Args:
            rag_engine: RAG引擎
            golden_dataset_path: 黄金数据集路径

        Returns:
            评估结果
        """
        dataset = self.load_golden_dataset(golden_dataset_path)

        faithfulness_scores = []
        relevancy_scores = []

        for item in dataset:
            question = item["question"]
            ground_truth = item["answer"]

            # 查询
            answer, sources = rag_engine.query(question)
            contexts = [s["content"] for s in sources]

            # 评估
            faith = self.evaluate_faithfulness(answer, contexts)
            rel = self.evaluate_relevancy(question, answer)

            faithfulness_scores.append(faith)
            relevancy_scores.append(rel)

        # 计算平均分
        results = {
            "faithfulness": sum(faithfulness_scores) / len(faithfulness_scores),
            "relevancy": sum(relevancy_scores) / len(relevancy_scores),
            "num_evaluated": len(dataset)
        }

        return results

步骤4：Streamlit界面¶

# 文件名：app.py
"""
InteliKB-Lite Streamlit应用
"""

import streamlit as st
import os
from pathlib import Path

from src.rag_engine import RAGEngine
from src.evaluator import RAGEvaluator

# 页面配置
st.set_page_config(
    page_title="InteliKB-Lite",
    page_icon="🤖",
    layout="wide"
)

# 自定义CSS
st.markdown("""
<style>
    .main-header {
        font-size: 3rem;
        font-weight: bold;
        text-align: center;
        color: #1f77b4;
    }
    .sub-header {
        font-size: 1.5rem;
        color: #666;
    }
</style>
""", unsafe_allow_html=True)

# 初始化session state
if 'engine' not in st.session_state:
    st.session_state.engine = None
if 'documents_loaded' not in st.session_state:
    st.session_state.documents_loaded = False

def main():
    """主函数"""

    # 标题
    st.markdown('<p class="main-header">🤖 InteliKB-Lite</p>', unsafe_allow_html=True)
    st.markdown('<p class="sub-header">智能文档问答系统</p>', unsafe_allow_html=True)

    st.sidebar.title("系统设置")

    # 侧边栏：API密钥
    api_key = st.sidebar.text_input("OpenAI API Key", type="password")
    if api_key:
        os.environ["OPENAI_API_KEY"] = api_key

    # 侧边栏：功能选择
    page = st.sidebar.radio("选择功能", ["文档管理", "智能问答", "系统评估"])

    # 页面路由
    if page == "文档管理":
        document_management_page()
    elif page == "智能问答":
        qa_page()
    elif page == "系统评估":
        evaluation_page()

def document_management_page():
    """文档管理页面"""
    st.header("📄 文档管理")

    # 上传文档
    uploaded_files = st.file_uploader(
        "上传文档",
        type=["txt", "md", "pdf"],
        accept_multiple_files=True,
        help="支持TXT、Markdown、PDF格式"
    )

    if uploaded_files:
        # 保存文档
        data_dir = Path("data/processed")
        data_dir.mkdir(parents=True, exist_ok=True)

        for file in uploaded_files:
            file_path = data_dir / file.name
            with open(file_path, "wb") as f:
                f.write(file.getbuffer())

        st.success(f"✓ 成功上传 {len(uploaded_files)} 个文件")

        # 构建索引
        if st.button("构建索引", type="primary"):
            with st.spinner("正在处理文档..."):
                engine = RAGEngine()
                documents = engine.load_documents("data/processed")
                engine.build_index(documents)
                engine.save_index()

                st.session_state.engine = engine
                st.session_state.documents_loaded = True

            st.success(f"✓ 成功处理 {len(documents)} 个文档")
            st.info(f"✓ 索引已保存到 ./chroma_db")

def qa_page():
    """问答页面"""
    st.header("💬 智能问答")

    # 检查是否已加载文档
    if not st.session_state.documents_loaded:
        st.warning("⚠️ 请先在'文档管理'页面上传并处理文档")
        return

    # 加载引擎
    if st.session_state.engine is None:
        try:
            st.session_state.engine = RAGEngine.load_index()
        except:
            st.error("❌ 无法加载索引，请重新构建")
            return

    engine = st.session_state.engine

    # 问答界面
    question = st.text_area(
        "输入你的问题",
        placeholder="例如：Python有什么特点？",
        height=100
    )

    if st.button("提问", type="primary"):
        if not question:
            st.warning("请输入问题")
            return

        with st.spinner("正在思考..."):
            try:
                answer, sources = engine.query(question)

                # 显示答案
                st.subheader("📝 答案")
                st.write(answer)

                # 显示来源
                st.subheader("📚 参考来源")
                for i, source in enumerate(sources, 1):
                    with st.expander(f"来源 {i}"):
                        st.write(source["content"])
                        st.caption(f"相关度: {source['score']:.3f}")

            except Exception as e:
                st.error(f"❌ 出错: {str(e)}")

def evaluation_page():
    """评估页面"""
    st.header("📊 系统评估")

    # 上传黄金数据集
    golden_file = st.file_uploader(
        "上传黄金数据集",
        type=["json"],
        help="JSON格式的问答对"
    )

    if golden_file:
        # 保存数据集
        data_dir = Path("data/eval")
        data_dir.mkdir(parents=True, exist_ok=True)

        dataset_path = data_dir / "golden_dataset.json"
        with open(dataset_path, "wb") as f:
            f.write(golden_file.getbuffer())

        st.success("✓ 数据集已上传")

        # 运行评估
        if st.button("开始评估", type="primary"):
            if not st.session_state.documents_loaded:
                st.warning("⚠️ 请先处理文档")
                return

            with st.spinner("正在评估..."):
                try:
                    evaluator = RAGEvaluator()
                    results = evaluator.run_evaluation(
                        st.session_state.engine,
                        str(dataset_path)
                    )

                    # 显示结果
                    st.subheader("评估结果")

                    col1, col2 = st.columns(2)

                    with col1:
                        st.metric(
                            "忠实度",
                            f"{results['faithfulness']:.2%}",
                            help="答案基于文档的程度"
                        )

                    with col2:
                        st.metric(
                            "相关性",
                            f"{results['relevancy']:.2%}",
                            help="答案与问题的相关程度"
                        )

                    st.info(f"✓ 评估了 {results['num_evaluated']} 个问答对")

                    # 评级
                    avg_score = (results['faithfulness'] + results['relevancy']) / 2

                    if avg_score > 0.8:
                        st.success("🏆 优秀！系统表现良好")
                    elif avg_score > 0.6:
                        st.info("👍 良好，仍有优化空间")
                    else:
                        st.warning("⚠️ 需要改进")

                except Exception as e:
                    st.error(f"❌ 评估失败: {str(e)}")

if __name__ == "__main__":
    main()

步骤5：依赖文件¶

# 文件名：requirements.txt
# InteliKB-Lite依赖

# 核心框架
llama-index-core>=0.10.0
llama-index-llms-openai>=0.1.0
llama-index-embeddings-openai>=0.1.0
llama-index-vector-stores-chroma>=0.1.0

# 向量数据库
chromadb>=0.4.0

# 文档处理
pypdf>=3.0.0
python-dotenv>=1.0.0

# Web界面
streamlit>=1.28.0

# 评估
openai>=1.0.0

项目运行指南¶

1. 安装依赖¶

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装依赖
pip install -r requirements.txt

2. 准备数据¶

# 将文档放入data/processed目录
# 支持格式：.txt, .md, .pdf

3. 配置API密钥¶

# 创建.env文件
echo "OPENAI_API_KEY=your_api_key_here" > .env

4. 启动应用¶

# 启动Streamlit
streamlit run app.py

5. 使用指南¶

步骤1：文档管理 - 上传你的文档（TXT、MD、PDF） - 点击"构建索引" - 等待处理完成

步骤2：智能问答 - 输入问题 - 查看答案和参考来源 - 评估答案质量

步骤3：系统评估 - 上传黄金数据集（JSON格式） - 运行评估 - 查看评估结果

项目交付清单¶

必须完成¶

加分项¶

文档要求¶

README.md（项目说明）
requirements.txt（依赖清单）
使用说明（用户指南）
评估报告

评分标准¶

项目	占比	评分标准
功能完整性	40%	实现所有必须功能
代码质量	30%	代码规范、注释完整、结构清晰
用户体验	20%	界面友好、交互流畅
文档说明	10%	README完整、使用说明清晰

总结¶

模块1学习成果¶

完成模块1后，你已经：

掌握了RAG基础
理解RAG的核心概念和价值
了解5大核心组件的作用
知道如何选择合适的技术栈
实现了基础RAG系统
能够加载和处理文档
实现了向量检索
生成了基于文档的答案
建立了评估体系
知道如何评估RAG系统
掌握了关键评估指标
能够基于评估优化系统
完成了实战项目
构建了完整的文档问答系统
综合运用了所学知识
获得了实际项目经验

下一步学习方向¶

模块2预告：核心优化技术

✨ 语义分块
✨ 查询增强（HyDE）
✨ 混合检索
✨ 重排序
✨ 性能优化

预计提升： - Hit Rate: 0.6 → 0.75 (+25%) - MRR: 0.5 → 0.65 (+30%) - 用户满意度: 70% → 85% (+21%)

扩展练习¶

进阶挑战¶

优化检索质量
实现查询重写
尝试不同的分块策略
使用更强的嵌入模型
改进用户体验
添加查询历史
实现相似问题推荐
支持文档高亮显示
性能优化
实现答案缓存
优化索引结构
减少API调用

创新项目¶

多语言RAG
支持中英文混合
跨语言检索
实时更新
文档增量索引
热更新机制
多模态支持
图像问答
表格理解