LangChain与DeepSeek的完美结合：从零开始构建一个智能问答系统

最近在折腾AI应用开发的朋友，大概都绕不开一个词：LangChain。这个框架就像是为大语言模型（LLM）应用开发量身定制的“乐高积木”，把那些复杂的组件——模型调用、记忆管理、外部工具集成、向量检索——都变成了标准化的接口。而DeepSeek，作为国内顶尖的大模型之一，以其出色的推理能力和极具竞争力的API价格，成为了许多开发者在构建实际应用时的首选。当LangChain的模块化设计遇上DeepSeek的强劲性能，我们就能以一种高效、优雅的方式，搭建起从原型到生产的智能问答系统。

这篇文章，我想和你分享的，不是简单的API调用教程，而是一套完整的构建思路和实战经验。无论你是想为企业内部搭建一个高效的知识库助手，还是为客服系统注入AI灵魂，甚至是为自己的产品增加一个智能交互入口，这个过程都极具参考价值。我们会从最基础的环境搭建开始，一步步深入到提示工程、向量检索、流式响应以及生产级部署的考量，过程中会穿插我实际踩过的“坑”和验证过的“最佳实践”。让我们开始吧。

1. 环境准备与核心概念理解

在动手写第一行代码之前，花点时间理清核心概念和准备好工具，能让你后续的开发过程顺畅数倍。LangChain的生态在快速迭代，确保你使用的是稳定且兼容的版本至关重要。

我强烈建议使用虚拟环境来管理你的项目依赖，这能避免不同项目间的包版本冲突。下面是我在多个项目中验证过的稳定依赖组合：

# 创建并激活虚拟环境（以conda为例）
conda create -n langchain-deepseek python=3.10
conda activate langchain-deepseek

# 安装核心依赖
pip install langchain==0.1.0
pip install langchain-core==0.1.0
pip install langchain-community==0.0.10
pip install langchain-openai==0.0.5

注意：langchain-openai这个包的名字可能会让人误解。它实际上是一个实现了OpenAI API兼容接口的客户端库。这意味着，任何遵循OpenAI API格式的模型服务（比如DeepSeek、通义千问等）都可以通过这个库来调用，这是LangChain设计上的一个巧妙之处。

除了LangChain本身，我们还需要处理向量嵌入和可能的数据库连接。一个基础的智能问答系统至少需要以下组件：

• 大语言模型接口：用于调用DeepSeek。
• 文本嵌入模型：将文档和问题转换为向量，用于语义检索。
• 向量数据库：存储和快速检索这些向量。
• 应用框架：用于构建Web服务，比如FastAPI。

因此，我们继续安装：

pip install openai==1.3.0  # 用于嵌入模型调用，同样兼容DeepSeek
pip install chromadb==0.4.18  # 一个轻量且功能强大的本地向量数据库
pip install tiktoken  # 用于文本分词和计数
pip install python-dotenv  # 管理环境变量
pip install fastapi uvicorn  # 用于构建API服务

安装完成后，建议你创建一个.env文件来管理敏感信息，比如API密钥。永远不要将密钥硬编码在代码中。

# .env 文件内容示例
DEEPSEEK_API_KEY=your_deepseek_api_key_here

1.1 LangChain的核心抽象：为什么是“链”？

LangChain的核心思想是“链”（Chain）。你可以把它理解为一个数据处理管道。一个最简单的链，就是把用户的输入（Input）通过一个提示词模板（PromptTemplate）格式化，然后送给大模型（LLM），最后输出结果（Output）。

但它的强大之处在于，这个链可以无限组合和扩展。比如：

• 可以在调用LLM前，先从一个向量数据库中检索出相关的文档片段，并把它们插入到提示词中（这就是检索增强生成，RAG）。
• 可以在链中加入一个“记忆”组件，让它记住之前的对话历史。
• 可以让链在得到模型输出后，自动调用一个计算器工具来验证结果。

这种模块化设计，让开发者可以像搭积木一样，构建出非常复杂的AI应用逻辑，而每一块“积木”都相对独立、可测试、可替换。理解这一点，是高效使用LangChain的关键。

2. 第一步：让DeepSeek在LangChain中跑起来

万事开头难，但让DeepSeek接入LangChain却异常简单。这得益于langchain-openai库对OpenAI API标准的兼容。我们不需要为DeepSeek写任何特殊的适配器。

首先，在代码中加载环境变量，并初始化我们的DeepSeek模型客户端。

# main.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

