基于LangChain框架搭建AI问答系统(附源码)

1. 背景知识

在人工智能技术飞速发展的今天，知识库问答系统已成为企业、教育、医疗等领域实现高效信息检索与智能交互的核心工具。传统问答系统往往依赖人工编写的规则或固定模板，难以应对复杂多变的业务场景和海量数据。而基于大语言模型（LLM）的智能问答系统，通过自然语言处理（NLP）技术，能够自动理解用户问题并从知识库中精准检索答案，显著提升了信息处理的效率与用户体验。

LangChain 作为当前最流行的框架之一，为开发者提供了构建端到端AI应用的强大工具链。它通过模块化设计，将大语言模型（如GPT、Llama等）、外部数据源（文档、数据库、API等）和智能代理（Agent）无缝集成，支持快速实现知识库问答、对话生成、自动化任务等复杂功能。相较于从零开发，LangChain大幅降低了技术门槛，使开发者能够专注于业务逻辑而非底层实现。

问答系统：是一种基于人工智能和自然语言处理技术的智能工具，它允许用户通过自然语言提问，并从文档中提取相关信息来生成准确的回答。

无论是希望快速搭建原型的技术爱好者，还是需要为企业定制智能客服的开发者，本项目均能提供从理论到实践的完整指导。接下来，我们将从环境配置开始，逐步实现一个功能完备的AI知识库问答系统。

2. 问答系统流程

本项目旨在基于 Python 和 LangChain框架，构建一个可扩展的知识库问答系统，其核心目标包括：

1. 支持多格式外部数据加载：从PDF、Word、CSV、excel等非结构化数据中提取知识，构建统一的知识库。
2. 结合大语言模型的语义理解能力：通过向量检索（Embedding）和上下文增强技术，实现精准答案生成。

通过本项目，读者将掌握：

• LangChain框架的核心组件（Document Loaders、Embeddings、Vector Stores、Chains）的使用方法；
• 如何将本地或在线文档转化为可查询的知识库(向量库)；
• 如何结合LLM实现语义搜索与答案生成；

本项目模式
支持：支持加载多种类型数据（pdf、word、excel、md、json、yaml、csv、html等）
支持：多种大模型（deepseek、千帆、智普等）

流程图

基于上述流程图涉及到的功能点如下

3. 知识问答系统相关组件

3.1 文档加载器

​ 在 LangChain 和 大语言模型（LLM）应用 中，文档加载器（Loader 的作用是 从 PDF、CSV、txt、excel等 文件中提取文本内容，并将其转换为 LLM 可处理的格式（如字符串、列表或结构化数据），以便后续进行 文本分割（Text Splitting）、嵌入（Embedding）、问答（QA）或总结（Summarization） 等任务。

我们的文档系统支持加载多种数据格式，比如pdf、txt、csv、excel, 这里加载文本后获取的数据有两种格式

1. 一种是返回原始的文本列表 list[Document]
2. 一种是统一转换为文本格式（个人发现：使用非结构化的文本数据，向量库的检索的准确率更高，模型回答效果更好）

下面列举上述文本加载的python代码（代码很多但是超级简单）

• 加载并转换为文档

import os

from PyPDF2 import PdfReader
from langchain_community.document_loaders import TextLoader, CSVLoader, UnstructuredExcelLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

# 文本转换为 list[Document]：对于pdf、txt非结构化数据处理
def textSplitForDocs(content: str) -> list[Document]:
    """
    将文本内容进行切割，返回切割后的文本列表
    使用 langchain_text_splitters 库进行切割
    Args:
        content (str): 待切割的文本内容

    Returns:
        list[str]: 切割后的文本列表
    """
   
    # 初始化递归字符分割器 （每个文档库100字、每个文档块可以允许重复10个字）
    # 重叠字符数：为了分割文字同时不破化其语义完整性
    splitter = RecursiveCharacterTextSplitter(
      chunk_size=100,      # 每个文档块的最大字符数
      chunk_overlap=10     # 相邻文档块之间的重叠字符数
    )
    # 将文本分割成多个 Document 对象
    docs = splitter.create_documents([content])
    return docs  



