我把100份公司文档喂给AI,它比老员工还懂业务——用LangChain搭本地知识库全过程

公司知识库：新人问制度流程，AI秒回+附文档出处

客服系统：用户问产品细节，AI从产品手册里找答案

个人学习：把你的笔记全扔进去，变成一个会回答问题的"第二大脑"

pip install langchain langchain-openai langchain-community

pip install chromadb     # 向量数据库（本地运行，不需要安装服务）

pip install pypdf        # 读取PDF

pip install sentence-transformers  # 本地Embedding模型（省钱方案）

from langchain_community.document_loaders import (

    PyPDFLoader,    # PDF

    TextLoader,     # TXT

    DirectoryLoader # 批量加载整个文件夹

)

# 加载单个PDF

loader = PyPDFLoader("公司手册.pdf")

docs = loader.load()

print(f"加载了 {len(docs)} 页")

# 输出示例：加载了 42 页

# 批量加载一整个文件夹的txt

loader = DirectoryLoader(

    "./公司文档",

    glob="**/*.txt",           # 匹配所有txt

    loader_cls=TextLoader,

    loader_kwargs={"encoding": "utf-8"}

)

docs = loader.load()

print(f"共加载 {len(docs)} 个文档")

PDF里的表格、图片不会自动提取。如果需要表格数据，用pdfplumber。

中文编码问题：Windows下TextLoader默认用utf-8，如果你的文件是GBK编码，会报错。用encoding="gbk"或转成utf-8。

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(

    chunk_size=500,       # 每段大约500个字符

    chunk_overlap=50,     # 前后重叠50个字符（防止信息被切断）

    separators=["\n\n", "\n", "。", ".", " "]  # 按段落>句子>空格切

)

chunks = text_splitter.split_documents(docs)

print(f"分割成 {len(chunks)} 个片段")

# 输出示例：分割成 387 个片段

chunk_size不是越大越好。太大会引入噪声（不相关内容），太小会丢失上下文。500-1000是经验值。

如果文档是FAQ格式（一问一答），用RecursiveCharacterTextSplitter效果差，应该按Q&A对切分。

# 方案A：用OpenAI的Embedding（效果好，要花钱）

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()

# 方案B：用本地模型（免费，不需要联网，效果够用）

from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(

    model_name="shibing624/text2vec-base-chinese"  # 中文效果不错

)

# 创建向量数据库

from langchain_community.vectorstores import Chroma

vectorstore = Chroma.from_documents(

    documents=chunks,

    embedding=embeddings,

    persist_directory="./my_knowledge_base"  # 本地存储路径

)

vectorstore.persist()  # 保存到磁盘

print("向量数据库构建完成！")

第一次用本地模型会下载模型文件（约400MB），需要等一会儿。

chromadb的数据库文件不要手动删除或移动，会损坏。

from langchain_openai import ChatOpenAI

from langchain.chains import RetrievalQA

from langchain.prompts import PromptTemplate

import os

# 初始化大模型

llm = ChatOpenAI(

    model="deepseek-chat",  # 用便宜的模型就够

    api_key=os.getenv("OPENAI_API_KEY"),

    base_url="https://api.deepseek.com/v1",

    temperature=0  # 问答要准确，温度设0

)

# 自定义Prompt——这一步决定回答质量

prompt_template = """基于以下已知信息回答用户问题。

如果已知信息中没有答案，请回答"根据现有资料，我无法回答这个问题"。

不要编造已知信息中没有的内容。

已知信息：

{context}

用户问题：{question}

回答："""

PROMPT = PromptTemplate(

    template=prompt_template,

    input_variables=["context", "question"]

)

# 构建检索问答链

qa_chain = RetrievalQA.from_chain_type(

    llm=llm,

    chain_type="stuff",           # 把所有检索结果一起塞给AI

    retriever=vectorstore.as_retriever(

        search_kwargs={"k": 4}    # 检索最相关的4个片段

    ),

    return_source_documents=True,  # 返回引用的原始文档

    chain_type_kwargs={"prompt": PROMPT}

)

# 提问！

result = qa_chain.invoke({"query": "员工请假流程是什么？"})

print(f"回答：{result['result']}")

print(f"\n参考来源：")

for i, doc in enumerate(result['source_documents'], 1):

    source = doc.metadata.get('source', '未知')

    print(f"  {i}. {source}")

import os

from pathlib import Path

from langchain_community.document_loaders import DirectoryLoader, TextLoader, PyPDFLoader

from langchain.text_splitter import RecursiveCharacterTextSplitter

from langchain_community.vectorstores import Chroma