# 加载 .env 文件中的环境变量
load_dotenv()

# 初始化DeepSeek模型
llm = ChatOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),  # 从环境变量读取密钥
    base_url="https://api.deepseek.com/v1",  # DeepSeek的API端点
    model="deepseek-chat",  # 指定模型，也可以是 deepseek-coder
    temperature=0.1,  # 控制创造性，问答系统建议较低值以保证稳定性
    streaming=True,  # 启用流式输出，为后续的实时响应做准备
    timeout=30.0,  # 设置超时时间
)

这里有几个参数值得深入聊聊：

• base_url：指向DeepSeek的API服务器。确保末尾有/v1，这是OpenAI API的版本路径。
• model：deepseek-chat是通用的对话模型，如果你的场景更偏向代码生成或逻辑推理，可以尝试deepseek-coder。
• temperature：这个参数控制输出的随机性。范围在0到2之间。值越低（如0.1），输出越确定、保守；值越高，输出越有创造性、不可预测。对于事实性强的问答系统，我通常设置在0.1到0.3之间。
• streaming=True：这是实现“打字机效果”流式响应的关键。我们稍后会详细实现它。

现在，我们可以构建第一个最简单的链了——一个直接回答问题的链。

from langchain_core.prompts import ChatPromptTemplate

# 创建一个简单的提示词模板
prompt = ChatPromptTemplate.from_template("请回答以下问题：{question}")

# 使用管道操作符 ‘|’ 将模板和模型组合成链
# 这行代码的含义是：数据先经过prompt处理，再送给llm处理
basic_chain = prompt | llm

# 调用链
question = "LangChain是什么？"
response = basic_chain.invoke({"question": question})
print(response.content)

如果一切顺利，你将看到DeepSeek返回的关于LangChain的解释。恭喜，你已经完成了最基础的集成！但这只是一个开始，一个真正的智能问答系统远不止于此。

3. 构建系统记忆：让对话拥有上下文

一个只会回答单轮问题的系统是“健忘”的。在实际的客服或对话场景中，用户经常会说“你刚才说的那个功能”、“再详细一点”之类的话。这就需要系统能记住之前的对话历史。

LangChain提供了多种记忆（Memory）组件。最常用的是ConversationBufferMemory，它就像一个简单的聊天记录本，保存着完整的对话历史。

from langchain.memory import ConversationBufferMemory
from langchain_core.runnables import RunnablePassthrough

# 初始化记忆存储，设定最多记住最近10轮对话
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True, max_token_limit=2000)

# 构建一个更复杂的提示词模板，其中包含“历史”变量
conversation_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业、友善的AI助手。请根据对话历史来回答用户的问题。如果历史中不包含相关信息，请直接根据你的知识回答。"),
    ("placeholder", "{chat_history}"),  # 这里将自动填入历史消息
    ("human", "{input}")
])

# 构建带记忆的链需要多一步：从memory中加载历史，并组合到输入中
def load_history(_):
    # 从memory组件中加载历史记录
    history = memory.load_memory_variables({})
    return history

conversation_chain = (
    RunnablePassthrough.assign(chat_history=load_history)  # 将加载的历史赋值给`chat_history`变量
    | conversation_prompt
    | llm
)

# 模拟多轮对话
user_inputs = [
    "我叫张三。",
    "我的名字是什么？",
    "LangChain适合用来做什么？"
]

for inp in user_inputs:
    print(f"用户: {inp}")
    response = conversation_chain.invoke({"input": inp})
    print(f"助手: {response.content}\n")
    # 非常重要：将本轮对话存入记忆
    memory.save_context({"input": inp}, {"output": response.content})

运行这段代码，你会发现系统在第二轮成功回答出了你的名字。这就是记忆在起作用。ConversationBufferMemory会将对话以HumanMessage和AIMessage的形式存储在内存中。在实际应用中，你可能需要更复杂的记忆管理，比如只记住关键信息（ConversationSummaryMemory），或者将记忆存储到数据库中以便持久化。

4. 从“知道”到“懂得”：集成向量数据库实现RAG

一个仅依赖模型自身知识的问答系统，其答案受限于模型的训练数据，且无法获取最新的、私有的或非常具体的信息。检索增强生成（RAG） 技术解决了这个问题。它的核心思想是：在回答之前，先从你自己的知识库（文档、手册、数据库）中检索出最相关的信息，然后将这些信息作为上下文提供给模型，让模型基于此生成答案。

实现RAG需要三个核心步骤：文档处理、向量检索、答案生成。

