当前位置：首页 > 学习笔记 > 正文内容

LangChain 开发实战教程

廖万里12小时前学习笔记2

# LangChain 开发实战教程：从入门到精通

LangChain 是当前最流行的 LLM 应用开发框架之一，它提供了构建 AI 应用的完整工具链。本文将从实战角度出发，带你掌握 LangChain 的核心概念和实际应用。

一、LangChain 简介

LangChain 是一个开源框架，旨在帮助开发者快速构建基于大语言模型（LLM）的应用程序。它提供了一套标准化的接口和工具，使得开发者可以轻松地将 LLM 与外部数据源、工具和知识库进行集成。

为什么选择 LangChain？

模块化设计：各个组件独立可替换，方便定制

丰富的集成：支持几乎所有主流 LLM 和向量数据库

链式调用：支持复杂的工作流程编排

记忆管理：内置多种记忆机制，实现上下文保持

开源社区活跃：文档完善，生态丰富

官方文档：LangChain Documentation

二、环境搭建

安装依赖

# 基础安装 pip install langchain

# 安装 LangChain 社区集成 pip install langchain-community

# 安装 OpenAI 集成（如使用 OpenAI 模型） pip install langchain-openai

# 安装向量数据库（以 Chroma 为例） pip install chromadb

配置 API Key

import os
from dotenv import load_dotenv
# 从 .env 文件加载环境变量
load_dotenv()
# 设置 OpenAI API Key
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
# 或者使用其他 LLM 提供商
# os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-key"
# os.environ["DEEPSEEK_API_KEY"] = "your-deepseek-key"

建议使用 .env 文件管理敏感信息，避免将 API Key 硬编码在代码中。

三、核心组件详解

1. 模型（Models）

LangChain 支持两种主要类型的模型：

#### LLM（大语言模型）

LLM 接收字符串输入，返回字符串输出：

from langchain_openai import OpenAI
# 初始化 LLM
llm = OpenAI(
    model="gpt-3.5-turbo-instruct",
    temperature=0.7,
    max_tokens=500
)
# 调用模型
response = llm.invoke("请用一句话解释什么是机器学习？")
print(response)

#### Chat Model（聊天模型）

聊天模型接收消息列表，返回消息对象：

from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage, AIMessage
# 初始化聊天模型
chat = ChatOpenAI(
    model="gpt-4",
    temperature=0.7
)
# 构建消息列表
messages = [
    SystemMessage(content="你是一个专业的 Python 开发专家"),
    HumanMessage(content="如何优化 Python 代码的性能？")
]
# 调用模型
response = chat.invoke(messages)
print(response.content)

2. 提示模板（Prompt Templates）

提示模板是 LangChain 的核心组件之一，它允许你动态构建提示词：

from langchain.prompts import PromptTemplate, ChatPromptTemplate
from langchain.chains import LLMChain
# 基础提示模板
template = """你是一个专业的{role}。
请为以下主题写一篇简短的介绍：{topic}
要求：语言简洁，重点突出。"""
prompt = PromptTemplate(
    input_variables=["role", "topic"],
    template=template
)
# 格式化提示词
formatted_prompt = prompt.format(
    role="数据科学家",
    topic="深度学习在医疗领域的应用"
)
print(formatted_prompt)

#### ChatPromptTemplate 示例

from langchain.prompts import ChatPromptTemplate
# 创建聊天提示模板
chat_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个{language}编程专家，擅长{specialty}"),
    ("human", "请解释{concept}的概念，并给出一个代码示例")
])
# 格式化消息
messages = chat_template.format_messages(
    language="Python",
    specialty="异步编程",
    concept="async/await"
)
# 打印消息
for msg in messages:
    print(f"{msg.type}: {msg.content}")

3. 输出解析器（Output Parsers）

输出解析器用于结构化 LLM 的输出：

