DeepSeek + LangChain 开发教程

# 检查 Python 版本 python --version # 输出示例: Python 3.11.8 # 如果没有安装，请访问 https://www.python.org/downloads/ 下载安装

# 创建项目目录 mkdir deepseek-langchain-app cd deepseek-langchain-app # 创建虚拟环境 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate

# 核心依赖 pip install langchain langchain-openai # 后续章节需要的可选依赖 pip install langchain-community faiss-cpu tiktoken

# Windows PowerShell $env:DEEPSEEK_API_KEY = "sk-your-api-key-here" # macOS/Linux export DEEPSEEK_API_KEY="sk-your-api-key-here" # 或者在 Python 代码中读取 import os api_key = os.getenv("DEEPSEEK_API_KEY")

安全提示

永远不要在代码中硬编码 API Key。使用环境变量或 .env 文件管理敏感信息。将 .env 文件添加到 .gitignore 中。

from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage # 创建 DeepSeek 模型实例 llm = ChatOpenAI( model="deepseek-chat", # V3 模型；R1 推理用 "deepseek-reasoner" api_key="sk-your-api-key-here", # 建议用环境变量 base_url="https://api.deepseek.com/v1", temperature=0.7, max_tokens=512, ) # 发送消息 messages = [ SystemMessage(content="你是一个专业的中文 AI 助手，回答简洁准确。"), HumanMessage(content="请用三句话介绍 LangChain 框架。"), ] response = llm.invoke(messages) print(response.content)