#PdfReader 加载器 加载pdf文件
def loadPyPdf2Loader(pdf_path: str) -> list[Document]:
    """
    从本地路径加载 PyPDF2 并返回 Document 列表
    使用 PyPDF2 库进行加载
    Args:
        pdf_path (str): PDF 的本地路径（如 "path/to/your/pdf.pdf"）

    Returns:
        list[Document]: 包含文本内容和元数据的 Document 列表
    """
    #入参 pdf文档路径 加载完成
    padReader = PdfReader(pdf_path)
    raw_text = ''
    # 遍历 PDF 的每一页
    for page in padReader.pages:
        #提取文本内存（目前不考虑图片相关资源）
        text = page.extract_text()
        #拼接成长文本
        if text:
            raw_text += text
    #长文本分割并转换为Document列表        
    return textSplitForDocs(raw_text)

#TextLoader 加载器 加载txt文件
def loadTextLoader(text_path: str) -> list[Document]:
    """
    从本地路径加载txt文件
    :param text_path: 文件路径
    :return: 文本信息
    """
    #校验文件路径
    if not os.path.exists(text_path):
        raise FileNotFoundError(f"输入文件不存在: {text_path}")
    #文件路径、编码方式(中文需要指定，否则会乱码)   创建加载器 
    load = TextLoader(text_path, encoding="utf-8")
    # 加载文本文件 直接会返回document列表 文件内容按自然段落或换行符分割成文档列表
    # 不完全符合我们要求，故此处还是拼接文本，在调用方式按照我们要求分隔文本
    docs = load.load()
    
    raw_text = ''
    for doc in docs:
        text = doc.page_content
        if text:
            raw_text += text
     #长文本分割并转换为Document列表         
    return textSplitForDocs(raw_text)
  
#CSVLoader 加载器 加载csv文件（结构化数据直接load 即可获取Document列表）
def loadCSVLoader(cvs_path: str) -> list[Document]:
    """
    加载CSV格式文件
    :param cvs_path:  csv文件路径
    :return: 获取到的文本信息
    """
    csvLoader = CSVLoader(cvs_path, encoding="utf-8")
    docs = csvLoader.load()
    return docs

#UnstructuredExcelLoader 加载器 加载excel文件（结构化数据直接load 即可获取Document列表）
def loadExclLoader(excl_path: str) -> list[Document]:
    """
    加载excel文件
    :param excl_path: 文件路径
    :return: excel中文本信息
    """
    loader = UnstructuredExcelLoader(excl_path, encoding="utf-8")
    docs = loader.load()
    return docs

##### 其他格式文档加载待补充  

• 加载并转换为纯文本格式

#逻辑与上述加载文本转换为文档列表大体一致，主要添加了将数据提取为长文本并分割
import os

from PyPDF2 import PdfReader
from langchain_community.document_loaders import TextLoader, CSVLoader, UnstructuredExcelLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

def loadPyPdf2Loader(pdf_path: str) -> list[str]:
    """
    从本地路径加载 PyPDF2 并返回 字符串结婚 列表
    使用 PyPDF2 库进行加载
    Args:
        pdf_path (str): PDF 的本地路径（如 "path/to/your/pdf.pdf"）

    Returns:
        list[str]: 包含文本内容列表
    """
    padReader = PdfReader(pdf_path)
    raw_text = ''
    for page in padReader.pages:
        text = page.extract_text()
        if text:
            raw_text += text
    return textSplitter(raw_text)

def loadTextLoader(text_path: str) -> list[str]:
    """
    从本地路径加载txt文件
    :param text_path: 文件路径
    :return: 文本信息
    """
    #校验文件路径
    if not os.path.exists(text_path):
        raise FileNotFoundError(f"输入文件不存在: {text_path}")
    load = TextLoader(text_path, encoding="utf-8")
    docs = load.load()
    raw_text = ''
    for doc in docs:
        text = doc.page_content
        if text:
            raw_text += text
    return textSplitter(raw_text)

