LangChain + DeepSeek 完整集成指南

LangChain + DeepSeek 概述

LangChain 是 Python/JavaScript 的 LLM 应用开发框架，DeepSeek 通过 OpenAI 兼容 API 可以无缝接入 LangChain 的所有组件：Chat Model、Embeddings、Chains、Agents、Memory、Retrievers 等。只需修改 base_url 即可使用。

1. 基础配置

# 安装
pip install langchain langchain-openai python-dotenv

# 配置 .env
DEEPSEEK_API_KEY=sk-your-key
DEEPSEEK_BASE_URL=https://api.deepseek.com

2. ChatModel 集成

from langchain_openai import ChatOpenAI
from langchain.schema import SystemMessage, HumanMessage
import os
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(
    model="deepseek-v4-flash",
    openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
    openai_api_base=os.getenv("DEEPSEEK_BASE_URL"),
    temperature=0.7,
    max_tokens=2048,
)

# 基础对话
messages = [
    SystemMessage(content="你是一个 Python 专家"),
    HumanMessage(content="解释装饰器的原理")
]
response = llm.invoke(messages)
print(response.content)

3. 流式输出

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

llm_stream = ChatOpenAI(
    model="deepseek-v4-flash",
    openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
    openai_api_base=os.getenv("DEEPSEEK_BASE_URL"),
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()],
    temperature=0.7,
)

# 流式生成——逐 token 输出
llm_stream.invoke("写一首关于AI的诗")

4. Embeddings 集成

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
    openai_api_base=os.getenv("DEEPSEEK_BASE_URL"),
)

# 生成文本向量
text = "DeepSeek V4 是当前性价比最高的AI模型"
vector = embeddings.embed_query(text)
print(f"向量维度: {len(vector)}")  # 1536

# 批量生成
texts = ["AI模型", "机器学习", "深度学习"]
vectors = embeddings.embed_documents(texts)
print(f"生成了 {len(vectors)} 个向量")

5. Tool Calling（Agent 工具）

from langchain.tools import tool
from langchain.agents import initialize_agent, AgentType

@tool
def get_weather(city: str) -> str:
    """获取城市天气"""
    # 实际调用天气 API
    return f"{city}今天晴天，25°C"

@tool
def calculator(expression: str) -> str:
    """执行数学计算"""
    return str(eval(expression))

tools = [get_weather, calculator]

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    handle_parsing_errors=True,
)

result = agent.invoke("北京天气怎么样？顺便算一下 256*128")
print(result['output'])

6. RAG Chain（检索增强生成）

from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA

# 1. 加载文档
loader = TextLoader("knowledge.txt")
docs = loader.load()

# 2. 分割文本
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = text_splitter.split_documents(docs)

# 3. 构建向量存储
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(
        model="text-embedding-3-small",
        openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
        openai_api_base=os.getenv("DEEPSEEK_BASE_URL"),
    ),
    persist_directory="./chroma_db"
)

# 4. 创建 RAG Chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    return_source_documents=True,
)

# 5. 查询
result = qa_chain.invoke({"query": "DeepSeek V4 的特点是什么？"})
print(result['result'])
print(f"来源文档数：{len(result['source_documents'])}")

7. Conversation Memory（对话记忆）

from langchain.memory import ConversationSummaryBufferMemory
from langchain.chains import ConversationChain

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,  # 摘要压缩阈值
    return_messages=True,
)

conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True,
)

# 多轮对话
conversation.predict(input="我叫张三，是一名Python开发者")
conversation.predict(input="我主要做Web后端开发")
conversation.predict(input="还记得我的名字和职业吗？")
# 输出: 你是张三，一名Python Web后端开发者

8. 生产环境优化

from langchain.callbacks import get_openai_callback

# Token 用量监控
with get_openai_callback() as cb:
    response = llm.invoke("解释什么是RAG")
    print(f"Token消耗: {cb.total_tokens}")
    print(f"费用估算: ¥{cb.total_cost:.4f}")

# 缓存（减少重复调用）
from langchain.cache import InMemoryCache
import langchain

langchain.llm_cache = InMemoryCache()
# 第二次调用同一 prompt 将直接从缓存返回