from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage llm = ChatOpenAI( model="deepseek-chat", api_key="sk-your-api-key-here", base_url="https://api.deepseek.com/v1", streaming=True, # 启用流式输出 ) messages = [HumanMessage(content="用 Python 写一个快速排序算法")] for chunk in llm.stream(messages): print(chunk.content, end="", flush=True)

from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage # 使用 DeepSeek R1 推理模型 llm = ChatOpenAI( model="deepseek-reasoner", # R1 推理模型 api_key="sk-your-api-key-here", base_url="https://api.deepseek.com/v1", ) messages = [HumanMessage(content="证明根号2是无理数")] response = llm.invoke(messages) print(response.content)

模型选择建议

• deepseek-chat（V3）：日常对话、内容写作、翻译、通用代码生成，性价比最高
• deepseek-reasoner（R1）：数学推理、复杂逻辑分析、算法设计、科学研究

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 初始化模型 llm = ChatOpenAI( model="deepseek-chat", api_key="sk-your-api-key-here", base_url="https://api.deepseek.com/v1", temperature=0.7, ) # 定义提示词模板 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个{role}，擅长{skill}。回答简洁专业。"), ("user", "{input}"), ]) # 构建链：Prompt -> LLM -> 输出解析 chain = prompt | llm | StrOutputParser() # 运行链 result = chain.invoke({ "role": "Python 后端工程师", "skill": Django REST Framework", "input": "如何设计一个高性能的 REST API？请列出 5 个关键原则。", }) print(result)

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm = ChatOpenAI( model="deepseek-chat", api_key="sk-your-api-key-here", base_url="https://api.deepseek.com/v1", temperature=0.7, ) # 第一步：生成大纲 outline_prompt = ChatPromptTemplate.from_template( "请为以下主题生成一个包含 3 个要点的文章大纲：\n\n主题：{topic}" ) # 第二步：根据大纲生成正文 write_prompt = ChatPromptTemplate.from_template( "根据以下大纲，写一篇 500 字左右的文章：\n\n大纲：{outline}\n\n请使用流畅的中文写作。" ) # 构建顺序链 outline_chain = outline_prompt | llm | StrOutputParser() write_chain = write_prompt | llm | StrOutputParser() # 使用 RunnableLambda 串联 from langchain_core.runnables import RunnableLambda full_chain = ( RunnableLambda(lambda x: {"outline": outline_chain.invoke({"topic": x["topic"]})}) | RunnableLambda(lambda x: write_chain.invoke({"outline": x["outline"]})) ) result = full_chain.invoke({"topic": "人工智能在医疗领域的应用"}) print(result)

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser llm = ChatOpenAI( model="deepseek-chat", api_key="sk-your-api-key-here", base_url="https://api.deepseek.com/v1", ) # LCEL 风格：翻译链 translate_prompt = ChatPromptTemplate.from_template( "将以下文本翻译成{target_lang}，只输出翻译结果：\n\n{text}" ) translate_chain = translate_prompt | llm | StrOutputParser() response = translate_chain.invoke({ "target_lang": "英文", "text": "人工智能正在深刻改变我们的生活方式和工作方式。", }) print(response)

from langchain_openai import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain.chains import ConversationChain llm = ChatOpenAI( model="deepseek-chat", api_key="sk-your-api-key-here", base_url="https://api.deepseek.com/v1", temperature=0.7, ) # 创建记忆组件 memory = ConversationBufferMemory() # 创建对话链 conversation = ConversationChain( llm=llm, memory=memory, verbose=True, # 打印调试信息 ) # 多轮对话 print(conversation.predict(input="你好，我叫小明。")) print(conversation.predict(input="我喜欢吃火锅，你能推荐几家北京的火锅店吗？")) print(conversation.predict(input="你还记得我叫什么名字吗？"))

from langchain.memory import ConversationBufferWindowMemory from langchain.chains import ConversationChain memory = ConversationBufferWindowMemory(k=3) # 只保留最近 3 轮对话 conversation = ConversationChain( llm=llm, memory=memory, verbose=True, ) print(conversation.predict(input="第一轮：我叫小明。")) print(conversation.predict(input="第二轮：我今年 25 岁。")) print(conversation.predict(input="第三轮：我住在北京。")) print(conversation.predict(input="第四轮：我最喜欢的颜色是蓝色。")) print(conversation.predict(input="我叫什么名字？")) # 超出窗口，可能忘记

from langchain.memory import ConversationSummaryMemory from langchain.chains import ConversationChain # 摘要记忆：用 LLM 自动总结历史对话 memory = ConversationSummaryMemory(llm=llm) conversation = ConversationChain( llm=llm, memory=memory, verbose=True, ) # 进行多轮复杂对话 print(conversation.predict( input="请详细解释一下什么是机器学习，包括监督学习、无监督学习和强化学习。" )) print(conversation.predict( input="基于你刚才讲的内容，深度学习属于哪一类？为什么？" )) # 查看当前摘要 print("\n=== 对话摘要 ===") print(memory.load_memory_variables({}))

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_community.chat_message_histories import ChatMessageHistory llm = ChatOpenAI( model="deepseek-chat", api_key="sk-your-api-key-here", base_url="https://api.deepseek.com/v1", ) # 创建带记忆槽位的提示词 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个友好的 AI 助手。"), MessagesPlaceholder(variable_name="history"), ("human", "{input}"), ]) chain = prompt | llm # 存储历史记录 store = {} def get_session_history(session_id: str): if session_id not in store: store[session_id] = ChatMessageHistory() return store[session_id] # 包装为带历史的链 with_history = RunnableWithMessageHistory( chain, get_session_history=get_session_history, input_messages_key="input", history_messages_key="history", ) # 多轮对话 response1 = with_history.invoke( {"input": "你好，我叫小明，是一名 Python 开发者。"}, config={"configurable": {"session_id": "user_001"}}, ) print(response1.content) response2 = with_history.invoke( {"input": "你还记得我的名字和职业吗？"}, config={"configurable": {"session_id": "user_001"}}, ) print(response2.content)

记忆组件选择建议

• 短对话（<10 轮）：ConversationBufferMemory，保留全部上下文
• 长对话（>10 轮）：ConversationBufferWindowMemory 或 ConversationSummaryMemory
• 生产环境：推荐 RunnableWithMessageHistory，结合 Redis 等外部存储

from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langgraph.prebuilt import create_react_agent # 初始化模型 llm = ChatOpenAI( model="deepseek-chat", api_key="sk-your-api-key-here", base_url="https://api.deepseek.com/v1", temperature=0.0, # Agent 场景建议低温度 ) # 定义自定义工具 @tool def calculator(expression: str) -> str: """计算数学表达式。输入应为有效的 Python 数学表达式，如 '2 + 3 * 4'。""" try: result = eval(expression, {"__builtins__": {}}, {}) return f"计算结果：{expression} = {result}" except Exception as e: return f"计算错误：{str(e)}" @tool def get_word_count(text: str) -> str: """统计文本的字数。输入为需要统计的文本字符串。""" # 中文字数统计 chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff') return f"字数统计：共 {chinese_chars} 个中文字符" # 创建 Agent tools = [calculator, get_word_count] agent = create_react_agent(llm, tools) # 测试 Agent response = agent.invoke({ "messages": [ ("user", "请计算 (15 + 27) * 3 - 8 的结果，然后告诉我'人工智能改变世界'这句话有多少个字。") ] }) for msg in response["messages"]: if msg.type == "ai": print(msg.content)

# 安装依赖 # pip install langchain-community tavily-python from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults from langgraph.prebuilt import create_react_agent llm = ChatOpenAI( model="deepseek-chat", api_key="sk-your-api-key-here", base_url="https://api.deepseek.com/v1", temperature=0.0, ) # 创建搜索工具（需要 Tavily API Key） search = TavilySearchResults( max_results=3, tavily_api_key="tvly-your-tavily-key", # 在 https://tavily.com 注册获取 ) tools = [search] agent = create_react_agent(llm, tools) response = agent.invoke({ "messages": [ ("user", "搜索 2026 年关于 DeepSeek 模型的最新进展，用中文总结。") ] }) for msg in response["messages"]: if msg.type == "ai": print(msg.content)

Agent 开发注意事项

• temperature 设为 0：Agent 需要确定性输出，避免随机性导致工具调用失败
• 工具描述要清晰：docstring 中的描述直接影响 Agent 是否正确调用工具
• 安全第一：eval() 仅用于演示，生产环境请使用安全的表达式解析器
• 使用 langgraph：LangGraph 是 LangChain 推荐的 Agent 框架，比旧版 AgentExecutor 更稳定

# 安装依赖 # pip install langchain-community faiss-cpu pypdf from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.document_loaders import TextLoader, PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import FAISS from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # 1. 初始化模型 llm = ChatOpenAI( model="deepseek-chat", api_key="sk-your-api-key-here", base_url="https://api.deepseek.com/v1", temperature=0.3, # RAG 场景建议较低温度 ) # 2. 加载文档 # 加载文本文件 loader = TextLoader("data/knowledge.txt", encoding="utf-8") documents = loader.load() # 如果加载 PDF # loader = PyPDFLoader("data/report.pdf") # documents = loader.load() # 3. 文本分割 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每块 500 字符 chunk_overlap=50, # 重叠 50 字符，保证上下文连贯 separators=["\n\n", "\n", "。", "！", "？", " "], ) chunks = text_splitter.split_documents(documents) print(f"文档已分割为 {len(chunks)} 个块") # 4. 创建向量存储 # 注意：DeepSeek 暂不提供官方 Embedding API， # 这里使用 OpenAI 的 Embedding 或本地模型 embeddings = OpenAIEmbeddings( model="text-embedding-3-small", # 如果使用本地模型，可选择 HuggingFace 等方案 ) vectorstore = FAISS.from_documents(chunks, embeddings) # 5. 创建检索器 retriever = vectorstore.as_retriever( search_type="similarity", # 相似度检索 search_kwargs={"k": 4}, # 返回最相关的 4 个文档块 ) # 6. 构建 RAG 提示词 rag_prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个基于文档内容的问答助手。 请根据以下上下文信息回答用户的问题。 如果上下文中没有相关信息，请如实说"根据提供的文档，无法回答此问题"。 不要编造任何信息。 上下文信息： {context}"""), ("human", "{question}"), ]) # 7. 构建 RAG 链 def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs) rag_chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | rag_prompt | llm | StrOutputParser() ) # 8. 提问 question = "这份文档中关于项目时间线的关键节点有哪些？" answer = rag_chain.invoke(question) print(answer)

# 安装依赖 # pip install sentence-transformers from langchain_community.embeddings import HuggingFaceEmbeddings # 使用本地 BGE 中文 Embedding 模型 embeddings = HuggingFaceEmbeddings( model_name="BAAI/bge-small-zh-v1.5", model_kwargs={"device": "cpu"}, encode_kwargs={"normalize_embeddings": True}, ) # 其余代码与上面相同 vectorstore = FAISS.from_documents(chunks, embeddings)

RAG 优化建议

• chunk_size 调优：太小丢失上下文，太大检索精度下降。中文文档建议 300-800 字符
• 混合检索：结合关键词检索（BM25）和向量检索，提升召回率
• 重排序：在检索后使用 Cross-Encoder 对结果进行重排序
• 引用溯源：在 prompt 中要求模型引用具体文档片段，增强可信度

# 安装依赖 # pip install langchain-ollama from langchain_ollama import ChatOllama from langchain_core.messages import HumanMessage, SystemMessage # 连接本地 Ollama 中的 DeepSeek 模型 llm = ChatOllama( model="deepseek-r1:8b", # 你的 Ollama 模型名称 base_url="http://localhost:11434", # Ollama 默认地址 temperature=0.7, ) messages = [ SystemMessage(content="你是一个专业的 Python 编程助手。"), HumanMessage(content="请用 Python 写一个装饰器，用于计算函数执行时间。"), ] response = llm.invoke(messages) print(response.content)

from langchain_ollama import ChatOllama, OllamaEmbeddings from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import FAISS from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # 使用本地 Ollama 模型和 Embedding llm = ChatOllama(model="deepseek-r1:8b", temperature=0.3) embeddings = OllamaEmbeddings(model="nomic-embed-text") # 需要先 ollama pull nomic-embed-text # 加载文档 loader = TextLoader("data/knowledge.txt", encoding="utf-8") documents = loader.load() # 分割文档 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=50, ) chunks = text_splitter.split_documents(documents) # 创建向量存储 vectorstore = FAISS.from_documents(chunks, embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 4}) # 构建 RAG 链 rag_prompt = ChatPromptTemplate.from_messages([ ("system", "基于以下上下文回答问题：\n\n{context}"), ("human", "{question}"), ]) def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs) rag_chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | rag_prompt | llm | StrOutputParser() ) question = "文档中提到了哪些关键技术？" answer = rag_chain.invoke(question) print(answer)

本地部署注意事项

• Ollama 服务运行中：确保 Ollama 在后台运行，默认监听 localhost:11434
• 模型已下载：使用 ollama list 确认模型已存在
• Embedding 模型：本地 RAG 需要单独的 Embedding 模型，如 nomic-embed-text 或 bge-m3
• 性能取决于硬件：8B 模型在消费级 GPU 上运行良好，CPU 推理速度较慢

chatbot/ ├── main.py # 主入口 ├── chatbot.py # 核心聊天逻辑 ├── knowledge.py # 知识库管理 ├── tools.py # 自定义工具 ├── config.py # 配置文件 ├── data/ │ └── faq.txt # 知识库文档 └── requirements.txt # 依赖列表

import os # DeepSeek API 配置 DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "sk-your-api-key") DEEPSEEK_BASE_URL = "https://api.deepseek.com/v1" DEEPSEEK_MODEL = "deepseek-chat" # 模型参数 TEMPERATURE = 0.7 MAX_TOKENS = 2048 # 知识库配置 KNOWLEDGE_BASE_DIR = "data" CHUNK_SIZE = 500 CHUNK_OVERLAP = 50 RETRIEVAL_K = 4 # 系统提示词 SYSTEM_PROMPT = """你是一个专业的智能客服助手，名称为"小深助手"。 你需要： 1. 用友好、专业的态度回答用户问题 2. 优先使用知识库中的信息回答产品相关问题 3. 如果知识库中没有相关信息，使用通用知识回答 4. 遇到计算类问题，使用计算器工具 5. 始终使用中文回复"""

from langchain_core.tools import tool from datetime import datetime @tool def calculator(expression: str) -> str: """计算数学表达式。输入为有效的 Python 数学表达式。""" try: safe_dict = { "abs": abs, "round": round, "min": min, "max": max, "sum": sum, "pow": pow, "int": int, "float": float, } result = eval(expression, {"__builtins__": safe_dict}, {}) return f"计算结果：{expression} = {result}" except Exception as e: return f"计算错误：{str(e)}" @tool def get_current_time() -> str: """获取当前日期和时间。""" now = datetime.now() return f"当前时间：{now.strftime('%Y年%m月%d日 %H:%M:%S')}（星期{['一','二','三','四','五','六','日'][now.weekday()]}）"

import os from langchain_community.document_loaders import TextLoader, DirectoryLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings from config import ( KNOWLEDGE_BASE_DIR, CHUNK_SIZE, CHUNK_OVERLAP, RETRIEVAL_K, DEEPSEEK_API_KEY, ) class KnowledgeBase: """知识库管理器""" def __init__(self): self.embeddings = OpenAIEmbeddings( model="text-embedding-3-small", api_key=DEEPSEEK_API_KEY, ) self.vectorstore = None self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP, separators=["\n\n", "\n", "。", "！", "？"], ) def load_documents(self) -> bool: """加载知识库文档""" if not os.path.exists(KNOWLEDGE_BASE_DIR): print(f"知识库目录不存在：{KNOWLEDGE_BASE_DIR}") return False loader = DirectoryLoader( KNOWLEDGE_BASE_DIR, glob="*.txt", loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"}, ) documents = loader.load() if not documents: print("知识库中没有找到文档") return False chunks = self.text_splitter.split_documents(documents) self.vectorstore = FAISS.from_documents(chunks, self.embeddings) print(f"知识库加载完成：{len(documents)} 个文档，{len(chunks)} 个片段") return True def search(self, query: str) -> str: """搜索相关知识""" if not self.vectorstore: return "" docs = self.vectorstore.similarity_search(query, k=RETRIEVAL_K) return "\n\n".join(doc.page_content for doc in docs)

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_community.chat_message_histories import ChatMessageHistory from config import ( DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL, DEEPSEEK_MODEL, TEMPERATURE, MAX_TOKENS, SYSTEM_PROMPT, ) from knowledge import KnowledgeBase class DeepSeekChatbot: """DeepSeek 智能客服机器人""" def __init__(self): # 初始化 LLM self.llm = ChatOpenAI( model=DEEPSEEK_MODEL, api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL, temperature=TEMPERATURE, max_tokens=MAX_TOKENS, streaming=True, ) # 初始化知识库 self.kb = KnowledgeBase() kb_loaded = self.kb.load_documents() # 构建提示词 if kb_loaded: system_template = SYSTEM_PROMPT + "\n\n知识库参考信息：\n{knowledge}" else: system_template = SYSTEM_PROMPT self.prompt = ChatPromptTemplate.from_messages([ ("system", system_template), MessagesPlaceholder(variable_name="history"), ("human", "{input}"), ]) # 创建链 chain = self.prompt | self.llm # 对话历史存储 self.store = {} self.chain_with_history = RunnableWithMessageHistory( chain, get_session_history=self._get_session_history, input_messages_key="input", history_messages_key="history", ) def _get_session_history(self, session_id: str): if session_id not in self.store: self.store[session_id] = ChatMessageHistory() return self.store[session_id] def chat(self, user_input: str, session_id: str = "default"): """发送消息并获得回复""" # 搜索知识库 knowledge = self.kb.search(user_input) if self.kb.vectorstore else "知识库未加载" # 调用链 response = self.chain_with_history.invoke( { "input": user_input, "knowledge": knowledge, }, config={"configurable": {"session_id": session_id}}, ) return response.content def chat_stream(self, user_input: str, session_id: str = "default"): """流式聊天""" knowledge = self.kb.search(user_input) if self.kb.vectorstore else "知识库未加载" for chunk in self.chain_with_history.stream( { "input": user_input, "knowledge": knowledge, }, config={"configurable": {"session_id": session_id}}, ): if chunk.content: yield chunk.content

from chatbot import DeepSeekChatbot def main(): """智能客服机器人主程序""" print("=" * 50) print(" 小深助手 - DeepSeek 智能客服机器人") print(" 输入 'quit' 退出, 'clear' 清空对话历史") print("=" * 50) # 初始化机器人 print("\n正在初始化...") bot = DeepSeekChatbot() print("初始化完成！\n") session_id = "user_default" while True: try: user_input = input("\n您: ").strip() if not user_input: continue if user_input.lower() == 'quit': print("\n感谢使用，再见！") break if user_input.lower() == 'clear': bot.store.pop(session_id, None) print("对话历史已清空。") continue # 流式输出回复 print("\n小深助手: ", end="", flush=True) for chunk in bot.chat_stream(user_input, session_id): print(chunk, end="", flush=True) print() except KeyboardInterrupt: print("\n\n感谢使用，再见！") break except Exception as e: print(f"\n发生错误：{e}") if __name__ == "__main__": main()

# 安装依赖 pip install langchain langchain-openai langchain-community langchain-text-splitters faiss-cpu # 设置 API Key # Windows: $env:DEEPSEEK_API_KEY = "sk-your-api-key" # macOS/Linux: export DEEPSEEK_API_KEY="sk-your-api-key" # 创建知识库文档 mkdir data echo "产品A售价999元，支持7天无理由退货。产品B售价1999元，提供2年质保。" > data/faq.txt # 运行 python main.py