def loadCSVLoader(cvs_path: str) -> list[str]:
    """
    加载CSV格式文件
    :param cvs_path:  csv文件路径
    :return: 获取到的文本信息
    """
    csvLoader = CSVLoader(cvs_path, encoding="utf-8")
    docs = csvLoader.load()
    raw_text = ''
    for doc in docs:
        text = doc.page_content
        if text:
            raw_text += text
    return textSplitter(raw_text)

def loadExclLoader(excl_path: str) -> list[str]:
    """
    加载excel文件
    :param excl_path: 文件路径
    :return: excel中文本信息
    """
    try:
        loader = UnstructuredExcelLoader(excl_path, encoding="utf-8")
        docs = loader.load()  # 在这里卡住说明是加载器内部问题
    except Exception as e:
        print(f"加载失败: {str(e)}")
        raise
    raw_text = ''
    for doc in docs:
        text = doc.page_content
        if text:
            raw_text += text
    return textSplitter(raw_text)

3.2 文档切割器

​ 在大语言模型（LLM）应用 中，文档切割器（Text Splitter） 的核心作用是 将长文本拆分成更小的、符合模型输入限制的片段（chunks），同时尽可能保留语义完整性。它是连接 原始文档 和 LLM 处理单元 的关键工具，尤其在处理长文档（如 PDF、书籍、报告）时必不可少。

如果上述返回的是纯文本，则需要长文本转换为小的满负荷模型输入限制的片段，文档分割器就派上用场了。

from langchain_text_splitters import RecursiveCharacterTextSplitter

def textSplitter(content: str) -> list[str]:
    """
    将文本内容进行切割，返回切割后的文本列表
    使用 langchain_text_splitters 库进行切割
    Args:
        content (str): 待切割的文本内容

    Returns:
        list[str]: 切割后的文本列表
    """

     splitter = RecursiveCharacterTextSplitter(
      chunk_size=100,      # 每个文档块的最大字符数
      chunk_overlap=10     # 相邻文档块之间的重叠字符数
    )
    # chunk_size:每个文档块的最大字符数  chunk_overlap:相邻文档块之间的重叠字符数
    splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=10)
    # 文档分割
    texts = splitter.split_text(content)
    return texts

3.3 嵌入模型包装器

​ 嵌入模型包装器是自然语言处理（NLP）框架中的核心组件，其核心作用是将 文本数据转换为浮点数值向量（Embedding），从而可以分析文本的相似度。后续将其存储在向量库中，便于后续的搜索使用等。

如下是我整合了千帆、智普、百川三个平台的模型嵌入包装器（为啥没有openAI 因为需要🪜）其他国产大模型的嵌入模型包装器我没有找到对应的demo ,参考文档：langchain嵌入模型包装器

import os
from pathlib import Path

from dotenv import load_dotenv
from langchain_community.embeddings import BaichuanTextEmbeddings, QianfanEmbeddingsEndpoint, ZhipuAIEmbeddings
from langchain_core.embeddings import Embeddings
from langchain_openai import  OpenAIEmbeddings

#支持百川大模型、千帆大模型、智普AI模型
def getEmbeeding(modelName: str) -> Embeddings:
    """
    获取嵌入模型包装器
    Args:
        modelName (str): 嵌入模型名称

    Returns:
        Embeddings: 嵌入模型对象
    """
    #加载配置信息 (安全) 
    #大模型相关key存储在resource/.env 
    BASE_DIR = Path(__file__).parent.parent
    load_dotenv(dotenv_path=BASE_DIR / "resource" / ".env")
    if modelName == "baichuan":
        # 百川大模型
        #1.获取对应的apikey
        bai_chuan_api_key = os.getenv("BAI_CHUAN_API_KEY")
        #2.创建百川嵌入模型包装器
        return BaichuanTextEmbeddings(baichuan_api_key=bai_chuan_api_key)
    elif modelName == "qianfan":
        # 千帆大模型（文心一言）
        #1.获取千帆ak、sk
        qianfan_ak = os.getenv("QIANFAN_AK")
        qianfan_sk = os.getenv("QIANFAN_SK")
        #2.创建千帆的嵌入模型包装器
        return QianfanEmbeddingsEndpoint(
            qianfan_ak=qianfan_ak,
            qianfan_sk=qianfan_sk
        )
    elif modelName == "zhipu":
        # 智普AI模型
        #1. 获取智普apikey
        zhipu_api_key = os.getenv("zhipu-api-key")
        #2. 创建智普的嵌入模型包装器
        return ZhipuAIEmbeddings(
            model="embedding-3",
            api_key=zhipu_api_key
        )
    elif modelName == "openai":
        # openai大模型 TODO 没有open_ai 的key
        return OpenAIEmbeddings()
    else:
        return None

