从零开始学LangChain（六）：实战项目 - 智能文档问答系统

cuizhenjie2026-01-042026-01-06

从零开始学LangChain（六）：实战项目 - 智能文档问答系统

本系列教程将带你从零开始学习LangChain框架，构建强大的AI应用程序。

项目概述

我们将构建一个智能文档问答系统，能够上传PDF/Word/Txt文档，自动解析内容，智能回答相关问题，记住对话上下文，并引用原文来源。

技术栈

系统使用OpenAI GPT-3.5/4作为语言模型，Chroma作为向量存储，LangChain Loaders解析文档，RecursiveCharacterTextSplitter分割文本，Similarity Search进行相似度搜索。这样的技术组合能够实现高效的文档检索和准确的答案生成。

应用场景

企业知识库问答、技术文档助手、学习资料辅导、合同条款查询等场景都可以使用这个系统。

项目架构

系统流程分为检索阶段、生成阶段和记忆管理。检索阶段向量化问题，搜索相似文档片段获取上下文。生成阶段构建提示词，调用LLM生成答案并返回结果。记忆管理保存对话历史。

步骤1：环境准备

安装依赖

# 创建项目目录
mkdir langchain-qa-system
cd langchain-qa-system

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

# 安装依赖
pip install langchain
pip install langchain-openai
pip install langchain-community
pip install chromadb
pip install pypdf
pip install python-dotenv
pip install tiktoken

配置环境变量

创建.env文件：

1 2	# .env OPENAI_API_KEY=your-api-key-here

步骤2：创建核心模块

2.1 文档加载模块

# document_loader.py
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from pathlib import Path
from typing import List
from langchain.schema import Document

class DocumentLoader:
    """文档加载器"""

    def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200):
        """
        初始化文档加载器

        Args:
            chunk_size: 每个文本块的大小
            chunk_overlap: 文本块之间的重叠大小
        """
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len,
            separators=["\n\n", "\n", "。", "！", "？", " ", ""]
        )

    def load_pdf(self, file_path: str) -> List[Document]:
        """加载PDF文档"""
        loader = PyPDFLoader(file_path)
        documents = loader.load()
        return self.text_splitter.split_documents(documents)

    def load_txt(self, file_path: str) -> List[Document]:
        """加载文本文件"""
        loader = TextLoader(file_path, encoding='utf-8')
        documents = loader.load()
        return self.text_splitter.split_documents(documents)

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

        for file in path.rglob("*"):
            if file.is_file():
                if file.suffix == ".pdf":
                    docs = self.load_pdf(str(file))
                elif file.suffix == ".txt":
                    docs = self.load_txt(str(file))
                else:
                    continue

                all_documents.extend(docs)

        return all_documents

2.2 向量存储模块

# vector_store.py
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.schema import Document
from typing import List, Optional
import shutil

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

    def __init__(self, persist_directory: str = "./chroma_db"):
        """
        初始化向量存储

        Args:
            persist_directory: 持久化目录
        """
        self.persist_directory = persist_directory
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = None

    def create_from_documents(self, documents: List[Document]) -> Chroma:
        """从文档创建向量存储"""
        # 删除旧的向量存储
        if self.vectorstore is None:
            self.vectorstore = Chroma.from_documents(
                documents=documents,
                embedding=self.embeddings,
                persist_directory=self.persist_directory
            )
            self.vectorstore.persist()
        return self.vectorstore

    def load_existing(self) -> Optional[Chroma]:
        """加载已存在的向量存储"""
        try:
            self.vectorstore = Chroma(
                persist_directory=self.persist_directory,
                embedding_function=self.embeddings
            )
            return self.vectorstore
        except Exception as e:
            print(f"加载向量存储失败: {e}")
            return None

    def similarity_search(self, query: str, k: int = 3) -> List[Document]:
        """相似度搜索"""
        if self.vectorstore is None:
            raise ValueError("向量存储未初始化")

        return self.vectorstore.similarity_search(query, k=k)

    def clear(self):
        """清除向量存储"""
        if self.persist_directory and Path(self.persist_directory).exists():
            shutil.rmtree(self.persist_directory)
        self.vectorstore = None

2.3 问答链模块

# qa_chain.py
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferWindowMemory
from langchain.schema.runnable import RunnablePassthrough, RunnableParallel
from langchain.output_parsers import StrOutputParser
from typing import List, Dict