from langchain.output_parsers import StructuredOutputParser, ResponseSchema
from langchain.prompts import PromptTemplate
# 定义输出结构
response_schemas = [
    ResponseSchema(name="name", description="产品名称"),
    ResponseSchema(name="price", description="产品价格"),
    ResponseSchema(name="features", description="产品特点，以列表形式返回")
]
# 创建解析器
output_parser = StructuredOutputParser.from_response_schemas(response_schemas)
# 获取格式化指令
format_instructions = output_parser.get_format_instructions()
# 创建提示模板
prompt = PromptTemplate(
    template="""请分析以下产品信息，并按照指定格式输出：
{product_info}
{format_instructions}""",
    input_variables=["product_info"],
    partial_variables={"format_instructions": format_instructions}
)
# 使用示例
product_info = "MacBook Pro 14英寸，搭载 M3 Pro 芯片，售价 14999 元，特点包括高性能、长续航、Retina 显示屏"
formatted_prompt = prompt.format(product_info=product_info)
print(formatted_prompt)

4. 记忆（Memory）

LangChain 提供了多种记忆机制来保持对话上下文：

from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory, ConversationBufferWindowMemory
# 初始化模型
llm = ChatOpenAI(model="gpt-3.5-turbo")
# 方式1：完整记忆（保留所有历史）
memory = ConversationBufferMemory()
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)
# 进行对话
response1 = conversation.predict(input="你好，我叫小明")
print(response1)
response2 = conversation.predict(input="你还记得我的名字吗？")
print(response2)
# 方式2：窗口记忆（只保留最近 N 轮对话）
window_memory = ConversationBufferWindowMemory(k=3)
conversation_window = ConversationChain(
    llm=llm,
    memory=window_memory
)

四、链（Chains）详解

链是 LangChain 的核心概念，它允许你将多个组件串联起来执行复杂任务。

1. LLMChain

最基础的链，将 LLM 和提示模板组合：

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain
# 初始化
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{style}风格的内容创作者"),
    ("human", "请为{product}写一句广告语")
])
# 创建链
chain = LLMChain(llm=llm, prompt=prompt)
# 执行
result = chain.run(style="幽默", product="智能手表")
print(result)

2. 简单顺序链（SimpleSequentialChain）

按顺序执行多个链，前一个链的输出作为后一个链的输入：

from langchain.chains import SimpleSequentialChain
# 第一个链：生成故事大纲
first_prompt = ChatPromptTemplate.from_template("请为一个关于{topic}的故事写一个简短大纲")
chain_one = LLMChain(llm=llm, prompt=first_prompt)
# 第二个链：根据大纲写完整故事
second_prompt = ChatPromptTemplate.from_template("根据以下大纲写一个完整的故事：\n{outline}")
chain_two = LLMChain(llm=llm, prompt=second_prompt)
# 组合成顺序链
overall_chain = SimpleSequentialChain(
    chains=[chain_one, chain_two],
    verbose=True
)
# 执行
story = overall_chain.run("人工智能觉醒")
print(story)

3. 路由链（Router Chain）

根据输入动态选择执行的链：

from langchain.chains.router import MultiPromptChain
from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser
from langchain.prompts import PromptTemplate
# 定义不同领域的提示模板
tech_template = """你是一个技术专家。请回答以下问题：
{input}"""
business_template = """你是一个商业顾问。请回答以下问题：
{input}"""
# 创建提示信息
prompt_infos = [
    {
        "name": "技术",
        "description": "适合回答编程、技术架构等问题",
        "prompt_template": tech_template
    },
    {
        "name": "商业",
        "description": "适合回答商业策略、市场营销等问题",
        "prompt_template": business_template
    }
]
# 创建目标链
destination_chains = {}
for p_info in prompt_infos:
    name = p_info["name"]
    prompt_template = PromptTemplate(template=p_info["prompt_template"], input_variables=["input"])
    chain = LLMChain(llm=llm, prompt=prompt_template)
    destination_chains[name] = chain
# 创建默认链
default_chain = LLMChain(llm=llm, prompt=PromptTemplate(template="{input}", input_variables=["input"]))
# 创建路由链
destinations = [f"{p['name']}: {p['description']}" for p in prompt_infos]
destinations_str = "\n".join(destinations)
router_template = f"""给定用户的输入，选择最合适的链来处理。
可能的选项：
{destinations_str}
用户输入: {{input}}
{{previous_messages}}
请选择最合适的链。"""
router_prompt = PromptTemplate(
    template=router_template,
    input_variables=["input", "previous_messages"],
    output_parser=RouterOutputParser()
)
router_chain = LLMRouterChain.from_llm(llm, router_prompt)
# 创建多提示链
chain = MultiPromptChain(
    router_chain=router_chain,
    destination_chains=destination_chains,
    default_chain=default_chain,
    verbose=True
)
# 测试
result = chain.run("如何设计一个高并发的微服务架构？")
print(result)

