使用 Ollama + LangChain 构建本地文档问答系统

背景介绍

日常工作中有大量技术文档、论文和内部知识库需要处理。关键词搜索的痛点在于无法理解语义查询——搜"如何安装软件"和"安装步骤"本应返回相同结果，但传统搜索引擎办不到。

云端大语言模型能力很强，数据隐私是个问题。把内部文档上传到第三方 API 服务，多少有些顾虑。另外调用成本也是持续支出。

Ollama 出现后，本地运行 LLM 成为现实。配合 LangChain 的 RAG（检索增强生成）能力，可以在个人电脑上搭建完全本地化的文档问答系统。数据不出本机，语义检索替代关键词匹配。

这篇文章会详细演示：从零开始，用 Ollama 运行开源模型（Llama 3、Qwen 等），结合 LangChain 和 Chroma 向量数据库，搭建一个支持 PDF、Markdown、TXT 格式的本地文档问答系统。

问题描述

搭建这个系统之前，需要解决几个核心问题：

文档处理和语义检索：传统倒排索引只认关键词，不理解语义。需要把文档内容转成向量，用相似度搜索来实现语义匹配。

大模型选择：不同模型在推理能力、速度和硬件要求上差异很大。消费级 GPU 显存有限，7B 参数的模型是合理的起点。纯 CPU 运行的话，16GB 内存是底线。

文本预处理：文档需要分块、向量化、存储。分块策略直接影响检索效果——块太小丢失上下文，块太大引入噪音。

答案生成：检索到的内容如何组织成流畅的回答，需要设计合适的提示词模板，避免大模型 hallucinate（编造内容）。

下面用完整代码逐一解决这些问题。

详细步骤

环境准备

Ollama 支持 macOS、Linux 和 Windows。这里以 Ubuntu/Debian 为例：

# 安装 Ollama
curl -fsSL https://ollama.com/install.sh | sh

# 验证安装
ollama --version

模型选择方面，考虑到硬件限制，推荐 Qwen2:7B 或 Llama3:8B。纯 CPU 环境需要至少 16GB 内存才能流畅运行 7B 模型。

# 下载 Qwen2 7B（推荐，中文理解能力强)
ollama pull qwen2:7b

# 也可用 Llama3 8B
# ollama pull llama3:8b

# 查看已下载的模型
ollama list

安装 Python 依赖：

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

# 安装 LangChain 和相关依赖
pip install langchain langchain-community langchain-ollama \
    chromadb langchain-text-splitters \
    pypdf pymupdf

文档处理与向量化

第一步是把各种格式的文档转成文本，然后分块。每个文本块转成向量，存入向量数据库。

创建 document_processor.py：

"""文档处理模块：支持 PDF、Markdown、TXT 格式的文档加载和分块"""
import os
from pathlib import Path
from typing import List
from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    MarkdownLoader,
    UnstructuredMarkdownLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document

class DocumentProcessor:
    """文档处理器，支持多种格式的加载和分块"""

    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        """
        初始化文档处理器

        Args:
            chunk_size: 每个文本块的最大字符数
            chunk_overlap: 相邻文本块之间的重叠字符数
        """
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

        # 根据不同文档类型配置加载器
        self.loaders = {
            ".pdf": PyPDFLoader,
            ".txt": TextLoader,
            ".md": UnstructuredMarkdownLoader,
            ".markdown": UnstructuredMarkdownLoader,
        }

        # 文本分割器配置
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len,
            separators=["\n\n", "\n", "。", ". ", " ", ""],
            keep_separator=True
        )

    def load_document(self, file_path: str) -> List[Document]:
        """加载单个文档"""
        path = Path(file_path)
        suffix = path.suffix.lower()

        if suffix not in self.loaders:
            raise ValueError(f"不支持的文档格式: {suffix}")

        loader_class = self.loaders[suffix]
        loader = loader_class(file_path)

        # 编码问题处理
        if suffix == ".txt":
            loader = TextLoader(file_path, encoding="utf-8")

        return loader.load()

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

        for file_path in path.rglob("*"):
            if file_path.is_file() and file_path.suffix.lower() in self.loaders:
                try:
                    docs = self.load_document(str(file_path))
                    # 为每个文档添加来源信息
                    for doc in docs:
                        doc.metadata["source"] = str(file_path)
                    all_docs.extend(docs)
                    print(f"已加载: {file_path.name}")
                except Exception as e:
                    print(f"加载失败 {file_path.name}: {e}")

        return all_docs

    def split_documents(self, documents: List[Document]) -> List[Document]:
        """将文档分割成小块"""
        return self.text_splitter.split_documents(documents)

