理解 LangChain：大模型应用开发的核心抽象

一、引言

随着 GPT-4、Claude、通义千问等大语言模型（LLM）的爆发式发展，开发者面临一个共同的挑战：如何高效地将大模型集成到实际应用中？直接调用 API 看似简单，但一旦涉及多轮对话、知识库检索、工具调用等复杂场景，代码很快就会变得臃肿且难以维护。

LangChain 正是为解决这一问题而生的框架。它提供了一套核心抽象，将大模型应用开发中的常见模式标准化，让你像搭积木一样构建复杂的 LLM 应用。本文将从零开始，逐一拆解 LangChain 的核心抽象，并通过 Python 代码示例让你快速上手。

---

二、Models：统一的大模型接口

LangChain 的第一层抽象是模型（Models）。它将各种大模型封装成统一的接口，让你可以无缝切换不同的模型提供商。

2.1 LLM 模型：文本进，文本出

# 安装依赖：pip install langchain langchain-openai
from langchain_openai import ChatOpenAI

# ----- 1. 初始化聊天模型 -----
# ChatOpenAI 封装了 OpenAI 的 GPT 系列模型
# model 参数指定模型版本，temperature 控制输出的随机性（0=确定性，1=创造性）
llm = ChatOpenAI(
    model="gpt-3.5-turbo",  # 模型名称，也可以换成 gpt-4、gpt-4o 等
    temperature=0.7,        # 温度参数：越低越保守，越高越有创造性
    max_tokens=512,         # 限制单次回复的最大 token 数，控制成本
    api_key="your-api-key"  # OpenAI API 密钥，建议通过环境变量管理
)

# ----- 2. 调用模型：invoke 方法 -----
# invoke 是 LangChain 中最核心的调用方法，几乎所有组件都使用它
# 传入一个字符串，模型返回一个 AIMessage 对象
response = llm.invoke("请用一句话介绍 Python 编程语言。")
print(response.content)  # 获取回复的文本内容

2.2 消息类型：构建多轮对话的基础

LangChain 将对话中的每条消息抽象为不同的消息类型，这是构建聊天应用的核心。

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# ----- 消息类型详解 -----
# LangChain 定义了三种核心消息类型，对应对话中的不同角色：

# 1. SystemMessage：系统提示词，用于设定 AI 的角色和行为规则
#    这条消息不会被展示给用户，但对模型的回复风格影响极大
system_msg = SystemMessage(
    content="你是一个经验丰富的 Python 导师，回答问题时要用通俗易懂的语言，"
            "并且每次都要给出一个代码示例。"
)

# 2. HumanMessage：用户发送的消息
#    代表人类用户的一次输入
human_msg = HumanMessage(
    content="请问什么是列表推导式？"
)

# 3. AIMessage：AI 返回的消息
#    代表模型的一次回复
#    这里我们手动创建一个示例，实际使用时由模型生成

# ----- 将消息列表传入模型，实现上下文对话 -----
# 模型会根据消息列表中的历史记录来理解上下文，生成连贯的回复
messages = [system_msg, human_msg]  # 消息列表，按对话顺序排列
response = llm.invoke(messages)
print(response.content)

---

三、Prompt Templates：动态提示词模板

硬编码提示词在实际开发中几乎不可行。LangChain 提供了提示词模板（Prompt Templates），让你可以动态生成提示词，实现参数化。

3.1 基础提示词模板

from langchain_core.prompts import ChatPromptTemplate

# ----- 定义提示词模板 -----
# 模板中的 {变量名} 是占位符，调用时会被实际值替换
# 这种方式让提示词可以复用，不用每次都拼接字符串
prompt_template = ChatPromptTemplate.from_messages([
    # SystemMessage 模板：设定 AI 角色，{role} 是动态变量
    ("system", "你是一个{role}，请用{style}的风格回答问题。"),
    # HumanMessage 模板：{question} 是用户的问题
    ("human", "{question}")
])

# ----- 填充模板：invoke 方法传入变量值 -----
# 模板的 invoke 返回一个填充后的消息列表，可以直接传给模型
filled_prompt = prompt_template.invoke({
    "role": "资深的 Python 后端工程师",  # 替换 {role}
    "style": "专业且幽默",               # 替换 {style}
    "question": "如何优化 Django 的数据库查询性能？"  # 替换 {question}
})

