引言:为什么选择 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:

  1. 访问 DeepSeek 开放平台
  2. 注册账号并完成实名认证
  3. 在控制台创建 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_sizechunk_overlap 参数
  • 尝试不同的文本分割策略
  • 增加检索数量 search_kwargs={"k": 5}

3. 响应速度慢

问题:问答延迟高
解决

  • 使用异步调用 ainvoke()astream()
  • 减少 max_tokens 限制
  • 考虑缓存常用查询结果

4. 内存占用过高

问题:处理大文档时内存不足
解决

  • 减小 chunk_size
  • 使用更高效的嵌入模型
  • 分批处理文档

下一步学习建议

  1. 深入组件:学习更多 LangChain 组件,如记忆(Memory)、工具(Tools)、代理(Agents)
  2. 集成其他服务:尝试将 LangChain 与数据库、API、文件系统等集成
  3. 部署应用:学习使用 FastAPI 或 Streamlit 将你的 LangChain 应用部署为 Web 服务
  4. 性能优化:探索缓存、批处理、异步等优化技巧
  5. 社区资源

总结

通过本教程,你已经掌握了使用 LangChain 1.3 和 DeepSeek 构建 AI 应用的基础知识。从简单的对话链到复杂的文档问答系统,LangChain 提供了强大而灵活的工具集,让大语言模型的应用开发变得更加高效。

关键要点:

  • LangChain 1.3 的 LCEL 语法让链式调用更简洁
  • DeepSeek 提供了与 OpenAI 兼容的高质量 API
  • 向量数据库是实现文档问答的核心技术
  • 异步和流式处理可以显著提升用户体验

现在,你可以基于这些知识构建更复杂的应用,如多轮对话系统、自动化工作流或智能数据分析工具。记住,最好的学习方式就是动手实践——尝试修改代码、添加新功能,或者将 LangChain 应用到你的实际项目中。

祝你开发愉快!

Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