你是否厌倦了每次提问都要联网？是否担心敏感数据上传到云端？本文将带你从零开始，用 Ollama 和 RAG 技术栈，在本地搭建一个完全离线、数据安全、可定制的 AI 知识库与对话助手。

---

前言

过去一年，ChatGPT、Claude 等云端大模型席卷全球，但随之而来的问题也愈发尖锐：数据隐私、网络依赖、使用成本、定制能力。许多企业和个人开发者都在问同一个问题——我能不能在本地跑一个完全属于自己的 AI 助手，既能私有化部署，又能结合我自己的知识库？

答案是：完全可以。而且比你想象的简单。

今天，我将带你用 Ollama + RAG（检索增强生成）技术栈，从零搭建一个离线 AI 知识库系统。这套方案已经在多个企业场景中落地——客服知识库、技术文档问答、合同审查辅助等。全程不需要 GPU，普通笔记本就能跑。

读完这篇文章，你将获得：

• 一个完全可运行的本地 AI 知识库助手
• 对 RAG 核心原理的深度理解
• 可直接复用的完整代码
• 性能优化的实战经验

---

一、技术栈全景图

在动手之前，先理解整个系统的架构：

┌──────────────────────────────────────────────────┐
│                   用户提问                         │
└──────────────────┬───────────────────────────────┘
                   ▼
┌──────────────────────────────────────────────────┐
│              1. 查询处理层                         │
│         (Query Rewriting + Intent Analysis)       │
└──────────────────┬───────────────────────────────┘
                   ▼
┌──────────────────────────────────────────────────┐
│              2. 向量检索层                         │
│     Embedding Model → Vector DB → Top-K 召回       │
└──────────────────┬───────────────────────────────┘
                   ▼
┌──────────────────────────────────────────────────┐
│              3. 重排序层（可选）                     │
│           Cross-Encoder → 精排                     │
└──────────────────┬───────────────────────────────┘
                   ▼
┌──────────────────────────────────────────────────┐
│              4. 生成层                             │
│        Prompt 组装 → Ollama LLM → 流式输出          │
└──────────────────────────────────────────────────┘

核心技术选型：

组件	选型	理由
本地 LLM	Ollama + Qwen2.5 (7B/14B)	中文能力强，Ollama 部署零门槛
Embedding	BGE-M3 / nomic-embed-text	多语言支持，本地运行
向量数据库	ChromaDB	轻量级，Python 原生，零配置
文档处理	LangChain + Unstructured	支持 PDF/Word/Markdown 等多种格式
界面	Streamlit	快速构建 Web UI

---

二、环境搭建

2.1 安装 Ollama

Ollama 是目前最流行的本地 LLM 运行工具，支持 macOS、Linux 和 Windows。

# macOS
brew install ollama

# Linux
curl -fsSL https://ollama.com/install.sh | sh

# Windows: 从 https://ollama.com 下载安装包

安装完成后，拉取模型：

# 拉取 Qwen2.5 7B 模型（推荐，平衡效果与速度）
ollama pull qwen2.5:7b

# 拉取 Embedding 模型
ollama pull nomic-embed-text

# 验证安装
ollama run qwen2.5:7b "你好，请用一句话介绍你自己"

为什么选择 Qwen2.5？

• 开源协议友好（Apache 2.0）
• 中文能力在开源模型中名列前茅
• 7B 版本在 16GB 内存的笔记本上流畅运行
• 支持 128K 上下文，处理长文档无压力

2.2 Python 环境配置

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

# 安装核心依赖
pip install langchain langchain-community langchain-chroma
pip install chromadb
pip install streamlit
pip install unstructured[all-docs]  # 文档解析
pip install pypdf  # PDF 支持
pip install python-docx  # Word 支持
pip install sentence-transformers  # 可选的本地 Embedding

完整的 requirements.txt：

langchain>=0.3.0
langchain-community>=0.3.0
langchain-chroma>=0.1.0
chromadb>=0.5.0
streamlit>=1.35.0
unstructured>=0.15.0
pypdf>=4.0.0
python-docx>=1.1.0
sentence-transformers>=3.0.0
tiktoken>=0.7.0

---

三、RAG 系统核心实现

3.1 文档加载与分割