# 查看填充后的消息内容
print("填充后的提示词：")
for msg in filled_prompt.messages:
    print(f"[{msg.__class__.__name__}] {msg.content}")

3.2 少样本提示词模板（Few-Shot Prompting）

from langchain_core.prompts import FewShotChatMessagePromptTemplate

# ----- 准备少样本示例 -----
# 通过在提示词中提供几个问答示例，让模型学会你期望的输出格式
# 这在文本分类、格式转换等任务中非常有效
examples = [
    {
        "input": "今天天气真好，心情很愉快！",
        "output": "情感：正面 | 置信度：高"
    },
    {
        "input": "快递又送错了，客服态度还特别差。",
        "output": "情感：负面 | 置信度：高"
    },
    {
        "input": "明天去超市买点东西。",
        "output": "情感：中性 | 置信度：中"
    },
]

# ----- 创建少样本模板 -----
# 将示例嵌入到提示词中，让模型参考示例的格式进行回复
example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),    # 示例中的用户输入
    ("ai", "{output}")       # 示例中的 AI 输出
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,  # 示例的展示格式
    examples=examples,              # 示例列表
)

# 将少样本模板与系统提示词组合成完整的提示词
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个情感分析助手。请分析用户输入的文本情感，"
               "输出格式为：情感：[正面/负面/中性] | 置信度：[高/中/低]"),
    few_shot_prompt,  # 插入少样本示例
    ("human", "{user_input}")  # 实际要分析的用户输入
])

# 测试效果
filled = final_prompt.invoke({"user_input": "这部电影意外地好看，推荐！"})
response = llm.invoke(filled)
print(response.content)

---

四、Chains：串联多个组件

Chain（链） 是 LangChain 最核心的抽象。它将多个组件串联成一个可执行的流水线，前一个组件的输出自动成为后一个组件的输入。

4.1 基础链：LCEL 表达式

LangChain 推荐使用 LCEL（LangChain Expression Language） 来构建链，语法简洁直观。

from langchain_core.output_parsers import StrOutputParser

# ----- 使用 LCEL 构建一条简单的链 -----
# 管道运算符 | 模拟 Unix 管道，将前一个组件的输出传给后一个组件
# 这种方式让代码逻辑清晰，易于阅读和调试

# StrOutputParser：将 AIMessage 对象转换为纯文本字符串
parser = StrOutputParser()

# 构建链：prompt_template -> llm -> parser
# 解读：输入经过模板填充 -> 传给大模型 -> 解析输出为纯文本
chain = prompt_template | llm | parser

# 调用链：invoke 方法传入字典，链会自动依次执行每个环节
result = chain.invoke({
    "role": "美食评论家",
    "style": "文艺且细腻",
    "question": "如何评价一碗正宗的兰州拉面？"
})
print(result)

4.2 顺序链：串联多个 LLM 调用

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# ----- 场景：先生成大纲，再根据大纲写正文 -----
# 这种多步骤的 LLM 调用在内容生成、文档撰写等场景中非常常见

# 第一步：生成大纲的提示词模板
outline_prompt = ChatPromptTemplate.from_template(
    "你是一个{role}。请为以下主题写一个详细的文章大纲，"
    "包含 3-5 个主要章节：\n主题：{topic}"
)

# 第二步：根据大纲写正文的提示词模板
# {outline} 是上一步生成的输出，会作为变量传入
article_prompt = ChatPromptTemplate.from_template(
    "你是一个{role}。请根据以下大纲，撰写一篇完整的文章。"
    "要求语言流畅，每个章节控制在 200 字左右。\n\n大纲：\n{outline}"
)

# ----- 定义辅助函数，串联两个步骤 -----
# 在 LCEL 中，可以使用 RunnableLambda 或自定义函数来处理中间步骤
def generate_article(topic: str, role: str) -> str:
    """
    两步生成文章：
    1. 先生成大纲（outline）
    2. 再根据大纲生成正文（article）
    """
    # 步骤1：生成大纲
    print(">>> 正在生成大纲...")
    outline = (outline_prompt | llm | StrOutputParser()).invoke({
        "role": role,
        "topic": topic
    })
    print(f"大纲内容：\n{outline}\n")

    # 步骤2：根据大纲生成正文
    print(">>> 正在撰写正文...")
    article = (article_prompt | llm | StrOutputParser()).invoke({
        "role": role,
        "outline": outline
    })
    return article