from langchain_openai import ChatOpenAI

from langchain.chains import RetrievalQA

from langchain.prompts import PromptTemplate

class LocalKnowledgeBase:

    """本地知识库——可复用工具

    用法：

        kb = LocalKnowledgeBase(docs_dir="./公司文档", db_dir="./kb_data")

        kb.build()           # 第一次：构建索引

        kb.load()            # 之后：直接加载已有索引

        answer = kb.ask("你的问题")

    """

    def __init__(self, docs_dir="./docs", db_dir="./kb_data"):

        self.docs_dir = docs_dir

        self.db_dir = db_dir

        self.llm = ChatOpenAI(

            model="deepseek-chat",

            api_key=os.getenv("OPENAI_API_KEY"),

            base_url="https://api.deepseek.com/v1",

            temperature=0

        )

        self.embeddings = None

        self.vectorstore = None

    def build(self):

        """从文档构建知识库索引"""

        # 1. 加载文档

        all_docs = []

        # 加载txt文件

        txt_loader = DirectoryLoader(

            self.docs_dir, glob="**/*.txt",

            loader_cls=TextLoader,

            loader_kwargs={"encoding": "utf-8"}

        )

        all_docs.extend(txt_loader.load())

        # 加载PDF文件

        for pdf_file in Path(self.docs_dir).rglob("*.pdf"):

            loader = PyPDFLoader(str(pdf_file))

            all_docs.extend(loader.load())

        if not all_docs:

            raise FileNotFoundError(f"在 {self.docs_dir} 中没有找到文档！")

        print(f"✓ 加载了 {len(all_docs)} 个文档/页面")

        # 2. 切块

        splitter = RecursiveCharacterTextSplitter(

            chunk_size=500,

            chunk_overlap=50,

            separators=["\n\n", "\n", "。", ".", " "]

        )

        chunks = splitter.split_documents(all_docs)

        print(f"✓ 分割成 {len(chunks)} 个片段")

        # 3. 向量化并存入数据库

        # 使用本地Embedding（省钱）

        from langchain_community.embeddings import HuggingFaceEmbeddings

        self.embeddings = HuggingFaceEmbeddings(

            model_name="shibing624/text2vec-base-chinese"

        )

        self.vectorstore = Chroma.from_documents(

            documents=chunks,

            embedding=self.embeddings,

            persist_directory=self.db_dir

        )

        self.vectorstore.persist()

        print(f"✓ 知识库构建完成！数据保存在 {self.db_dir}")

    def load(self):

        """加载已有的知识库索引"""

        from langchain_community.embeddings import HuggingFaceEmbeddings

        from langchain_community.vectorstores import Chroma

        self.embeddings = HuggingFaceEmbeddings(

            model_name="shibing624/text2vec-base-chinese"

        )

        self.vectorstore = Chroma(

            persist_directory=self.db_dir,

            embedding_function=self.embeddings

        )

        print("✓ 知识库加载完成")

    def ask(self, question, k=4):

        """向知识库提问，返回答案和来源"""

        if not self.vectorstore:

            raise RuntimeError("请先调用 build() 或 load() 初始化知识库")

        prompt_template = (

            "基于以下已知信息回答用户问题。\n"

            "如果已知信息中没有答案，请回答\"根据现有资料，我无法回答这个问题\"。\n"

            "不要编造已知信息中没有的内容。\n\n"

            "已知信息：\n{context}\n\n"

            "用户问题：{question}\n\n回答："

        )

        PROMPT = PromptTemplate(

            template=prompt_template,

            input_variables=["context", "question"]

        )

        qa = RetrievalQA.from_chain_type(

            llm=self.llm,

            chain_type="stuff",

            retriever=self.vectorstore.as_retriever(

                search_kwargs={"k": k}

            ),

            return_source_documents=True,

            chain_type_kwargs={"prompt": PROMPT}

        )

        result = qa.invoke({"query": question})

        sources = []

        for doc in result.get('source_documents', []):

            sources.append(doc.metadata.get('source', '未知来源'))

        return {

            "answer": result['result'],

            "sources": list(set(sources))  # 去重

        }

# ===== 使用示例 =====

if __name__ == "__main__":

    kb = LocalKnowledgeBase(docs_dir="./公司文档", db_dir="./kb_data")

    # 首次使用：构建索引（只需要一次）

    kb.build()

    # 之后使用：加载索引（秒级）

    # kb.load()

    # 提问

    result = kb.ask("员工请假流程是什么？")

    print(f"回答：{result['answer']}")

    print(f"来源：{result['sources']}")

from langchain.memory import ConversationBufferMemory