# 测试代码
if __name__ == "__main__":
    # 示例用法
    processor = DocumentProcessor(chunk_size=500, chunk_overlap=50)

    # 加载单个 PDF
    # docs = processor.load_document("sample.pdf")

    # 加载整个目录
    # all_docs = processor.load_directory("./documents")
    # chunks = processor.split_documents(all_docs)

    print("文档处理器初始化完成")

向量存储与检索

创建 vector_store.py，用 Chroma 作为向量数据库，Ollama 提供嵌入模型：

"""向量存储模块：使用 Chroma 和 Ollama 嵌入模型构建向量数据库"""
import os
from typing import List, Optional
from langchain_ollama import OllamaEmbeddings
from langchain_chroma import Chroma
from langchain.schema import Document

class VectorStoreManager:
    """向量存储管理器"""

    def __init__(
        self,
        persist_directory: str = "./chroma_db",
        embedding_model: str = "nomic-embed-text",
        model_base_url: str = "http://localhost:11434"
    ):
        """
        初始化向量存储管理器

        Args:
            persist_directory: Chroma 数据库持久化目录
            embedding_model: 嵌入模型名称
            model_base_url: Ollama 服务地址
        """
        self.persist_directory = persist_directory
        self.embedding_model = embedding_model
        self.model_base_url = model_base_url

        # 初始化嵌入模型
        self.embeddings = OllamaEmbeddings(
            model=embedding_model,
            base_url=model_base_url
        )

        # 向量存储实例
        self.vector_store: Optional[Chroma] = None

    def create_vector_store(self, documents: List[Document]) -> Chroma:
        """从文档创建向量存储"""
        print(f"正在创建向量数据库，包含 {len(documents)} 个文本块...")

        self.vector_store = Chroma.from_documents(
            documents=documents,
            embedding=self.embeddings,
            persist_directory=self.persist_directory
        )

        print(f"向量数据库已保存至: {self.persist_directory}")
        return self.vector_store

    def load_vector_store(self) -> Chroma:
        """加载已存在的向量数据库"""
        if not os.path.exists(self.persist_directory):
            raise FileNotFoundError(f"向量数据库不存在: {self.persist_directory}")

        self.vector_store = Chroma(
            persist_directory=self.persist_directory,
            embedding_function=self.embeddings
        )

        print(f"已加载向量数据库，包含 {self.vector_store._collection.count()} 个向量")
        return self.vector_store

    def similarity_search(
        self,
        query: str,
        k: int = 5,
        filter: Optional[dict] = None
    ) -> List[Document]:
        """相似度搜索"""
        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 = 5
    ) -> List[tuple]:
        """带分数的相似度搜索"""
        if self.vector_store is None:
            raise ValueError("向量数据库未初始化")

        return self.vector_store.similarity_search_with_score(
            query=query,
            k=k
        )

# 测试代码
if __name__ == "__main__":
    # 示例：创建新的向量数据库
    # from document_processor import DocumentProcessor
    # 
    # processor = DocumentProcessor()
    # docs = processor.load_directory("./documents")
    # chunks = processor.split_documents(docs)
    # 
    # manager = VectorStoreManager()
    # manager.create_vector_store(chunks)

    print("向量存储管理器初始化完成")

RAG 问答系统

核心部分：整合大模型和检索模块。创建 rag_qa_system.py：

"""RAG 问答系统：整合检索和大模型生成答案"""
from typing import List, Optional, Dict
from langchain_ollama import ChatOllama
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.prompts import ChatPromptTemplate
from langchain.schema import Document
from vector_store import VectorStoreManager