4.1 文档加载与分块

首先，你需要将你的知识库文档（如PDF、Word、Markdown、网页）加载进来，并切割成适合检索的小片段。

from langchain_community.document_loaders import TextLoader, PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 1. 加载文档（以PDF为例）
loader = PyPDFLoader("path/to/your/product_manual.pdf")
documents = loader.load()

# 2. 文本分块
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,  # 每个块的最大字符数
    chunk_overlap=50,  # 块之间的重叠字符，避免语义被切断
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]  # 分割符优先级
)
chunks = text_splitter.split_documents(documents)
print(f"原始文档被分割成了 {len(chunks)} 个文本块。")

提示：chunk_size的选择是个权衡。太小会丢失上下文，太大会降低检索精度并增加模型处理负担。对于通用问答，500-1000是个不错的起点。chunk_overlap能有效防止一个完整的句子或概念被拦腰截断。

4.2 向量化与存储

接下来，我们需要一个嵌入模型（Embedding Model）将文本块转换成向量，并存入向量数据库。这里我们依然使用与DeepSeek API兼容的服务。

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 初始化嵌入模型
# 注意：虽然类名是OpenAIEmbeddings，但通过base_url可以指向DeepSeek（如果其提供嵌入服务）
# 目前DeepSeek主要提供LLM服务，嵌入模型我们可以先用OpenAI的或开源的。
# 这里假设我们使用一个开源的嵌入模型，例如通过`langchain_community`集成。
# 为了演示，我们使用一个本地模拟或兼容API。实际生产需替换为真实嵌入API。
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",  # 使用一个较小的嵌入模型
    # 如果使用其他服务，同样可以配置 base_url 和 api_key
)

# 创建向量数据库，并将文本块存入
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 指定持久化目录，否则数据只在内存中
)
print("向量数据库创建并持久化完成。")

Chroma是一个轻量级、易用的本地向量数据库，非常适合原型开发和中小规模应用。它会把向量数据持久化到本地磁盘的./chroma_db目录。对于更大规模或分布式场景，可以考虑Weaviate、Pinecone或Qdrant。

4.3 构建RAG链

现在，我们将检索器、提示词模板和LLM组合起来，形成一个完整的RAG链。

from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 首先，将向量数据库转换为一个检索器（Retriever）
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})  # 每次检索返回最相关的3个片段

# 定义一个函数，用于格式化检索到的文档
def format_docs(docs):
    return "\n\n".join([doc.page_content for doc in docs])

# 构建RAG提示词模板
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个专业的问答助手。请严格根据以下提供的上下文信息来回答问题。如果上下文信息不足以回答问题，请直接说“根据提供的信息，我无法回答这个问题”，不要编造信息。

上下文信息：
{context}