这是整个系统的数据入口。实际场景中，你的知识库可能包含 PDF、Word、Markdown、网页等各种格式。

"""
document_loader.py - 文档加载与智能分割模块
"""
import os
from typing import List, Optional
from langchain_community.document_loaders import (
    PyPDFLoader,
    Docx2txtLoader,
    TextLoader,
    UnstructuredMarkdownLoader,
    CSVLoader,
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document


class SmartDocumentLoader:
    """智能文档加载器，支持多种格式，自动分段"""
    
    def __init__(
        self,
        chunk_size: int = 500,
        chunk_overlap: int = 50,
        separators: Optional[List[str]] = None,
    ):
        """
        初始化文档分割器
        
        Args:
            chunk_size: 每个文本块的最大长度
            chunk_overlap: 块之间的重叠长度
            separators: 分隔符优先级列表
        """
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        
        # 中文友好的分隔符
        if separators is None:
            separators = [
                "\n\n",    # 段落分隔
                "\n",      # 换行
                "。",      # 中文句号
                "！",      # 中文感叹号
                "？",      # 中文问号
                "；",      # 中文分号
                ".",       # 英文句号
                "!",
                "?",
                ";",
                " ",       # 空格
                "",        # 字符级分割
            ]
        
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=separators,
            length_function=len,
            is_separator_regex=False,
        )
    
    def load_file(self, file_path: str) -> List[Document]:
        """加载单个文件"""
        ext = os.path.splitext(file_path)[1].lower()
        
        loader_map = {
            '.pdf': PyPDFLoader,
            '.docx': Docx2txtLoader,
            '.txt': TextLoader,
            '.md': UnstructuredMarkdownLoader,
            '.csv': CSVLoader,
        }
        
        loader_class = loader_map.get(ext)
        if loader_class is None:
            raise ValueError(f"不支持的文件格式: {ext}")
        
        loader = loader_class(file_path)
        documents = loader.load()
        
        # 为每个文档添加来源元数据
        for doc in documents:
            doc.metadata["source"] = file_path
            doc.metadata["filename"] = os.path.basename(file_path)
        
        return documents
    
    def load_directory(self, dir_path: str, recursive: bool = True) -> List[Document]:
        """加载目录下所有支持的文件"""
        all_documents = []
        
        for root, dirs, files in os.walk(dir_path):
            for file in files:
                file_path = os.path.join(root, file)
                ext = os.path.splitext(file)[1].lower()
                
                if ext in ['.pdf', '.docx', '.txt', '.md', '.csv']:
                    try:
                        documents = self.load_file(file_path)
                        all_documents.extend(documents)
                        print(f"✓ 已加载: {file_path} ({len(documents)} 页/段)")
                    except Exception as e:
                        print(f"✗ 加载失败: {file_path}, 错误: {e}")
            
            if not recursive:
                break
        
        return all_documents
    
    def split_documents(self, documents: List[Document]) -> List[Document]:
        """将文档分割为适合检索的小块"""
        chunks = self.text_splitter.split_documents(documents)
        print(f"文档分割完成: {len(documents)} 个文档 → {len(chunks)} 个文本块")
        return chunks
    
    def load_and_split_file(self, file_path: str) -> List[Document]:
        """加载并分割单个文件（一步到位）"""
        documents = self.load_file(file_path)
        return self.split_documents(documents)
    
    def load_and_split_directory(self, dir_path: str, recursive: bool = True) -> List[Document]:
        """加载并分割目录（一步到位）"""
        documents = self.load_directory(dir_path, recursive)
        return self.split_documents(documents)


# 使用示例
if __name__ == "__main__":
    loader = SmartDocumentLoader(chunk_size=500, chunk_overlap=50)
    
    # 加载单个文件
    chunks = loader.load_and_split_file("./docs/技术文档.pdf")
    
    # 或加载整个目录
    # chunks = loader.load_and_split_directory("./knowledge_base/")
    
    for i, chunk in enumerate(chunks[:3]):
        print(f"\n--- 文本块 {i+1} ---")
        print(f"来源: {chunk.metadata.get('source', 'unknown')}")
        print(f"内容: {chunk.page_content[:200]}...")

3.2 向量存储与检索

这是 RAG 的核心——如何将文档转为向量，并进行高效检索。