五、智能体（Agents）

智能体是 LangChain 的高级功能，它允许 LLM 自主决策使用哪些工具来完成任务。

1. 基础智能体

from langchain.agents import initialize_agent, Tool
from langchain_openai import ChatOpenAI
import math
# 定义工具
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误：{e}"
def word_count(text: str) -> str:
    """统计字数"""
    return f"共有 {len(text)} 个字符"
# 创建工具列表
tools = [
    Tool(
        name="Calculator",
        func=calculator,
        description="用于计算数学表达式，输入应该是一个数学表达式字符串"
    ),
    Tool(
        name="WordCount",
        func=word_count,
        description="用于统计文本的字数，输入应该是一段文本"
    )
]
# 初始化 LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 初始化智能体
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent="zero-shot-react-description",
    verbose=True
)
# 运行智能体
result = agent.run("计算 123 * 456，然后告诉我这个结果有多少位数")
print(result)

2. 使用内置工具

LangChain 提供了许多内置工具：

from langchain.agents import load_tools
from langchain.agents import initialize_agent
from langchain_openai import ChatOpenAI
# 初始化 LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 加载内置工具
tools = load_tools(
    ["serpapi", "llm-math"],
    llm=llm
)
# 初始化智能体
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent="zero-shot-react-description",
    verbose=True
)
# 运行
result = agent.run("今天北京的温度是多少？换算成华氏度是多少？")
print(result)

3. 自定义工具

创建自定义工具非常简单：

from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field
# 定义输入模式
class SearchInput(BaseModel):
    query: str = Field(description="搜索查询关键词")
# 自定义工具类
class CustomSearchTool(BaseTool):
    name = "custom_search"
    description = "用于搜索特定信息"
    args_schema: Type[BaseModel] = SearchInput
    def _run(self, query: str) -> str:
        """执行搜索"""
        # 这里可以实现实际的搜索逻辑
        # 示例：模拟搜索结果
        return f"搜索 '{query}' 的结果：[模拟搜索内容]"
    async def _arun(self, query: str) -> str:
        """异步执行"""
        return self._run(query)
# 使用自定义工具
tools = [CustomSearchTool()]
agent = initialize_agent(tools, llm, agent="zero-shot-react-description", verbose=True)
result = agent.run("搜索 LangChain 教程")

六、RAG 应用实战

RAG（检索增强生成）是 LangChain 最流行的应用场景之一。

1. 文档加载

from langchain.document_loaders import TextLoader, PyPDFLoader, WebBaseLoader
# 加载文本文件
text_loader = TextLoader("./data/example.txt")
text_docs = text_loader.load()
# 加载 PDF
pdf_loader = PyPDFLoader("./data/document.pdf")
pdf_docs = pdf_loader.load()
# 加载网页
web_loader = WebBaseLoader("https://example.com/article")
web_docs = web_loader.load()
print(f"加载了 {len(text_docs)} 个文本文档")
print(f"加载了 {len(pdf_docs)} 个 PDF 页面")

2. 文档分割

from langchain.text_splitter import RecursiveCharacterTextSplitter, CharacterTextSplitter
# 创建分割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,          # 每个块的大小
    chunk_overlap=200,        # 块之间的重叠
    length_function=len,
    separators=["\n\n", "\n", " ", ""]
)
# 分割文档
splits = text_splitter.split_documents(pdf_docs)
print(f"分割成 {len(splits)} 个文本块")

3. 向量存储

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
# 初始化嵌入模型
embeddings = OpenAIEmbeddings()
# 创建向量存储
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings,
    persist_directory="./chroma_db"
)
# 相似度搜索
query = "什么是机器学习？"
docs = vectorstore.similarity_search(query, k=3)
for i, doc in enumerate(docs):
    print(f"\n--- 文档 {i+1} ---")
    print(doc.page_content[:200])