# 测试顺序链
result = generate_article(
    topic="Python 异步编程入门",
    role="Python 技术博主"
)
print("=" * 50)
print(result)

---

五、Memory：让对话拥有记忆

默认情况下，大模型是无状态的——每次调用都是独立的。Memory（记忆） 抽象让模型能够记住历史对话，实现连贯的多轮交互。

from langchain_core.runnables import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# ----- 创建对话历史存储 -----
# ChatMessageHistory 负责存储和读取对话记录
# 实际项目中通常会换成 Redis 或数据库存储，以支持持久化
store = {}  # 会话存储字典，key 是会话 ID，value 是对话历史

def get_session_history(session_id: str):
    """
    获取或创建指定会话的对话历史
    这是一个工厂函数，LangChain 在每次调用时用它来获取历史记录
    """
    if session_id not in store:
        # 如果是新会话，创建一个空的对话历史
        store[session_id] = ChatMessageHistory()
    return store[session_id]


# ----- 构建带有记忆的提示词模板 -----
# 这个模板会接收 {history} 变量，由 LangChain 自动填充历史对话
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手，请记住用户之前说过的话。"),
    # {history} 占位符：LangChain 会自动将历史消息插入到这里
    ("placeholder", "{history}"),
    # {input} 占位符：当前用户的最新输入
    ("human", "{input}")
])

# 构建基础链
chain = chat_prompt | llm | StrOutputParser()

# ----- 将链包装为带记忆的链 -----
# RunnableWithMessageHistory 是 LangChain 提供的记忆包装器
# 它会自动在每次调用时读取历史、执行链、保存新消息
chain_with_memory = RunnableWithMessageHistory(
    chain,                      # 原始链
    get_session_history,        # 获取历史记录的函数
    input_messages_key="input", # 输入字典中 "用户消息" 的键名
    history_messages_key="history"  # 提示词中 "历史消息" 的变量名
)

# ----- 模拟多轮对话 -----
session_id = "user_123"  # 模拟一个会话 ID

# 第一轮对话：告诉模型你的名字
response1 = chain_with_memory.invoke(
    {"input": "你好，我叫小明，我喜欢打篮球。"},
    config={"configurable": {"session_id": session_id}}
)
print(f"AI: {response1}")

# 第二轮对话：模型应该能记住你的名字和爱好
response2 = chain_with_memory.invoke(
    {"input": "你还记得我的名字和爱好吗？"},
    config={"configurable": {"session_id": session_id}}
)
print(f"AI: {response2}")

# 验证：查看存储中的对话历史
print("\n--- 对话历史记录 ---")
for msg in store[session_id].messages:
    print(f"[{msg.__class__.__name__}] {msg.content}")

---

六、RAG：检索增强生成

RAG（Retrieval-Augmented Generation） 是 LangChain 最强大的应用场景之一。它让大模型能够基于外部知识库回答问题，而不是仅依赖训练数据。

6.1 文档加载与分割

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# ----- 1. 加载文档 -----
# LangChain 支持加载多种格式：TXT、PDF、Markdown、网页等
# TextLoader 是最基础的加载器，适合纯文本文件
loader = TextLoader("knowledge.txt", encoding="utf-8")
documents = loader.load()  # 返回一个 Document 对象列表
print(f"加载了 {len(documents)} 个文档")

# ----- 2. 分割文档 -----
# 大模型有 token 限制，所以需要将长文档切分成小片段（chunk）
# RecursiveCharacterTextSplitter 按段落 -> 句子 -> 字符的顺序递归分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,       # 每个片段最多 500 个字符
    chunk_overlap=50,     # 相邻片段之间重叠 50 个字符（保持上下文连贯）
    separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""]  # 分割优先级
)

# split_documents 将文档列表切分成多个片段
chunks = text_splitter.split_documents(documents)
print(f"切分后共 {len(chunks)} 个片段")