"""
vector_store.py - 向量存储与检索模块
"""
from typing import List, Optional, Dict, Any
from langchain_chroma import Chroma
from langchain_community.embeddings import OllamaEmbeddings
from langchain.schema import Document
import chromadb
from chromadb.config import Settings as ChromaSettings


class VectorStoreManager:
    """向量存储管理器，封装 ChromaDB 操作"""
    
    def __init__(
        self,
        embedding_model: str = "nomic-embed-text",
        persist_directory: str = "./chroma_db",
        collection_name: str = "knowledge_base",
        ollama_base_url: str = "http://localhost:11434",
    ):
        """
        初始化向量存储
        
        Args:
            embedding_model: Ollama 中的 Embedding 模型名称
            persist_directory: 向量数据库持久化路径
            collection_name: 集合名称
            ollama_base_url: Ollama 服务地址
        """
        self.embedding_model = embedding_model
        self.persist_directory = persist_directory
        self.collection_name = collection_name
        
        # 初始化 Ollama Embedding
        self.embeddings = OllamaEmbeddings(
            model=embedding_model,
            base_url=ollama_base_url,
        )
        
        # 初始化 ChromaDB 客户端（持久化模式）
        self.chroma_client = chromadb.PersistentClient(
            path=persist_directory,
            settings=ChromaSettings(
                anonymized_telemetry=False,
                allow_reset=True,
            )
        )
        
        # LangChain Chroma 包装器
        self.vector_store: Optional[Chroma] = None
    
    def create_from_documents(
        self,
        documents: List[Document],
        batch_size: int = 100,
    ) -> Chroma:
        """
        从文档创建向量存储
        
        Args:
            documents: 文档块列表
            batch_size: 批量处理的文档数量
        
        Returns:
            Chroma 向量存储实例
        """
        print(f"正在为 {len(documents)} 个文档块生成向量...")
        
        # 批量处理以避免内存溢出
        self.vector_store = Chroma.from_documents(
            documents=documents,
            embedding=self.embeddings,
            persist_directory=self.persist_directory,
            collection_name=self.collection_name,
            client=self.chroma_client,
        )
        
        print(f"✓ 向量存储已创建并持久化到: {self.persist_directory}")
        return self.vector_store
    
    def load_existing(self) -> Optional[Chroma]:
        """加载已存在的向量存储"""
        try:
            self.vector_store = Chroma(
                collection_name=self.collection_name,
                embedding_function=self.embeddings,
                persist_directory=self.persist_directory,
                client=self.chroma_client,
            )
            
            # 验证是否有数据
            count = self.vector_store._collection.count()
            print(f"✓ 已加载向量存储，包含 {count} 个向量")
            return self.vector_store
        except Exception as e:
            print(f"未找到已存在的向量存储: {e}")
            return None
    
    def similarity_search(
        self,
        query: str,
        k: int = 4,
        filter: Optional[Dict[str, Any]] = None,
    ) -> List[Document]:
        """
        相似度检索
        
        Args:
            query: 查询文本
            k: 返回文档数量
            filter: 元数据过滤条件
        
        Returns:
            相关文档列表
        """
        if self.vector_store is None:
            raise ValueError("请先创建或加载向量存储")
        
        return self.vector_store.similarity_search(
            query=query,
            k=k,
            filter=filter,
        )
    
    def similarity_search_with_score(
        self,
        query: str,
        k: int = 4,
    ) -> List[tuple]:
        """
        带相似度分数的检索
        
        Returns:
            [(Document, score), ...] 分数越低表示越相似
        """
        if self.vector_store is None:
            raise ValueError("请先创建或加载向量存储")
        
        return self.vector_store.similarity_search_with_relevance_scores(
            query=query,
            k=k,
        )
    
    def max_marginal_relevance_search(
        self,
        query: str,
        k: int = 4,
        fetch_k: int = 20,
        lambda_mult: float = 0.5,
    ) -> List[Document]:
        """
        MMR 检索，平衡相关性与多样性
        
        Args:
            query: 查询文本
            k: 最终返回的文档数
            fetch_k: 初始候选文档数
            lambda_mult: 多样性系数 (0=最大多样性, 1=最大相关性)
        """
        if self.vector_store is None:
            raise ValueError("请先创建或加载向量存储")
        
        return self.vector_store.max_marginal_relevance_search(
            query=query,
            k=k,
            fetch_k=fetch_k,
            lambda_mult=lambda_mult,
        )
    
    def delete_collection(self):
        """删除集合"""
        if self.vector_store:
            self.vector_store.delete_collection()
            print(f"✓ 集合 '{self.collection_name}' 已删除")