from langchain.chains import ConversationalRetrievalChain

def create_conversational_kb(vectorstore, llm):

    """带对话记忆的知识库"""

    memory = ConversationBufferMemory(

        memory_key="chat_history",

        return_messages=True,

        output_key="answer"

    )

    qa = ConversationalRetrievalChain.from_llm(

        llm=llm,

        retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),

        memory=memory,

        return_source_documents=True,

        output_key="answer"

    )

    return qa

# 使用

# qa = create_conversational_kb(kb.vectorstore, kb.llm)

# 

# result = qa.invoke({"question": "员工请假流程是什么？"})

# print(result["answer"])

# 

# # 可以追问！AI会记住上一轮的上下文

# result = qa.invoke({"question": "那病假呢？"})

# print(result["answer"])  # AI知道你在问"病假的请假流程"

对话记忆存在内存里，程序重启就没了。生产环境要把chat_history存到数据库。

ConversationBufferMemory会无限增长，对话轮数多了会超出token限制。商业项目要用ConversationSummaryMemory（自动总结历史对话）。

把所有文档放到./公司文档文件夹

运行kb.build()，等待约30秒

部署成一个简单的Web界面（用Gradio，5行代码）

import gradio as gr

kb = LocalKnowledgeBase(docs_dir="./公司文档")

kb.load()

def answer(question):

    result = kb.ask(question)

    answer = result["answer"]

    sources = "\n".join(f"- {s}" for s in result["sources"])

    return f"{answer}\n\n---\n参考来源：\n{sources}"

demo = gr.Interface(fn=answer, inputs="text", outputs="markdown")

demo.launch(share=True)  # share=True生成公网链接

技术文档里代码块很多，RecursiveCharacterTextSplitter会把代码块切碎。解决方案：用MarkdownTextSplitter按标题层级切分。

代码片段的Embedding效果差（语义不像自然语言），建议在代码块前面加一句自然语言描述。

[ ] 准备3-5个文档（txt或PDF），放到一个文件夹

[ ] 安装依赖：pip install langchain langchain-community chromadb sentence-transformers

[ ] 运行本文的LocalKnowledgeBase代码，构建你的第一个知识库

[ ] 试着问5个你的文档里有的问题，看看回答质量

[ ] 调整chunk_size（300/500/1000），对比回答质量差异

import pdfplumber

def extract_tables_from_pdf(pdf_path):

    """从PDF中提取表格，返回结构化数据"""

    tables_data = []

    try:

        with pdfplumber.open(pdf_path) as pdf:

            for page_num, page in enumerate(pdf.pages, 1):

                tables = page.extract_tables()

                for table in tables:

                    if table:

                        # 把表格转成文本格式，方便AI理解

                        for row in table:

                            clean_row = [str(cell).strip() if cell else "" for cell in row]

                            tables_data.append(" | ".join(clean_row))

    except Exception as e:

        print(f"提取表格失败（第{page_num}页）：{e}")

    return tables_data

import shutil

import os

def reset_knowledge_base(db_dir):

    """安全地重置知识库"""

    if os.path.exists(db_dir):

        try:

            shutil.rmtree(db_dir)

            print(f"✓ 已删除旧数据库：{db_dir}")

        except PermissionError:

            print("✗ 数据库文件被占用，请先关闭所有使用它的程序")

            raise

from langchain.memory import ConversationSummaryMemory

# 替换之前的ConversationBufferMemory

memory = ConversationSummaryMemory(

    llm=llm,  # 需要一个LLM来做总结

    memory_key="chat_history",

    return_messages=True,

    output_key="answer"

)

def read_file_auto_encoding(filepath):

    """自动检测编码读取文件"""

    encodings = ["utf-8-sig", "utf-8", "gbk", "gb2312", "gb18030"]

    for enc in encodings:

        try:

            with open(filepath, "r", encoding=enc) as f:

                content = f.read()

            return content

        except (UnicodeDecodeError, UnicodeError):

            continue

    raise ValueError(f"无法识别文件编码：{filepath}")

import time

def ask_with_retry(qa_chain, question, max_retries=3):

    """带重试的问答"""

    for attempt in range(max_retries):

        try:

            result = qa_chain.invoke({"query": question})

            return result

        except Exception as e:

            if attempt < max_retries - 1:

                wait = 2 ** attempt  # 指数退避：1秒、2秒、4秒

                print(f"第{attempt+1}次失败，{wait}秒后重试...")

                time.sleep(wait)

            else:

                return {"result": "抱歉，服务暂时不可用，请稍后重试。", 

                        "source_documents": []}