资源添加resource/.env

#百川大模型 apiKey(注册申请)
BAI_CHUAN_API_KEY=sk-XXXXX

#智普AI的api-key
zhipu-api-key=fxxx

#千帆大模型
QIANFAN_AK=xxx
QIANFAN_SK=xxxx

#openai
OPEN_AI_KEY=xxxxx

3.4 向量存储库

向量存储库（Vector Store）是专门用于存储、索引和检索高维向量数据的数据库系统，支持存储向量。从而支持相似性搜索、语义理解能力、多模态支持等。

1. 高效存储高维向量：支持海量高维向量的压缩存储，例如文本、图像、音频等非结构化数据经嵌入模型转换后的向量。
2. 相似性搜索：通过近似最近邻（ANN）算法（如HNSW、IVF-PQ、LSH）快速查找与查询向量最相似的向量集合，而非传统数据库的精确匹配。
3. 语义理解能力：基于向量间的距离度量（如余弦相似度、欧氏距离），实现基于语义的搜索，而非关键词匹配。
4. 多模态支持：可处理文本、图像、音频等多种数据类型的向量表示，支持跨模态检索。
5. 可扩展性：设计用于处理大规模数据，支持分布式部署和水平扩展。

这里我使用的是FAISS向量库，存储嵌入模型包装器（该向量库对于数字类型查询不太理想，准确度较低，对于文字相关搜索精确度挺高，后续看演示吧）

保存向量库数据

def saveFAISSVector(indexPath: str) -> None:
    """
    加载外部数据并将其存储到向量库中
    :param indexPath: 向量库路径
    """
    # 加载pdf文件
    pdf_texts = loadPyPdf2Loader("doc_data/member.pdf")

    # 加载文本文件
    txt_texts = loadTextLoader("doc_data/crm_info.txt")

    # 加载CSV格式文件
    csv_texts = loadCSVLoader("doc_data/cust_info.csv")

    # 加载Excel格式文件
    excel_texts = loadExclLoader("doc_data/dms.xlsx")

    # 3. 合并 PDF 和 TXT 的文档列表
    texts = pdf_texts + txt_texts + csv_texts + excel_texts

    # 3.使用嵌入模型包装器，将文档转换嵌入向量
    embeddings = getEmbeeding(embeeding_model)
    # vectors = embeddings.embed_documents(texts)

    # 4. 创建向量数据库并与嵌入模型连接，
    # 将文本转换为向量，并保存到向量数据库中
    docSearch = FAISS.from_texts(texts, embeddings)
    # 持久化向量库
    docSearch.save_local(indexPath)
    print("向量库保存成功！")

加载向量库 用于查询使用

def loadFAISSVector(indexPath: str) -> FAISS:
    """
    加载 FAISS 向量库
    :param index: 向量库路径
    :return: FAISS 对象
    """
    return FAISS.load_local(indexPath, getEmbeeding(embeeding_model),allow_dangerous_deserialization=True)

文档检索

我们从向量库中加载相关向量后，需要从向量库中进行检索，并将检索出来的数据(document 列表)作为背景信息传递给大模型

 # 5.查询数据从向量库
 # query是自然语言查询条件
 docs = docSearch.similarity_search(query, k=6)
 print("命中的文档数：", len(docs))

3.5 模型包装器

模型包装器（Model Wrappers）是 LangChain 的核心组件之一，其作用是为大语言模型提供一层抽象接口，封装模型的调用、输入输出处理以及配置管理等功能。