# 使用示例
if __name__ == "__main__":
    # 初始化
    vs_manager = VectorStoreManager(
        embedding_model="nomic-embed-text",
        persist_directory="./my_knowledge_base",
    )
    
    # 尝试加载已有数据库
    vector_store = vs_manager.load_existing()
    
    if vector_store is None:
        # 如果没有，则创建新的
        from document_loader import SmartDocumentLoader
        
        loader = SmartDocumentLoader()
        chunks = loader.load_and_split_directory("./docs/")
        
        vector_store = vs_manager.create_from_documents(chunks)
    
    # 执行检索
    results = vs_manager.similarity_search_with_score(
        "如何配置数据库连接？",
        k=5,
    )
    
    for doc, score in results:
        print(f"\n分数: {score:.4f}")
        print(f"来源: {doc.metadata.get('source', 'unknown')}")
        print(f"内容: {doc.page_content[:150]}...")

3.3 RAG 链——将检索与生成串联

"""
rag_chain.py - RAG 核心链，串联检索与生成
"""
from typing import List, Dict, Any, Optional, Generator
from langchain_community.llms import Ollama
from langchain.prompts import ChatPromptTemplate
from langchain.schema import Document
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
import re


class RAGChain:
    """RAG 链——将检索到的上下文注入 LLM 生成"""
    
    def __init__(
        self,
        vector_store_manager,
        llm_model: str = "qwen2.5:7b",
        ollama_base_url: str = "http://localhost:11434",
        temperature: float = 0.3,
        max_tokens: int = 2048,
    ):
        """
        初始化 RAG 链
        
        Args:
            vector_store_manager: 向量存储管理器实例
            llm_model: Ollama 中 LLM 模型名称
            ollama_base_url: Ollama 服务地址
            temperature: 生成温度
            max_tokens: 最大生成 token 数
        """
        self.vs_manager = vector_store_manager
        self.llm_model = llm_model
        
        # 初始化 LLM
        self.llm = Ollama(
            model=llm_model,
            base_url=ollama_base_url,
            temperature=temperature,
            num_predict=max_tokens,
        )
        
        # 构建提示词模板
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", self._build_system_prompt()),
            ("human", "{question}"),
        ])
        
        # 构建 RAG 链
        self.chain = (
            {
                "context": self._retrieve_context,
                "question": RunnablePassthrough(),
            }
            | self.prompt
            | self.llm
            | StrOutputParser()
        )
    
    def _build_system_prompt(self) -> str:
        """构建系统提示词"""
        return """你是一个专业的知识库助手。请根据提供的参考文档回答用户的问题。

## 回答规则
1. 优先使用参考文档中的信息回答问题
2. 如果文档中有明确答案，直接引用相关内容
3. 如果文档信息不完整，可以结合你的知识进行补充，但要明确说明哪些来自文档，哪些是你的推理
4. 如果问题完全超出文档范围，诚实告知用户，并建议补充相关文档
5. 回答要结构清晰，重要信息使用列表或分点说明
6. 引用文档内容时，标注来源

## 参考文档
{context}

## 用户问题
"""
    
    def _retrieve_context(self, question: str) -> str:
        """检索相关上下文并格式化"""
        docs = self.vs_manager.similarity_search_with_score(question, k=5)
        
        if not docs:
            return "未找到相关文档。"
        
        context_parts = []
        for i, (doc, score) in enumerate(docs):
            source = doc.metadata.get("filename", doc.metadata.get("source", "未知来源"))
            # 分数归一化（ChromaDB 的 distance 分数）
            relevance = max(0, min(1, 1 - score / 2))
            context_parts.append(
                f"[文档{i+1} | 来源: {source} | 相关度: {relevance:.2f}]\n{doc.page_content}"
            )
        
        return "\n\n---\n\n".join(context_parts)
    
    def query(self, question: str) -> Dict[str, Any]:
        """
        执行 RAG 查询
        
        Returns:
            {
                "question": str,
                "answer": str,
                "sources": List[Dict],
            }
        """
        # 获取检索到的文档（用于返回来源）
        docs_with_scores = self.vs_manager.similarity_search_with_score(
            question, k=5
        )
        
        # 生成回答
        answer = self.chain.invoke(question)
        
        # 整理来源信息
        sources = []
        for doc, score in docs_with_scores:
            sources.append({
                "source": doc.metadata.get("filename", doc.metadata.get("source", "unknown")),
                "relevance": round(max(0, min(1, 1 - score / 2)), 2),
                "preview": doc.page_content[:100] + "...",
            })
        
        return {
            "question": question,
            "answer": answer,
            "sources": sources,
        }
    
    def query_stream(self, question: str) -> Generator[str, None, None]:
        """流式查询，逐字输出"""
        return self.chain.stream(question)