# 打印前两个片段，看看切分效果
for i, chunk in enumerate(chunks[:2]):
    print(f"\n--- 片段 {i+1} ---")
    print(chunk.page_content[:200] + "...")  # 只打印前 200 个字符

6.2 向量化与存储

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# ----- 3. 文本向量化（Embedding）-----
# 将文本转换为向量，语义相近的文本向量距离也近
# 这是实现语义搜索的关键——不再依赖关键词匹配，而是理解语义
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",  # OpenAI 的嵌入模型，性价比高
    api_key="your-api-key"
)

# ----- 4. 创建向量数据库 -----
# Chroma 是一个轻量级的向量数据库，适合本地开发和小规模项目
# 生产环境可以换成 Pinecone、Weaviate、Milvus 等
vector_store = Chroma.from_documents(
    documents=chunks,          # 文档片段列表
    embedding=embeddings,      # 嵌入模型
    persist_directory="./chroma_db"  # 持久化存储路径，重启后数据不丢失
)

# 查看向量数据库中的文档数量
print(f"向量数据库中存储了 {vector_store._collection.count()} 个文档片段")

6.3 构建 RAG 问答链

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# ----- 5. 构建检索器（Retriever）-----
# 检索器负责根据用户问题，从向量数据库中找出最相关的文档片段
# k 参数控制返回多少个最相关的片段
retriever = vector_store.as_retriever(
    search_kwargs={"k": 3}  # 每次检索返回最相关的 3 个片段
)

# ----- 6. 定义 RAG 提示词模板 -----
# 关键：将检索到的文档片段作为上下文注入到提示词中
# 这样模型就能基于外部知识来回答问题，而不是凭空编造
rag_prompt = ChatPromptTemplate.from_template("""
你是一个基于知识库的问答助手。请严格根据以下提供的上下文信息来回答问题。
如果上下文中没有相关信息，请直接说"我不知道"，不要编造答案。

上下文信息：
{context}

用户问题：{question}

回答：
""")

# ----- 7. 构建 RAG 链 -----
# 使用 LCEL 将所有组件串联起来
def format_docs(docs):
    """
    将检索到的文档片段格式化为纯文本
    用换行符分隔不同片段，方便模型阅读
    """
    return "\n\n".join([doc.page_content for doc in docs])

rag_chain = (
    # 第一步：构建输入
    # RunnablePassthrough 将用户问题原样传递
    # retriever 同时检索相关文档，并通过 format_docs 格式化
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    # 第二步：填充提示词模板
    | rag_prompt
    # 第三步：调用大模型
    | llm
    # 第四步：解析输出为纯文本
    | StrOutputParser()
)

# ----- 8. 测试 RAG 问答 -----
question = "什么是 LangChain？它有哪些核心组件？"
answer = rag_chain.invoke(question)
print(f"问题：{question}")
print(f"回答：{answer}")

---

七、Agents：让模型自主调用工具

Agent（智能体） 是 LangChain 中最高级的抽象。它让模型能够自主决定使用哪些工具、按什么顺序使用，从而实现复杂任务的自动化。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate

# ----- 1. 定义工具（Tools）-----
# 工具就是模型可以调用的函数，用来完成特定任务
# 使用 @tool 装饰器将普通 Python 函数注册为 LangChain 工具

@tool
def calculate(expression: str) -> str:
    """
    计算数学表达式的结果。
    接收一个数学表达式字符串（如 "2+3*4"），返回计算结果。
    注意：仅支持加减乘除运算。
    """
    try:
        # eval 计算表达式，返回结果
        result = eval(expression)
        return f"计算结果：{expression} = {result}"
    except Exception as e:
        return f"计算错误：{e}"


@tool
def get_current_weather(city: str) -> str:
    """
    查询指定城市的当前天气。
    接收城市名称（中文），返回该城市的天气信息。
    """
    # 模拟天气数据，实际项目中应调用天气 API
    weather_data = {
        "北京": "晴天，25°C，湿度 40%",
        "上海": "多云，28°C，湿度 65%",
        "深圳": "阵雨，30°C，湿度 80%",
    }
    return weather_data.get(city, f"抱歉，暂时没有 {city} 的天气数据。")