#目前支持 deepseek、qianfan、zhipu
def getLLM(modelName: str) -> BaseChatModel:
    """
    获取大模型对象
    Args:
        modelName (str): 模型名称
    Returns:
        BaseChatModel: 模型对象
    """
    #加载配置信息 (安全)
    BASE_DIR = Path(__file__).parent.parent.parent
    load_dotenv(dotenv_path=BASE_DIR / "resource" / ".env")
    if modelName == "deepseek":
        # deepseek大模型
        api_key = os.getenv("DEEPSEEK_API_KEY")
        return ChatDeepSeek(
            model="deepseek-chat",
            temperature=0,
            max_tokens=None,
            timeout=None,
            max_retries=2,
            openai_api_key=api_key,
        )
    elif modelName == "qianfan":
        # 千帆大模型
        app_id = os.getenv("QIANFAN_APPID")
        access_token = os.getenv("QIANFAN_TOKEN")
        return QianfanChatEndpoint(
            model="ERNIE-Tiny-8K",
            temperature=0.2,
            timeout=30,
            app_id=app_id,
            access_token=access_token
        )
    elif modelName == "zhipu":
        # 智普AI模型
        api_key = os.getenv("zhipu-api-key")
        # 初始化一个 ChatOpenAI 对象（即一个 LLM 模型接口）
        return ChatOpenAI(
            temperature=0.95,  # 控制生成文本的随机性（0=确定性强，1=随机性强）
            model="glm-4-air-250414",  # 指定模型名称（这里是智谱 GLM-4 的某个版本）
            openai_api_key=api_key,  # 你的 API 密钥（需提前定义变量 api_key）
            openai_api_base="https://open.bigmodel.cn/api/paas/v4/"  # 智谱 AI 提供的 API 基础地址
        )
    elif modelName == "openai":
        # openai大模型 TODO 没有open_ai 的key
        return ChatOpenAI()
    else:
        return None

3.6 链组件

​ 在 LangChain 中，链（Chain） 是核心抽象概念之一，它代表一个将多个组件（如大语言模型LLM、提示词模板prompt、工具调用toool、其他链等）按特定逻辑组合起来的可执行流程。链的设计使得复杂任务可以分解为可复用的模块，并通过组合这些模块实现灵活的功能扩展。下面列举几个常用链

• 基础链（Basic Chains）

简单任务的基础组件。

• 2. RAG 链（Retrieval-Augmented Generation Chains）

通过检索实时更新的外部知识库（如新闻、数据库、API），将最新信息作为上下文输入大模型，生成准确回答。

• 3. 代理链（Agent Chains）

通过工具扩展 LLM 能力，实现自动化。

• **4. 高级推理链（Advanced Reasoning Chains） **

针对特定场景优化（如知识图谱、辩论）。

• 多模态链（Multimodal Chains）

支持非文本数据（图像、音频）与 LLM 交互。