# 使用示例
if __name__ == "__main__":
    from vector_store import VectorStoreManager
    
    # 初始化向量存储
    vs_manager = VectorStoreManager(persist_directory="./my_knowledge_base")
    vs_manager.load_existing()
    
    # 初始化 RAG 链
    rag = RAGChain(
        vector_store_manager=vs_manager,
        llm_model="qwen2.5:7b",
    )
    
    # 提问
    result = rag.query("公司的远程办公政策是什么？")
    
    print("=" * 60)
    print(f"问题: {result['question']}")
    print("=" * 60)
    print(f"\n回答:\n{result['answer']}")
    print("\n" + "=" * 60)
    print("参考来源:")
    for src in result["sources"]:
        print(f"  📄 {src['source']} (相关度: {src['relevance']})")

---

四、进阶优化策略

基础 RAG 系统搭好后，实际使用中往往会遇到各种问题。以下是几个关键的优化方向：

4.1 混合检索——BM25 + 向量检索的融合

纯向量检索对精确关键词匹配不够好。混合检索结合传统 BM25 和向量相似度：

"""
hybrid_search.py - BM25 + 向量混合检索
"""
from typing import List, Tuple
from rank_bm25 import BM25Okapi
import jieba
import numpy as np
from langchain.schema import Document


class HybridRetriever:
    """混合检索器：BM25（关键词） + 向量（语义）"""
    
    def __init__(
        self,
        vector_store_manager,
        alpha: float = 0.5,
    ):
        """
        Args:
            vector_store_manager: 向量存储管理器
            alpha: 向量检索权重 (1-alpha 为 BM25 权重)
        """
        self.vs_manager = vector_store_manager
        self.alpha = alpha
        self.bm25 = None
        self.documents = []
    
    def build_bm25_index(self, documents: List[Document]):
        """构建 BM25 索引"""
        self.documents = documents
        # 中文分词后构建 BM25
        tokenized_corpus = [
            list(jieba.cut(doc.page_content)) for doc in documents
        ]
        self.bm25 = BM25Okapi(tokenized_corpus)
        print(f"BM25 索引构建完成，共 {len(documents)} 个文档")
    
    def search(
        self, 
        query: str, 
        k: int = 5
    ) -> List[Tuple[Document, float]]:
        """执行混合检索"""
        if self.bm25 is None or len(self.documents) == 0:
            # 如果没有 BM25 索引，回退到纯向量检索
            return self.vs_manager.similarity_search_with_score(query, k)
        
        # 1. 向量检索
        vector_results = self.vs_manager.similarity_search_with_score(
            query, k=k * 2
        )
        
        # 2. BM25 检索
        tokenized_query = list(jieba.cut(query))
        bm25_scores = self.bm25.get_scores(tokenized_query)
        
        # 获取 BM25 top-k 的文档及其分数
        bm25_top_indices = np.argsort(bm25_scores)[::-1][:k * 2]
        bm25_results = []
        for idx in bm25_top_indices:
            if bm25_scores[idx] > 0:
                bm25_results.append((self.documents[idx], bm25_scores[idx]))
        
        # 3. 分数融合（RRF: Reciprocal Rank Fusion）
        return self._reciprocal_rank_fusion(
            vector_results, 
            bm25_results, 
            k=k
        )
    
    def _reciprocal_rank_fusion(
        self,
        vector_results: List[Tuple[Document, float]],
        bm25_results: List[Tuple[Document, float]],
        k: int = 5,
        c: int = 60,
    ) -> List[Tuple[Document, float]]:
        """
        RRF 融合算法
        
        公式: RRF(d) = Σ 1/(c + rank_i(d))
        其中 c 是常数，rank_i 是文档在第 i 个排序列表中的排名
        """
        scores = {}
        doc_map = {}  # 用于存储文档对象
        
        # 处理向量检索结果
        for rank, (doc, _) in enumerate(vector_results):
            doc_id = doc.page_content[:100]  # 用内容前100字符做标识
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (c + rank + 1)
            doc_map[doc_id] = doc
        
        # 处理 BM25 结果
        for rank, (doc, _) in enumerate(bm25_results):
            doc_id = doc.page_content[:100]
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (c + rank + 1)
            doc_map[doc_id] = doc
        
        # 排序并返回 top-k
        sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        return [(doc_map[doc_id], score) for doc_id, score in sorted_docs[:k]]