4. 完整 RAG 链

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
# 初始化组件
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
embeddings = OpenAIEmbeddings()
# 加载已有的向量存储
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)
# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)
# 创建自定义提示
prompt_template = """基于以下上下文回答问题。如果上下文中没有相关信息，请说"我不知道"。
上下文：
{context}
问题：{question}
回答："""
PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)
# 创建 RAG 链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True,
    chain_type_kwargs={"prompt": PROMPT}
)
# 提问
question = "文档中提到了哪些主要观点？"
result = qa_chain({"query": question})
print("回答：", result["result"])
print("\n来源文档：")
for doc in result["source_documents"]:
    print(f"- {doc.metadata.get('source', 'Unknown')}")

七、LangChain Expression Language (LCEL)

LCEL 是 LangChain 的新一代语法，提供更简洁的链式调用：

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
# 初始化
llm = ChatOpenAI(model="gpt-3.5-turbo")
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_texts(
    ["LangChain 是一个 LLM 应用框架", "RAG 是检索增强生成"],
    embedding=embeddings
)
# 创建检索器
retriever = vectorstore.as_retriever()
# 创建提示模板
prompt = ChatPromptTemplate.from_template("""基于以下上下文回答问题：
{context}
问题：{question}""")
# 使用 LCEL 构建链
rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)
# 执行
answer = rag_chain.invoke("什么是 LangChain？")
print(answer)

LCEL 的优势

简洁：使用管道操作符 | 连接组件

灵活：支持并行、条件分支等复杂逻辑

可观测：内置追踪和调试功能

类型安全：提供完整的类型提示

八、最佳实践与性能优化

1. 批处理

# 批量处理可以显著提高效率
questions = [
    "什么是人工智能？",
    "什么是机器学习？",
    "什么是深度学习？"
]
# 使用 batch 方法
responses = llm.batch([prompt.format(question=q) for q in questions])

2. 缓存

from langchain.cache import InMemoryCache, SQLiteCache
from langchain.globals import set_llm_cache
# 内存缓存
set_llm_cache(InMemoryCache())
# SQLite 缓存（持久化）
# set_llm_cache(SQLiteCache(database_path=".langchain.db"))
# 现在相同的查询会直接从缓存返回
response1 = llm.invoke("Hello")  # 实际调用 API
response2 = llm.invoke("Hello")  # 从缓存返回

3. 异步处理

import asyncio
from langchain_openai import ChatOpenAI
async def process_batch():
    llm = ChatOpenAI(model="gpt-3.5-turbo")
    
    tasks = [
        llm.ainvoke("写一首关于春天的诗"),
        llm.ainvoke("写一首关于夏天的诗"),
        llm.ainvoke("写一首关于秋天的诗")
    ]
    
    results = await asyncio.gather(*tasks)
    return results
# 运行
results = asyncio.run(process_batch())
for r in results:
    print(r.content)

4. Token 管理

from langchain.callbacks import get_openai_callback
# 追踪 Token 使用量
with get_openai_callback() as cb:
    response = llm.invoke("请解释量子计算的基本原理")
    print(response.content)
    print(f"Total Tokens: {cb.total_tokens}")
    print(f"Prompt Tokens: {cb.prompt_tokens}")
    print(f"Completion Tokens: {cb.completion_tokens}")
    print(f"Total Cost (USD): ${cb.total_cost}")

九、部署与生产化

1. 使用 LangServe

LangServe 是 LangChain 的部署工具，可以快速将链部署为 API：

from langserve import add_routes
from fastapi import FastAPI
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
# 创建 FastAPI 应用
app = FastAPI(
    title="LangChain Server",
    version="1.0",
    description="一个简单的 LangChain API 服务器"
)
# 创建链
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("请解释：{topic}")
chain = prompt | llm
# 添加路由
add_routes(app, chain, path="/explain")
# 运行：uvicorn server:app --reload

2. 流式输出

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo", streaming=True)
for chunk in llm.stream("请写一首关于 AI 的诗"):
    print(chunk.content, end="", flush=True)