问题：{question}
请根据上下文给出答案："""),
])

# 构建RAG链
# 这个链的数据流是：输入 -> 检索相关文档 -> 格式化文档 -> 填入提示词 -> 调用LLM -> 输出答案
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()  # 将LLM的输出解析为纯字符串
)

# 测试RAG链
question = "你们的产品支持哪些支付方式？"
answer = rag_chain.invoke(question)
print(f"问题：{question}")
print(f"答案：{answer}")

这个链的工作流程非常清晰：

1. 用户输入一个问题。
2. retriever从向量数据库中检索出最相关的几个文档片段。
3. format_docs函数将这些片段合并成一个字符串。
4. rag_prompt模板将合并后的上下文（context）和原始问题（question）组装成最终的提示词。
5. LLM根据这个富含上下文的提示词生成答案。
6. StrOutputParser提取出答案文本。

至此，一个具备私有知识库查询能力的智能问答核心就构建完成了。它的答案质量，很大程度上取决于你的文档质量、分块策略以及提示词的设计。

5. 提升体验：实现流式输出与异步处理

当答案较长时，让用户等待整个响应生成完毕再显示，体验会很差。流式输出（Streaming）可以像真人打字一样，逐词逐句地返回结果，极大地提升了交互感和响应感知。

我们在初始化llm时已经设置了streaming=True，现在来实现它。同时，考虑到Web服务的并发需求，我们使用异步（Async）调用。

# streaming_async.py
import asyncio
from langchain_core.output_parsers import StrOutputParser

# 构建一个简单的流式链（这里以基础的问答链为例，RAG链同理）
streaming_chain = prompt | llm | StrOutputParser()

# 异步流式调用
async def stream_response(question: str):
    print(f"用户: {question}")
    print("助手: ", end="", flush=True)
    
    full_response = ""
    # 注意：对于流式调用，我们使用 `.astream()` 方法
    async for chunk in streaming_chain.astream({"question": question}):
        print(chunk, end="", flush=True)  # 逐块打印，模拟打字效果
        full_response += chunk
    print()  # 换行
    return full_response

# 测试异步流式
async def main():
    answer = await stream_response("请用一段话介绍人工智能的现状。")
    # 你可以在这里对完整的 answer 进行后续处理

asyncio.run(main())

在Web框架（如FastAPI）中，你可以很容易地将这个异步生成器整合到接口中，实现真正的实时流式响应。这对于构建聊天机器人前端至关重要。

6. 迈向生产：架构优化与部署考量

一个能在实验室跑通的Demo和一個能扛住生产环境流量的服务之间，还有很长的路要走。以下是一些关键的优化和部署建议。

6.1 性能与稳定性优化

• 超时与重试：网络和API服务并不总是稳定的。为你的LLM调用配置合理的超时和重试机制。

from tenacity import retry, stop_after_attempt, wait_exponential

llm = ChatOpenAI(
    # ... 其他参数同上
    max_retries=3,  # LangChain内置的重试机制
    timeout=30.0,
)

# 更精细化的重试控制（使用tenacity库）
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def robust_invoke(chain, input_data):
    """一个包装函数，为链调用增加重试逻辑"""
    return chain.invoke(input_data)

• 缓存：对于相同或相似的问题，重复调用LLM是巨大的浪费。可以为嵌入模型和LLM添加缓存。

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

# 使用内存缓存（生产环境建议使用Redis或数据库）
set_llm_cache(InMemoryCache())
# 现在，相同的提示词调用会直接返回缓存结果，极大节省成本和时间。

6.2 监控与日志

没有监控的系统就是在“裸奔”。你需要知道你的问答系统：

• 响应延迟是多少？
• 每次调用消耗了多少Token（成本）？
• 用户的哪些问题经常检索不到答案？
• 模型生成的答案质量如何？

LangChain官方提供了LangSmith这个强大的监控平台，它可以可视化跟踪每一次链调用的详细步骤、输入输出、耗时和Token使用情况。

# 基础配置示例，需要在LangSmith官网创建账户并获取API Key
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = "your_langsmith_api_key"
os.environ["LANGCHAIN_PROJECT"] = "My-QA-System"  # 你的项目名

配置好后，所有通过LangChain执行的调用都会自动记录到LangSmith仪表盘，供你分析优化。

6.3 部署模式：从脚本到服务

最终，我们需要将系统封装成服务。FastAPI是一个高性能的现代Python Web框架，非常适合部署AI应用。

# app/main_api.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import asyncio
from .chains import get_rag_chain  # 假设你的RAG链定义在其他模块

app = FastAPI(title="智能问答系统API")

class QueryRequest(BaseModel):
    question: str
    session_id: str = None  # 用于区分不同对话会话

@app.post("/ask")
async def ask_question(request: QueryRequest):
    """同步问答接口（非流式）"""
    try:
        chain = get_rag_chain()
        answer = chain.invoke({"question": request.question})
        return {"answer": answer}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/ask/stream")
async def ask_question_stream(request: QueryRequest):
    """流式问答接口"""
    async def event_generator():
        chain = get_rag_chain()
        async for chunk in chain.astream({"question": request.question}):
            # 按照Server-Sent Events格式 yield 数据
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

这个简单的API提供了同步和流式两种问答端点。你可以使用uvicorn来运行它：uvicorn app.main_api:app --reload --host 0.0.0.0 --port 8000。

6.4 安全与成本控制

• API密钥管理：永远使用环境变量或密钥管理服务，不要在代码或版本库中暴露。
• 输入验证与过滤：对用户输入进行清洗，防止提示词注入攻击。
• 限流与配额：在API网关或应用层对用户进行限流，防止滥用和成本失控。
• Token计数与预算：监控每次调用的Token消耗，为不同用户或功能设置预算上限。tiktoken库可以帮助你精确计算提示词和响应的Token数。

构建一个健壮的智能问答系统，技术实现只是一部分，围绕它的运维、监控、安全和成本管理，才是保证其长期稳定运行的关键。从LangChain和DeepSeek这个强大的组合出发，不断迭代你的提示词、优化检索策略、丰富知识库，你就能打造出一个真正懂业务、有知识的AI助手。