class RAGQASystem:
    """基于 RAG 的问答系统"""

    def __init__(
        self,
        llm_model: str = "qwen2:7b",
        temperature: float = 0.3,
        vector_store_path: str = "./chroma_db",
        retrieval_k: int = 5
    ):
        """
        初始化 RAG 问答系统

        Args:
            llm_model: 大语言模型名称
            temperature: 生成温度，控制随机性
            vector_store_path: 向量数据库路径
            retrieval_k: 检索返回的文档数量
        """
        self.llm_model = llm_model
        self.retrieval_k = retrieval_k

        # 初始化大语言模型
        self.llm = ChatOllama(
            model=llm_model,
            temperature=temperature,
            base_url="http://localhost:11434"
        )

        # 加载向量数据库
        self.vector_manager = VectorStoreManager(
            persist_directory=vector_store_path
        )
        self.retriever = self.vector_manager.load_vector_store().as_retriever(
            search_kwargs={"k": retrieval_k}
        )

        # 构建提示词模板
        self._build_prompt()

        # 创建问答链
        self._build_chain()

    def _build_prompt(self):
        """构建提示词模板"""
        # 使用中文提示词，优化回答风格
        self.prompt = ChatPromptTemplate.from_template(
            """你是一个专业的技术文档问答助手。请根据以下参考文档回答用户的问题。

参考文档：
{context}

用户问题：{input}

要求：
1. 只根据提供的参考文档回答，不要编造信息
2. 如果参考文档中没有相关信息，请明确告知用户
3. 回答要简洁准确，使用中文
4. 在回答末尾注明参考来源

你的回答："""
        )

    def _build_chain(self):
        """构建检索-回答链"""
        # 创建文档合并链
        self.combine_docs_chain = create_stuff_documents_chain(
            self.llm,
            self.prompt
        )

        # 创建完整的检索链
        self.retrieval_chain = create_retrieval_chain(
            self.retriever,
            self.combine_docs_chain
        )

    def ask(self, question: str) -> Dict:
        """
        提问并获取答案

        Args:
            question: 用户问题

        Returns:
            包含答案和参考来源的字典
        """
        # 执行问答
        result = self.retrieval_chain.invoke({"input": question})

        # 提取参考来源
        sources = []
        for doc in result.get("context", []):
            source_info = {
                "content": doc.page_content[:200] + "...",
                "source": doc.metadata.get("source", "未知")
            }
            sources.append(source_info)

        return {
            "answer": result["answer"],
            "sources": sources,
            "question": question
        }

    def ask_with_stream(self, question: str):
        """流式输出回答（如果模型支持）"""
        # 获取检索结果
        retrieved_docs = self.retriever.invoke(question)

        # 构建文档上下文
        context = "\n\n".join(doc.page_content for doc in retrieved_docs)

        # 构建完整提示
        full_prompt = f"""你是一个专业的技术文档问答助手。请根据以下参考文档回答用户的问题。

参考文档：
{context}

用户问题：{question}

要求：
1. 只根据提供的参考文档回答，不要编造信息
2. 如果参考文档中没有相关信息，请明确告知用户
3. 回答要简洁准确，使用中文

你的回答："""

        # 流式调用
        for chunk in self.llm.stream(full_prompt):
            yield chunk.content

# 主程序入口
if __name__ == "__main__":
    # 初始化系统
    print("正在初始化 RAG 问答系统...")
    print("确保 Ollama 服务正在运行 (ollama serve)")

    qa_system = RAGQASystem(
        llm_model="qwen2:7b",
        temperature=0.3,
        vector_store_path="./chroma_db"
    )

    # 交互式问答循环
    print("\n=== RAG 问答系统已就绪 ===")
    print("输入问题进行咨询，输入 quit 或 exit 退出\n")

    while True:
        question = input("请输入问题: ")

        if question.lower() in ["quit", "exit", "q"]:
            print("再见!")
            break

        if not question.strip():
            continue

        print("\n正在思考...")

        try:
            result = qa_system.ask(question)

            print("\n" + "="*50)
            print("回答:")
            print(result["answer"])
            print("\n参考来源:")
            for i, source in enumerate(result["sources"], 1):
                print(f"  {i}. {source[\"source\"]}")
            print("="*50 + "\n")

        except Exception as e:
            print(f"Error: {e}")

构建索引脚本

最后创建 build_index.py，一键构建文档索引：

"""构建文档索引的脚本"""
import os
import argparse
from document_processor import DocumentProcessor
from vector_store import VectorStoreManager