class QAChain:
    """问答链"""

    def __init__(
        self,
        vector_store,
        model_name: str = "gpt-3.5-turbo",
        temperature: float = 0.3,
        k: int = 3
    ):
        """
        初始化问答链

        Args:
            vector_store: 向量存储实例
            model_name: 模型名称
            temperature: 温度参数
            k: 检索的文档数量
        """
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        self.vector_store = vector_store
        self.k = k
        self.memory = ConversationBufferWindowMemory(
            k=5,
            return_messages=True,
            output_key="answer"
        )

        # 创建提示词模板
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一位专业的文档助手。请根据以下上下文回答用户的问题。

上下文：
{context}

要求：
1. 只使用提供的上下文信息回答
2. 如果上下文中没有相关信息，明确告诉用户
3. 回答要准确、简洁、专业
4. 引用原文时标注来源"""),
            MessagesPlaceholder(variable_name="history"),
            ("human", "{question}")
        ])

    def _get_context(self, question: str) -> str:
        """获取相关上下文"""
        docs = self.vector_store.similarity_search(question, k=self.k)
        context = "\n\n".join([
            f"【来源{i+1}】\n{doc.page_content}"
            for i, doc in enumerate(docs)
        ])
        return context

    def create_chain(self):
        """创建问答链"""

        def retrieve_and_format(inputs):
            """检索并格式化输入"""
            question = inputs["question"]
            context = self._get_context(question)
            return {
                "context": context,
                "question": question,
                "history": inputs.get("history", [])
            }

        # 创建链
        chain = (
            RunnablePassthrough.assign(**retrieve_and_format)
            | self.prompt
            | self.llm
            | StrOutputParser()
        )

        return chain

    def ask(self, question: str) -> Dict[str, str]:
        """提问"""
        chain = self.create_chain()

        # 生成答案
        answer = chain.invoke({
            "question": question,
            "history": self.memory.load_memory_variables({})["history"]
        })

        # 保存到记忆
        self.memory.save_context(
            {"question": question},
            {"answer": answer}
        )

        return {"answer": answer, "question": question}

    def clear_memory(self):
        """清除对话历史"""
        self.memory.clear()

步骤3：主程序集成

# main.py
import os
from dotenv import load_dotenv
from document_loader import DocumentLoader
from vector_store import VectorStore
from qa_chain import QAChain

class DocumentQASystem:
    """文档问答系统"""

    def __init__(self, vector_db_path: str = "./chroma_db"):
        load_dotenv()
        self.vector_store = VectorStore(vector_db_path)
        self.doc_loader = DocumentLoader()
        self.qa_chain = None

    def index_documents(self, source_path: str):
        """索引文档"""
        print(f"正在加载文档: {source_path}")

        # 加载文档
        if os.path.isfile(source_path):
            if source_path.endswith(".pdf"):
                documents = self.doc_loader.load_pdf(source_path)
            else:
                documents = self.doc_loader.load_txt(source_path)
        else:
            documents = self.doc_loader.load_directory(source_path)

        print(f"共加载 {len(documents)} 个文档片段")

        # 创建向量存储
        print("正在创建向量索引...")
        self.vector_store.create_from_documents(documents)
        print("索引创建完成！")

    def load_index(self):
        """加载已存在的索引"""
        self.vector_store.load_existing()
        print("向量索引加载完成")

    def create_qa_chain(self):
        """创建问答链"""
        self.qa_chain = QAChain(self.vector_store)
        print("问答系统初始化完成")

    def ask(self, question: str) -> str:
        """提问"""
        if self.qa_chain is None:
            raise ValueError("问答系统未初始化，请先调用 create_qa_chain()")

        result = self.qa_chain.ask(question)
        return result["answer"]

    def chat_loop(self):
        """交互式对话循环"""
        print("\n=== 文档问答系统 ===")
        print("输入 'quit' 或 'exit' 退出")
        print("输入 'clear' 清除对话历史")
        print("-" * 50)

        while True:
            user_input = input("\n您的问题: ").strip()

            if user_input.lower() in ['quit', 'exit']:
                print("再见！")
                break

            if user_input.lower() == 'clear':
                self.qa_chain.clear_memory()
                print("对话历史已清除")
                continue

            if not user_input:
                continue

            try:
                answer = self.ask(user_input)
                print(f"\n答案:\n{answer}")
            except Exception as e:
                print(f"抱歉，出现问题: {e}")

# 使用示例
if __name__ == "__main__":
    system = DocumentQASystem()

    # 首次使用：索引文档
    # system.index_documents("./documents")

    # 后续使用：加载索引
    system.load_index()
    system.create_qa_chain()

    # 开始对话
    system.chat_loop()

步骤4：创建示例文档

创建测试文档documents/sample.txt：

Python是一种高级编程语言，由Guido van Rossum于1991年创建。
Python的设计哲学强调代码的可读性和简洁的语法。
特别是使用空格缩进来划分代码块，而不是使用大括号或关键字。

Python支持多种编程范式，包括面向对象、命令式、函数式和过程式编程。
它拥有一个动态类型系统和自动内存管理功能。

Python的应用领域非常广泛，包括：
1. Web开发（Django、Flask、FastAPI）
2. 数据科学和机器学习（NumPy、Pandas、Scikit-learn）
3. 自动化脚本和测试
4. 科学计算
5. 游戏开发

Python的官方网站是python.org，可以在那里找到最新的版本和文档。

优化建议

提升检索质量

使用混合检索结合相似度和关键词搜索，提高检索准确性。

from langchain.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

def create_hybrid_retriever(vector_store, docs):
    """创建混合检索器"""
    # 向量检索
    vector_retriever = vector_store.as_retriever(search_kwargs={"k": 3})

    # 关键词检索
    bm25_retriever = BM25Retriever.from_documents(docs)

    # 集成检索
    ensemble_retriever = EnsembleRetriever(
        retrievers=[vector_retriever, bm25_retriever],
        weights=[0.5, 0.5]
    )

    return ensemble_retriever

添加引用链接

在QAChain中添加元数据，返回来源信息。

def ask_with_sources(self, question: str) -> Dict:
    """带来源的问答"""
    docs = self.vector_store.similarity_search(question, k=self.k)
    sources = [doc.metadata.get('source', '') for doc in docs]

    answer = self.ask(question)
    answer['sources'] = sources

    return answer

性能优化

使用缓存避免重复计算。

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

set_llm_cache(InMemoryCache())

# 批量处理文档
def batch_process_documents(file_paths: List[str], batch_size: int = 10):
    """批量处理文档"""
    for i in range(0, len(file_paths), batch_size):
        batch = file_paths[i:i+batch_size]
        # 处理批次...

部署建议

使用Streamlit创建Web界面

# app.py
import streamlit as st
from main import DocumentQASystem

st.set_page_config(page_title="智能文档问答", page_icon="📚")

st.title("📚 智能文档问答系统")

# 侧边栏
with st.sidebar:
    st.header("设置")
    api_key = st.text_input("OpenAI API Key", type="password")
    if api_key:
        os.environ["OPENAI_API_KEY"] = api_key

    uploaded_files = st.file_uploader(
        "上传文档",
        type=["pdf", "txt"],
        accept_multiple_files=True
    )

    if st.button("索引文档"):
        with st.spinner("正在索引..."):
            # 保存文件并索引
            st.success("索引完成！")

# 主界面
if "messages" not in st.session_state:
    st.session_state.messages = []

# 显示对话历史
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# 用户输入
if user_input := st.chat_input("您的问题"):
    st.session_state.messages.append({"role": "user", "content": user_input})
    with st.chat_message("user"):
        st.markdown(user_input)

    # 获取AI回答
    with st.chat_message("assistant"):
        with st.spinner("思考中..."):
            response = system.ask(user_input)
            st.markdown(response)

    st.session_state.messages.append({"role": "assistant", "content": response})

运行：

1	streamlit run app.py

项目总结

我们学习了文档加载、文本分割、向量化、相似度检索、问答生成、记忆管理和系统集成。这些知识综合运用，构建了一个完整的文档问答系统。

项目扩展方向

支持更多文档格式（Word、Excel、PPT）
多用户支持和权限管理
答案评分和反馈
Web界面部署
实时文档更新
多语言支持

全系列总结

恭喜你完成了《从零开始学LangChain》全部教程！

学习回顾

第一章介绍了LangChain的核心价值和环境配置。第二章讲解了Models模型，包括LLM和Chat Model的区别、参数控制、流式输出和结构化输出。第三章深入Prompts提示词，包括提示词工程、模板、少样本提示和输出解析器。第四章介绍了Chains链，学习LCEL表达式语言、顺序链和并行链、条件链以及复杂工作流。第五章讲解了Memory记忆，包括对话记忆管理和不同记忆类型。第六章通过实战项目，学习了完整的文档问答系统架构和部署优化。