4.2 查询重写——提升检索命中率

用户的问题往往不够精确，查询重写可以显著提升检索质量：

"""
query_rewriter.py - 查询重写优化检索
"""
from langchain_community.llms import Ollama


class QueryRewriter:
    """使用 LLM 改写用户查询，提升检索召回率"""
    
    def __init__(self, llm_model: str = "qwen2.5:7b"):
        self.llm = Ollama(model=llm_model, temperature=0.1)
        
        self.rewrite_prompt = """你是一个查询改写专家。请将用户的问题改写为更适合文档检索的形式。

改写规则：
1. 提取核心关键词和概念
2. 补全可能的同义词和相关术语
3. 如果是口语化表达，改为正式书面语
4. 将模糊描述具体化
5. 直接输出改写后的查询，不要解释

原始问题：{question}

改写后的查询："""
    
    def rewrite(self, question: str) -> str:
        """改写查询"""
        prompt = self.rewrite_prompt.format(question=question)
        rewritten = self.llm.invoke(prompt)
        return rewritten.strip()
    
    def multi_query_rewrite(self, question: str, n: int = 3) -> list:
        """生成多个改写版本，用于多路召回"""
        multi_prompt = f"""请将以下问题从 {n} 个不同角度改写为检索查询，
每个角度一行，不要编号，不要解释。

原始问题：{question}

改写查询："""
        
        result = self.llm.invoke(multi_prompt)
        queries = [q.strip() for q in result.strip().split('\n') if q.strip()]
        return queries[:n]


# 使用示例
if __name__ == "__main__":
    rewriter = QueryRewriter()
    
    original = "那个远程工作的规定是啥来着？"
    rewritten = rewriter.rewrite(original)
    print(f"原始: {original}")
    print(f"改写: {rewritten}")
    # 输出: 公司远程办公政策规定

4.3 对话历史管理

真正的对话助手需要记住上下文：

"""
conversation_manager.py - 对话历史管理
"""
from typing import List, Dict
from collections import deque


class ConversationManager:
    """管理多轮对话历史"""
    
    def __init__(self, max_history: int = 5, max_tokens_estimate: int = 4000):
        self.max_history = max_history
        self.max_tokens = max_tokens_estimate
        self.history: deque = deque(maxlen=max_history)
    
    def add_turn(self, question: str, answer: str):
        """添加一轮对话"""
        self.history.append({
            "question": question,
            "answer": answer,
        })
    
    def get_context(self, max_turns: int = 3) -> str:
        """获取最近的对话上下文"""
        if not self.history:
            return "这是对话的开始。"
        
        recent = list(self.history)[-max_turns:]
        context_parts = ["## 对话历史"]
        
        for i, turn in enumerate(recent, 1):
            context_parts.append(f"用户{i}: {turn['question']}")
            context_parts.append(f"助手{i}: {turn['answer'][:200]}...")
        
        return "\n".join(context_parts)
    
    def clear(self):
        """清空历史"""
        self.history.clear()
    
    def get_history_list(self) -> List[Dict]:
        """获取完整的对话历史列表"""
        return list(self.history)

---

五、完整 Web 应用——用 Streamlit 构建 UI

将所有组件整合为一个可用的 Web 应用：