@tool
def get_current_time() -> str:
    """获取当前的日期和时间。"""
    from datetime import datetime
    now = datetime.now()
    return f"当前时间：{now.strftime('%Y年%m月%d日 %H:%M:%S')}"


# 将工具注册到工具列表中
tools = [calculate, get_current_weather, get_current_time]

# ----- 2. 创建 Agent 提示词 -----
# Agent 提示词需要告诉模型它有哪些工具可用，以及如何使用
agent_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能助手，可以调用工具来完成任务。"
               "当需要计算时使用 calculate 工具，"
               "查天气时使用 get_current_weather 工具，"
               "查时间时使用 get_current_time 工具。"
               "如果不需要工具，直接回答即可。"),
    ("human", "{input}"),
    # {agent_scratchpad} 是 Agent 的内部工作区，记录工具调用过程
    ("placeholder", "{agent_scratchpad}")
])

# ----- 3. 创建 Agent 和 Executor -----
# create_tool_calling_agent：创建支持函数调用的 Agent
# AgentExecutor：负责执行 Agent 的决策循环（思考 -> 调用工具 -> 观察结果 -> 再思考）
agent = create_tool_calling_agent(llm, tools, agent_prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # 开启详细日志，方便观察 Agent 的思考过程
    max_iterations=5,  # 最多执行 5 轮工具调用，防止无限循环
)

# ----- 4. 测试 Agent -----
# Agent 会自动判断需要调用哪些工具，并组合使用
print("=" * 60)
print("测试 1：数学计算")
result = agent_executor.invoke({"input": "帮我计算 (15 + 25) * 3 的结果"})
print(result["output"])

print("\n" + "=" * 60)
print("测试 2：天气查询")
result = agent_executor.invoke({"input": "深圳今天天气怎么样？"})
print(result["output"])

print("\n" + "=" * 60)
print("测试 3：多工具组合调用")
result = agent_executor.invoke({
    "input": "北京的温度是多少？如果温度低于 30 度，就帮我算一下 25 度加 10 度等于多少。"
})
print(result["output"])

---

八、LangChain 核心抽象关系图

┌─────────────────────────────────────────────────────┐
│                     Agent                            │
│  ┌─────────┐  ┌──────────┐  ┌───────────────────┐  │
│  │  Memory  │  │  Tools   │  │  Prompt Template   │  │
│  └─────────┘  └──────────┘  └───────────────────┘  │
│         │           │                │               │
│         └───────────┴────────────────┘               │
│                      │                               │
│               ┌──────▼──────┐                        │
│               │   Chain      │                       │
│               └──────┬──────┘                        │
│                      │                               │
│               ┌──────▼──────┐                        │
│               │   Model      │                       │
│               └─────────────┘                        │
└─────────────────────────────────────────────────────┘

• Model：最底层，负责与大模型通信
• Prompt Template：参数化提示词，提升复用性
• Chain：串联多个组件，形成执行流水线
• Memory：为对话提供上下文记忆
• Tools：让模型调用外部函数
• RAG：基于外部知识库增强回答
• Agent：最高层，自主决策调用工具和链

---

九、总结

LangChain 通过以下核心抽象，将大模型应用开发简化为"搭积木"的过程：

抽象	作用	一句话理解
Models	封装大模型调用	统一的"模型接口"
Prompt Templates	动态生成提示词	参数化的"提示词"
Chains	串联多个组件	流水线
Memory	记住对话历史	对话的"记忆"
RAG	基于知识库回答	给模型"查资料"
Agents	自主调用工具	模型的"手和脚"

学习建议：

1. 先掌握 Chain 和 Prompt Template —— 这是使用 LangChain 的基本功
2. 再学习 RAG —— 这是目前最实用的 LLM 应用模式
3. 最后攻克 Agent —— 理解 Agent 的决策循环后，你就打开了智能体开发的大门

LangChain 的生态还在快速演进中（如 LangGraph 用于有状态 Agent、LangSmith 用于调试监控），但核心抽象始终保持稳定。掌握这些基础概念，你就能从容应对各种 LLM 应用开发场景。

---

本文代码基于 LangChain 0.2+ 版本编写，建议使用 pip install langchain langchain-openai langchain-community langchain-text-splitters chromadb 安装相关依赖。