LangChain 1.3 入门指南:基于 DeepSeek 构建你的第一个 AI 应用
引言:为什么选择 LangChain?
在当今 AI 应用开发浪潮中,如何高效地将大语言模型(LLM)与外部数据、工具和业务逻辑连接起来,是每个开发者面临的挑战。LangChain 作为一个开源框架,专门为解决这一问题而生。它提供了一套标准化的接口、组件和链式调用模式,让开发者能够像搭积木一样构建复杂的 AI 应用。
随着 LangChain 1.3 版本的发布,框架在稳定性、易用性和功能丰富度上都有了显著提升。本教程将带你从零开始,使用最新的 LangChain 1.3 和国产优秀大模型 DeepSeek,快速搭建你的第一个智能对话或文本处理应用。
环境准备与安装
在开始编码之前,我们需要准备好 Python 开发环境并安装必要的依赖。
1. 创建虚拟环境(推荐)
为了避免包冲突,建议使用虚拟环境。
# 使用 conda
conda create -n langchain-demo python=3.10
conda activate langchain-demo
# 或使用 venv
python -m venv langchain-demo
source langchain-demo/bin/activate # Linux/Mac
# 或 .\langchain-demo\Scripts\activate # Windows
2. 安装核心依赖
我们将安装 LangChain 1.3 的核心包,以及用于调用 DeepSeek 模型的 OpenAI 兼容接口包。
pip install langchain==1.3.0
pip install langchain-community==0.3.0
pip install openai==1.30.0
3. 获取 DeepSeek API Key
DeepSeek 提供了与 OpenAI 兼容的 API 接口,你需要先注册并获取 API Key:
- 访问 DeepSeek 开放平台
- 注册账号并完成实名认证
- 在控制台创建 API Key 并妥善保存
第一个 LangChain 应用:与 DeepSeek 对话
让我们从一个最简单的例子开始,使用 LangChain 调用 DeepSeek 模型进行对话。
1. 基础对话链
创建一个 chat.py 文件,写入以下代码:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
# 配置 DeepSeek API(使用 OpenAI 兼容接口)
llm = ChatOpenAI(
model="deepseek-chat", # 或 "deepseek-coder" 用于代码生成
openai_api_key="你的-DeepSeek-API-Key",
base_url="https://api.deepseek.com/v1", # DeepSeek API 地址
temperature=0.7, # 控制创造性,0-1之间
)
# 创建提示词模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个乐于助人的 AI 助手,请用中文回答用户的问题。"),
("user", "{input}")
])
# 构建链:提示词 -> 模型 -> 输出解析
chain = prompt | llm | StrOutputParser()
# 运行链
response = chain.invoke({"input": "请用简单的语言解释什么是 LangChain?"})
print("DeepSeek 的回答:", response)
运行这个脚本,你将看到 DeepSeek 对 LangChain 的解释。这里的关键点:
ChatOpenAI是 LangChain 对 OpenAI 兼容接口的封装ChatPromptTemplate用于管理对话格式的提示词|操作符是 LangChain 1.3 推荐的链式语法(LCEL)
2. 流式输出
对于较长的回答,流式输出可以提供更好的用户体验:
# 启用流式输出
for chunk in chain.stream({"input": "Python 中的装饰器是什么?"}):
print(chunk, end="", flush=True)
核心概念深入:组件与链
LangChain 的核心设计思想是将复杂任务分解为可重用的组件,并通过链(Chain)将它们连接起来。
1. 模型(Models)
LangChain 支持多种模型接口。对于 DeepSeek,我们使用 ChatOpenAI,因为它兼容 OpenAI 的聊天格式。
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="deepseek-chat",
openai_api_key="your-api-key",
base_url="https://api.deepseek.com/v1",
max_tokens=1000, # 限制最大输出长度
)
2. 提示词(Prompts)
提示词模板让你可以动态插入变量,创建可复用的提示词。
from langchain_core.prompts import ChatPromptTemplate
# 多轮对话模板
prompt_template = ChatPromptTemplate.from_messages([
("system", "你是一位{role},请用{language}回答。"),
("human", "问题:{question}"),
])
# 填充模板
formatted_prompt = prompt_template.invoke({
"role": "编程专家",
"language": "中文",
"question": "如何优化 Python 代码性能?"
})
3. 输出解析器(Output Parsers)
将模型的原始输出转换为结构化数据。
from langchain_core.output_parsers import CommaSeparatedListOutputParser
parser = CommaSeparatedListOutputParser()
result = parser.invoke("苹果, 香蕉, 橙子")
print(result) # ['苹果', '香蕉', '橙子']
4. 链(Chains)
链是 LangChain 的核心抽象,将多个组件连接成工作流。
from langchain_core.runnables import RunnablePassthrough
# 创建复杂链
complex_chain = (
{"topic": RunnablePassthrough()} # 传递输入
| prompt_template # 应用提示词
| llm # 调用模型
| parser # 解析输出
)
# 执行链
result = complex_chain.invoke("Python 机器学习库")
实战项目:构建智能文档问答系统
现在让我们构建一个更实用的应用:一个可以回答关于特定文档内容的问答系统。
1. 项目结构
document-qa/
├── main.py # 主程序
├── documents/ # 存放文档
│ └── langchain_intro.txt
└── requirements.txt
2. 文档加载与分割
首先,我们需要加载文档并将其分割成适合处理的片段。
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 加载文档
loader = TextLoader("documents/langchain_intro.txt")
documents = loader.load()
# 分割文档
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 每个块的最大字符数
chunk_overlap=50, # 块之间的重叠字符数
separators=["\n\n", "\n", "。", ",", " ", ""] # 分割符优先级
)
chunks = text_splitter.split_documents(documents)
print(f"原始文档数:{len(documents)}")
print(f"分割后的块数:{len(chunks)}")
3. 向量化与存储
使用向量数据库存储文档片段,以便快速检索。
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
# 初始化嵌入模型(DeepSeek 也提供嵌入接口)
embeddings = OpenAIEmbeddings(
model="text-embedding-ada-002", # 可以使用其他兼容模型
openai_api_key="your-api-key",
base_url="https://api.deepseek.com/v1",
)
# 创建向量数据库
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db" # 持久化存储
)
# 创建检索器
retriever = vectorstore.as_retriever(
search_type="similarity", # 相似度搜索
search_kwargs={"k": 3} # 返回最相关的3个片段
)
4. 构建问答链
将检索到的文档片段与问题结合,让模型生成答案。
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough
# 定义提示词模板
template = """基于以下上下文回答问题。如果你不知道答案,就说不知道,不要编造。
上下文:
{context}
问题:{question}
答案:"""
prompt = ChatPromptTemplate.from_template(template)
# 初始化 DeepSeek 模型
llm = ChatOpenAI(
model="deepseek-chat",
openai_api_key="your-api-key",
base_url="https://api.deepseek.com/v1",
temperature=0,
)
# 构建问答链
qa_chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
# 测试问答
question = "LangChain 的主要作用是什么?"
answer = qa_chain.invoke(question)
print(f"问题:{question}")
print(f"答案:{answer}")
5. 完整应用示例
import os
from typing import List
class DocumentQASystem:
def __init__(self, api_key: str, document_path: str = None):
self.api_key = api_key
self.document_path = document_path
self.vectorstore = None
self.qa_chain = None
def load_and_process_documents(self):
"""加载并处理文档"""
if not self.document_path or not os.path.exists(self.document_path):
print("警告:未提供文档路径或路径不存在")
return
# 加载文档
loader = TextLoader(self.document_path)
documents = loader.load()
# 分割文档
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
separators=["\n\n", "\n", "。", ",", " ", ""]
)
chunks = text_splitter.split_documents(documents)
# 创建向量存储
embeddings = OpenAIEmbeddings(
model="text-embedding-ada-002",
openai_api_key=self.api_key,
base_url="https://api.deepseek.com/v1",
)
self.vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db"
)
print(f"文档处理完成,共 {len(chunks)} 个片段")
def setup_qa_chain(self):
"""设置问答链"""
if not self.vectorstore:
print("请先加载文档")
return
retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3})
# 提示词模板
template = """你是一个专业的文档助手。请基于以下上下文回答问题。
上下文:
{context}
问题:{question}
请提供准确、简洁的答案。如果上下文没有相关信息,请说"根据提供的上下文,我无法回答这个问题。"。
答案:"""
prompt = ChatPromptTemplate.from_template(template)
# 初始化模型
llm = ChatOpenAI(
model="deepseek-chat",
openai_api_key=self.api_key,
base_url="https://api.deepseek.com/v1",
temperature=0.3,
max_tokens=500,
)
# 构建链
self.qa_chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
print("问答链设置完成")
def ask(self, question: str) -> str:
"""提问"""
if not self.qa_chain:
return "请先设置问答链"
return self.qa_chain.invoke(question)
def interactive_mode(self):
"""交互模式"""
print("=== 文档问答系统 ===")
print("输入 'quit' 或 '退出' 结束对话")
print("-" * 30)
while True:
question = input("\n你的问题:").strip()
if question.lower() in ['quit', '退出', 'exit']:
print("再见!")
break
if not question:
continue
answer = self.ask(question)
print(f"\n答案:{answer}")
# 使用示例
if __name__ == "__main__":
# 替换为你的 DeepSeek API Key
API_KEY = "your-deepseek-api-key"
# 初始化系统
qa_system = DocumentQASystem(
api_key=API_KEY,
document_path="documents/langchain_intro.txt"
)
# 处理文档
qa_system.load_and_process_documents()
# 设置问答链
qa_system.setup_qa_chain()
# 单次提问
# answer = qa_system.ask("LangChain 是什么?")
# print(answer)
# 或进入交互模式
qa_system.interactive_mode()
LangChain 1.3 新特性与最佳实践
1. LCEL(LangChain Expression Language)
LCEL 是 LangChain 1.3 推荐的新语法,使用 | 操作符连接组件,代码更简洁。
# 传统方式 vs LCEL
# 传统
chain = LLMChain(llm=llm, prompt=prompt)
# LCEL(推荐)
chain = prompt | llm | parser
2. 异步支持
LangChain 1.3 全面支持异步操作,提升并发性能。
import asyncio
async def async_qa():
# 异步调用链
result = await qa_chain.ainvoke("什么是异步编程?")
print(result)
# 运行异步函数
asyncio.run(async_qa())
3. 流式处理
除了流式输出,还可以流式处理中间结果。
# 流式处理每个步骤
async for chunk in qa_chain.astream("Python 的 GIL 是什么?"):
print(chunk, end="", flush=True)
4. 调试与监控
使用回调函数监控链的执行过程。
from langchain_core.callbacks import StdOutCallbackHandler
# 添加回调
chain_with_callback = qa_chain.with_config(
callbacks=[StdOutCallbackHandler()]
)
result = chain_with_callback.invoke("如何学习 LangChain?")
常见问题与解决方案
1. API 调用失败
问题:APIError: Invalid API Key
解决:
- 检查 API Key 是否正确
- 确认 API Key 是否有足够余额
- 验证
base_url是否正确设置为https://api.deepseek.com/v1
2. 文档检索不准确
问题:问答系统返回无关答案
解决:
- 调整
chunk_size和chunk_overlap参数 - 尝试不同的文本分割策略
- 增加检索数量
search_kwargs={"k": 5}
3. 响应速度慢
问题:问答延迟高
解决:
- 使用异步调用
ainvoke()或astream() - 减少
max_tokens限制 - 考虑缓存常用查询结果
4. 内存占用过高
问题:处理大文档时内存不足
解决:
- 减小
chunk_size - 使用更高效的嵌入模型
- 分批处理文档
下一步学习建议
- 深入组件:学习更多 LangChain 组件,如记忆(Memory)、工具(Tools)、代理(Agents)
- 集成其他服务:尝试将 LangChain 与数据库、API、文件系统等集成
- 部署应用:学习使用 FastAPI 或 Streamlit 将你的 LangChain 应用部署为 Web 服务
- 性能优化:探索缓存、批处理、异步等优化技巧
- 社区资源:
总结
通过本教程,你已经掌握了使用 LangChain 1.3 和 DeepSeek 构建 AI 应用的基础知识。从简单的对话链到复杂的文档问答系统,LangChain 提供了强大而灵活的工具集,让大语言模型的应用开发变得更加高效。
关键要点:
- LangChain 1.3 的 LCEL 语法让链式调用更简洁
- DeepSeek 提供了与 OpenAI 兼容的高质量 API
- 向量数据库是实现文档问答的核心技术
- 异步和流式处理可以显著提升用户体验
现在,你可以基于这些知识构建更复杂的应用,如多轮对话系统、自动化工作流或智能数据分析工具。记住,最好的学习方式就是动手实践——尝试修改代码、添加新功能,或者将 LangChain 应用到你的实际项目中。
祝你开发愉快!
更多推荐

所有评论(0)