"""
app.py - 完整的 RAG 知识库 Web 应用
"""
import streamlit as st
import os
import tempfile
from pathlib import Path

from document_loader import SmartDocumentLoader
from vector_store import VectorStoreManager
from rag_chain import RAGChain
from query_rewriter import QueryRewriter
from conversation_manager import ConversationManager


# ---- 页面配置 ----
st.set_page_config(
    page_title="本地 AI 知识库助手",
    page_icon="📚",
    layout="wide",
)

# ---- 初始化 Session State ----
def init_session_state():
    """初始化会话状态"""
    if "rag_chain" not in st.session_state:
        st.session_state.rag_chain = None
    if "vs_manager" not in st.session_state:
        st.session_state.vs_manager = None
    if "conversation" not in st.session_state:
        st.session_state.conversation = ConversationManager()
    if "messages" not in st.session_state:
        st.session_state.messages = []
    if "knowledge_base_ready" not in st.session_state:
        st.session_state.knowledge_base_ready = False

init_session_state()

# ---- 侧边栏 ----
with st.sidebar:
    st.title("📚 知识库管理")
    
    # 向量存储路径
    db_path = st.text_input("向量数据库路径", value="./chroma_db")
    collection_name = st.text_input("集合名称", value="knowledge_base")
    
    # 初始化向量存储管理器
    if st.button("🔗 连接/初始化知识库"):
        with st.spinner("正在连接..."):
            vs_manager = VectorStoreManager(
                persist_directory=db_path,
                collection_name=collection_name,
            )
            vector_store = vs_manager.load_existing()
            
            if vector_store:
                st.session_state.vs_manager = vs_manager
                st.session_state.rag_chain = RAGChain(
                    vector_store_manager=vs_manager,
                )
                st.session_state.knowledge_base_ready = True
                st.success("✅ 知识库连接成功！")
            else:
                st.warning("⚠️ 未找到知识库，请先上传文档")
    
    st.divider()
    
    # 文档上传
    st.subheader("📤 上传文档")
    uploaded_files = st.file_uploader(
        "支持 PDF, Word, Markdown, TXT",
        accept_multiple_files=True,
        type=["pdf", "docx", "md", "txt"],
    )
    
    if uploaded_files and st.button("📥 导入知识库"):
        with st.spinner("正在处理文档..."):
            # 保存上传文件到临时目录
            temp_dir = tempfile.mkdtemp()
            saved_paths = []
            
            for uploaded_file in uploaded_files:
                file_path = os.path.join(temp_dir, uploaded_file.name)
                with open(file_path, "wb") as f:
                    f.write(uploaded_file.getbuffer())
                saved_paths.append(file_path)
            
            # 加载并分割文档
            loader = SmartDocumentLoader()
            all_chunks = []
            for path in saved_paths:
                chunks = loader.load_and_split_file(path)
                all_chunks.extend(chunks)
            
            # 创建向量存储
            vs_manager = VectorStoreManager(
                persist_directory=db_path,
                collection_name=collection_name,
            )
            vs_manager.create_from_documents(all_chunks)
            
            # 更新会话状态
            st.session_state.vs_manager = vs_manager
            st.session_state.rag_chain = RAGChain(
                vector_store_manager=vs_manager,
            )
            st.session_state.knowledge_base_ready = True
            
            st.success(f"✅ 成功导入 {len(uploaded_files)} 个文件，共 {len(all_chunks)} 个文本块")
    
    st.divider()
    
    # 设置
    st.subheader("⚙️ 模型设置")
    llm_model = st.selectbox(
        "LLM 模型",
        ["qwen2.5:7b", "qwen2.5:14b", "llama3.1:8b", "mistral:7b"],
        index=0,
    )
    temperature = st.slider("温度", 0.0, 1.0, 0.3, 0.1)
    retrieval_k = st.slider("检索文档数", 1, 10, 5)
    
    if st.button("🔄 应用设置"):
        if st.session_state.vs_manager:
            st.session_state.rag_chain = RAGChain(
                vector_store_manager=st.session_state.vs_manager,
                llm_model=llm_model,
                temperature=temperature,
            )
            st.success("设置已更新")


# ---- 主界面 ----
st.title("🤖 本地 AI 知识库助手")
st.caption("基于 Ollama + RAG | 完全离线 | 数据安全")

