基于 LangChain 搭建本地知识库系统

完整代码实现

import os
import getpass
from typing import Any

# 导入 LangChain 相关组件
from langchain_community.vectorstores import Chroma
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.embeddings import QianfanEmbeddingsEndpoint
from langchain_openai import AzureChatOpenAI

# 1. 初始化 Embedding 模型
def init_embedding():
    os.environ["QIANFAN_AK"] = "YOUR_QIANFAN_AK"
    os.environ["QIANFAN_SK"] = "YOUR_QIANFAN_SK"
    return QianfanEmbeddingsEndpoint()

# 2. 初始化 LLM
def init_llm():
    os.environ["AZURE_OPENAI_ENDPOINT"] = "YOUR_ENDPOINT"
    os.environ["AZURE_OPENAI_API_KEY"] = "YOUR_KEY"
    return AzureChatOpenAI(
        azure_deployment="gpt-4",
        openai_api_version="2023-05-15"
    )

# 3. 加载与处理文档
loader = WebBaseLoader("https://docs.dify.ai/v/zh-hans/getting-started/readme")
docs = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
splits = splitter.split_documents(docs)

# 4. 创建向量库
vectorstore = Chroma.from_documents(
    documents=splits,
    collection_name="rag_chroma",
    embedding=init_embedding()
)
retriever = vectorstore.as_retriever()

# 5. 构建 Prompt 模板
template = """Answer the question in Chinese based only on the following context:
{context}

Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)

# 6. 构建 Chain
llm = init_llm()
chain = (
    RunnableParallel({"context": retriever, "question": RunnablePassthrough()})
    | prompt
    | llm
    | StrOutputParser()
)

# 7. 定义输入类型
class Question(BaseModel):
    __root__: str

chain = chain.with_types(input_type=Question)

# 8. 执行查询
if __name__ == "__main__":
    print(chain.invoke("dify 是什么？"))
    print(chain.invoke("dify 能干什么？请用列表形式回答"))
    print(chain.invoke("dify 可以导入哪些数据？"))

启动服务

如果使用 langchain serve 命令启动服务：

langchain serve

启动成功后，访问本地 Playground 地址进行交互测试，例如：http://127.0.0.1:8000/dify/playground/。

注意事项

1. 分片策略：chunk_size 和 chunk_overlap 需根据 Embedding 模型的 Token 限制调整。过大可能导致信息丢失，过小影响语义连贯性。
2. 密钥安全：切勿将 API Key 硬编码提交至代码仓库，建议使用环境变量管理。
3. 向量数据库：本例使用 Chroma，生产环境可考虑 Milvus、Pinecone 等更成熟的方案。

总结

本文详细演示了如何使用 LangChain 结合 RAG 技术构建本地知识库。通过引入外部检索机制，有效解决了大模型知识滞后和幻觉问题，同时保护了私有数据安全。开发者可根据实际需求替换 Embedding 模型和向量数据库，灵活部署不同场景下的智能问答系统。