def main():
    parser = argparse.ArgumentParser(description="构建文档向量索引")
    parser.add_argument(
        "--docs-dir",
        type=str,
        default="./documents",
        help="文档目录路径"
    )
    parser.add_argument(
        "--output-dir",
        type=str,
        default="./chroma_db",
        help="向量数据库输出路径"
    )
    parser.add_argument(
        "--chunk-size",
        type=int,
        default=500,
        help="文本块大小"
    )
    parser.add_argument(
        "--chunk-overlap",
        type=int,
        default=50,
        help="文本块重叠大小"
    )
    parser.add_argument(
        "--embedding-model",
        type=str,
        default="nomic-embed-text",
        help="嵌入模型名称"
    )

    args = parser.parse_args()

    # 检查文档目录
    if not os.path.exists(args.docs_dir):
        print(f"错误: 文档目录不存在: {args.docs_dir}")
        print("请创建目录并放入待索引的文档（PDF、Markdown、TXT）")
        return

    print(f"文档目录: {args.docs_dir}")
    print(f"输出路径: {args.output_dir}")
    print(f"文本块大小: {args.chunk_size}")
    print(f"嵌入模型: {args.embedding_model}")
    print("-" * 50)

    # 加载文档
    processor = DocumentProcessor(
        chunk_size=args.chunk_size,
        chunk_overlap=args.chunk_overlap
    )

    print("\n正在加载文档...")
    documents = processor.load_directory(args.docs_dir)
    print(f"共加载 {len(documents)} 个文档")

    # 分割文档
    print("\n正在分割文档...")
    chunks = processor.split_documents(documents)
    print(f"共生成 {len(chunks)} 个文本块")

    # 创建向量存储
    print("\n正在生成向量嵌入...")
    vector_manager = VectorStoreManager(
        persist_directory=args.output_dir,
        embedding_model=args.embedding_model
    )
    vector_manager.create_vector_store(chunks)

    print("\n" + "=" * 50)
    print("索引构建完成!")
    print(f"向量数据库位置: {args.output_dir}")
    print("=" * 50)

if __name__ == "__main__":
    main()

运行结果

索引构建

准备一个包含文档的目录，运行索引构建脚本：

# 创建文档目录并放入文档
mkdir -p documents
# 将 PDF、Markdown 或 TXT 文件放入 documents 目录

# 运行索引构建
python build_index.py --docs-dir ./documents --chunk-size 500

运行输出示例：

文档目录: ./documents
输出路径: ./chroma_db
文本块大小: 500
嵌入模型: nomic-embed-text
--------------------------------------------------

正在加载文档...
已加载: 技术文档.pdf
已加载: 常见问题.md
已加载: 使用指南.txt
共加载 3 个文档

正在分割文档...
共生成 156 个文本块

正在生成向量嵌入...
正在创建向量数据库，包含 156 个文本块...
向量数据库已保存至: ./chroma_db

==================================================
索引构建完成!
向量数据库位置: ./chroma_db
==================================================

问答测试

启动 Ollama 服务，运行问答系统：

# 确保 Ollama 服务正在运行
ollama serve

# 启动问答系统
python rag_qa_system.py

交互示例：

请输入问题: 如何安装这个软件？

正在思考...

==================================================
回答:
根据参考文档，安装步骤如下：

1. 下载安装包
2. 解压到指定目录
3. 运行安装脚本：
   ```bash
   ./install.sh
   ```
4. 配置环境变量

注意：安装前请确保系统满足最低配置要求（Python 3.8+，4GB 内存）。

参考来源:
  1. 使用指南.txt
  2. 技术文档.pdf
==================================================

另一个例子：

请输入问题: 这个工具支持哪些编程语言？

正在思考...

==================================================
回答:
根据已加载的文档，该工具支持以下编程语言：

- Python
- JavaScript/TypeScript
- Go
- Rust

详细的功能对比请参考技术文档中的语言支持章节。

参考来源:
  1. 技术文档.pdf
  2. 常见问题.md
==================================================

总结

这套系统的核心优势是数据完全本地化，不需要网络就能用。隐私敏感的场景下，这个方案比云端 API 安全得多。

技术链路其实很清晰：文档通过 LangChain 的加载器处理，用 RecursiveCharacterTextSplitter 分块，Ollama 的嵌入模型转成向量存进 Chroma。查询时先做向量检索，把最相关的文本块交给大模型生成答案。

几个可优化方向：模型层面可以换更强的 72B 模型；分块策略可以尝试按章节或段落而非固定字符数；加个对话历史缓存实现多轮对话；简单包装个 Web 界面给非技术用户使用。

个人开发者或小团队用这个方案，成本基本就是电费。不需要把内部文档交给第三方，自己掌控数据，部署一套专属的智能文档助手。