# 聊天界面
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])
        if "sources" in message and message["sources"]:
            with st.expander("📎 参考来源"):
                for src in message["sources"]:
                    st.caption(f"📄 {src['source']} (相关度: {src['relevance']})")

# 输入框
if prompt := st.chat_input("请输入你的问题..."):
    if not st.session_state.knowledge_base_ready:
        st.error("请先在侧边栏连接或创建知识库")
    else:
        # 添加用户消息
        st.session_state.messages.append({"role": "user", "content": prompt})
        with st.chat_message("user"):
            st.markdown(prompt)
        
        # 生成回答
        with st.chat_message("assistant"):
            with st.spinner("思考中..."):
                # 查询重写（可选）
                rewriter = QueryRewriter()
                rewritten_query = rewriter.rewrite(prompt)
                
                # RAG 查询
                result = st.session_state.rag_chain.query(rewritten_query)
                
                st.markdown(result["answer"])
                
                # 显示来源
                if result["sources"]:
                    with st.expander("📎 参考来源"):
                        for src in result["sources"]:
                            st.caption(f"📄 {src['source']} (相关度: {src['relevance']})")
        
        # 保存消息
        st.session_state.messages.append({
            "role": "assistant",
            "content": result["answer"],
            "sources": result["sources"],
        })
        
        # 更新对话历史
        st.session_state.conversation.add_turn(prompt, result["answer"])

# 清空对话按钮
if st.session_state.messages:
    if st.button("🗑️ 清空对话"):
        st.session_state.messages = []
        st.session_state.conversation.clear()
        st.rerun()

---

六、运行与部署

6.1 启动步骤

# 1. 确保 Ollama 正在运行
ollama serve

# 2. 确保已下载模型
ollama pull qwen2.5:7b
ollama pull nomic-embed-text

# 3. 启动 Web 应用
streamlit run app.py

# 4. 访问 http://localhost:8501

6.2 Docker 部署（可选）

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8501

CMD ["streamlit", "run", "app.py", "--server.address=0.0.0.0"]

6.3 性能基准参考

配置	文档大小	检索速度	生成速度
M1 MacBook Air 16GB	1000页 PDF	<2s	~15 token/s
RTX 3060 12GB	5000页 PDF	<1s	~40 token/s
RTX 4090 24GB	10000页 PDF	<0.5s	~80 token/s
CPU Only (i9)	500页 PDF	<5s	~5 token/s

---

七、常见问题与解决方案

问题1：检索结果不相关

解决方案：

• 调整 chunk_size（建议 300-800）
• 增大 chunk_overlap（建议 chunk_size 的 10-20%）
• 使用查询重写模块
• 启用混合检索

问题2：Ollama 生成速度慢

解决方案：

• 使用更小的模型（如 qwen2.5:3b）
• 设置 num_predict 限制生成长度
• 使用 GPU 加速（Ollama 自动检测）
• 开启 num_gpu 参数指定 GPU 层数

问题3：内存不足

解决方案：

• 减小 chunk_size
• 分批处理文档
• 使用更小的 Embedding 模型
• 限制 k 的检索数量

---

八、更进一步——企业级增强

如果你想让这个系统真正用于生产环境，可以考虑以下增强：

1. 多模态支持：接入 OCR 处理扫描件，支持图片问答
2. 权限管理：不同用户可见不同的文档集合
3. 增量更新：新增文档无需重建整个向量库
4. 评估体系：RAGAS 框架自动评估检索和生成质量
5. Agent 能力：让助手不仅能回答问题，还能调用 API、查询数据库

---

结语

Ollama + RAG 的组合极大降低了私有化 AI 知识库的门槛。你现在拥有的不仅是一个演示项目，而是一个真正可以在企业中落地的方案。

核心价值：

• 🔒 数据不出本地：所有处理都在你的机器上完成
• 💰 零 API 费用：告别按 token 计费
• 🎛️ 完全可控：模型、数据、提示词全部可定制
• 📱 离线可用：不依赖互联网连接

下一步，我建议你用自己的真实文档跑一遍，感受从"通用 AI"到"专属知识库助手"的质变。当 AI 准确回答出只存在于你文档中的信息时，你会真正理解 RAG 的威力。

---