我们的问答系统，需要将向量库使用RAG代理增强，对文本进行代理增强。实现了stuff链(文档问答链）代码如下

stuff链(文档问答链)

def qa_stuff(docs: list[Document], query: str):
    """
    创建一个基于 Stuff 策略的问答链，将多个文档合并后传给 LLM 回答用户问题。
    
    Args:
        docs (list[Document]): 文档列表，每个文档包含文本内容和元数据。
        query (str): 用户提出的查询问题。
    
    Returns:
        str: LLM 生成的回答结果。
    """
    
    # 1. 定义系统提示模板（System Prompt）
    #    - 指导模型如何利用提供的文档回答问题
    #    - 强调“不知道就说明不知道”，避免编造答案
    system_template = "使用以下背景信息来回答用户的问题。如果你不知道答案，只需说明不知道，不要编造答案。{context}"
    
    # 2. 构建完整的提示模板（Prompt Template）
    #    - 包含系统消息（System Message）和用户消息（Human Message）
    messages = [
        # 系统消息：设置模型行为规则
        SystemMessagePromptTemplate.from_template(system_template),
        # 用户消息：包含实际查询问题
        HumanMessagePromptTemplate.from_template("{question}"),
    ]
    
    # 3. 创建聊天提示模板（ChatPromptTemplate）
    #    - 将消息列表转换为 LangChain 可识别的提示格式
    chat_prompt = ChatPromptTemplate.from_messages(messages)
    
    # 4. 创建 Stuff 文档链
    #    - 使用自定义的聊天提示模板
    #    - 指定 LLM 模型（这里调用 getLLM 函数获取 "deepseek" 模型）
    chain = create_stuff_documents_chain(
        llm=getLLM("deepseek"),  # 获取 LLM 模型实例
        prompt=chat_prompt       # 使用上面定义的聊天提示模板
    )
    
    # 5. 执行链式调用
    #    - 输入参数：
    #       - context: 要合并的文档列表（会自动拼接）
    #       - question: 用户查询问题
    result = chain.invoke({
        "context": docs,  # 文档上下文（会自动转换为字符串）
        "question": query # 用户问题
    })
    
    # 6. 返回结果
    return result

至此，langchain的所有核心组件讲完，同时我们问答系统的所有流程也编写完成！下面让我们进入演示环节。

4. 问答系统演示

4.1 问答程序

调用我们上面章节的相关代码，我们便编写完了一个基于知识库的AI智能问答系统。流程如下

def qa_system(query: str) -> None:
    """
    基于向量检索和LLM的问答系统主流程：
    1. 将数据保存到FAISS向量库（若不存在则创建）
    2. 从本地加载向量库
    3. 执行相似度检索获取相关文档
    4. 使用Stuff链将文档和问题传给LLM生成答案
    
    Args:
        query (str): 用户输入的自然语言查询问题
        
    Returns:
        None (结果直接打印，如需返回可修改为return result)
    """
    
    # 1. 保存/更新数据到FAISS向量库
    #    - 如果向量库已存在则直接使用，不存在则创建新索引
    #    - faiss_index_name 是全局常量，指定索引存储路径
    saveFAISSVector(faiss_index_name)
    
    # 2. 从本地加载FAISS向量库
    #    - 返回的docSearch是已初始化的向量检索器
    #    - 内部包含嵌入模型(embedding model)和FAISS索引
    docSearch = loadFAISSVector(faiss_index_name)
    
    # 3. 执行相似度检索
    #    - 参数说明：
    #       - query: 用户查询文本
    #       - k=6: 返回最相似的6个文档（可根据需求调整）
    #    - 返回的docs是Document对象列表，包含文本内容和元数据
    docs = docSearch.similarity_search(query, k=6)
    print("命中的文档数：", len(docs))  # 调试信息：实际检索到的文档数量
    
    # 4. 调用Stuff链处理文档并生成答案
    #    - qa_stuff函数实现：
    #      1) 将所有文档内容合并为单个字符串
    #      2) 添加系统提示词指导LLM回答
    #      3) 调用LLM生成最终答案
    result = qa_stuff(docs, query)
    
    # 5. 输出最终结果
    print("最终答案：", result)
    
#启动程序
if __name__ == "__main__":
    #用户输入提问
    query = input("你想要问些什么？")
    #问答
    qa_system(query)

4.2 演示大模型回答效果

知识库和向量库

图中是我的建立的知识库以及向量库。

pdf_qa_system.py 是将所有数据转换为文档存入向量 (针对数字类型的问答不准确)

pdf_qa_system_text.py 是将所有数据转换为长文本分割成多个文档存入向量 （各个维度查询准确度都很高）

命中知识库的回答：

​ 首先在【3.1 文档加载器】加载了不同类型的文档知识，例如加载了：csv格式的客户数据：

可以针对上述数据任何维度进行问答。问答效果如下：

不命中的效果

5.问答系统代码

链接: 代码原文件地址

项目目录结构
lang_chain_qa/
└── doc_data/ 知识库资源 （目前是空自行补充）
├── faiss_index/ 向量库持久化文件（文档类型）
├── faiss_text_index/ 向量库持久化文件（文本类型）
├── crm_info.txt 资源
├── cust_info.csv 资源
├── dms.xlsx 资源
├── mathPix.md 资源
├── member.pdf 资源
├── pdf_qa_system.py 文档类型数据知识问答
├── pdf_qa_system_text.py 文本类型数据知识问答
└── rerank_prompt.py

至此，我们一步一步的手写了一份知识库问答系统，如有问题欢迎指正。原创不易，对你有帮助还请一键三连。