跳至主要內容

LangChain 核心概念与架构

HSXISAWD2026/5/14大约 8 分钟AILangChainLLM架构设计

LangChain 核心概念与架构

LangChain 整体架构

六大核心模块

LangChain 的设计围绕六个核心模块展开，每个模块解决 LLM 应用开发中的一个关键问题：

模块英文名解决的问题类比
模型 I/OModel I/O如何统一调用不同模型翻译官
提示词Prompts如何管理和复用提示词模板
输出解析Output Parsers如何让模型输出结构化数据数据校验器
链Chains如何编排多步骤工作流流水线
记忆Memory如何管理对话上下文笔记本
检索Retrieval如何接入外部知识图书馆
代理Agents如何让模型自主决策管家

数据流视角

从数据流的角度理解 LangChain，整个框架就是一个「输入处理 → 模型调用 → 输出处理」的管道：

核心抽象：Runnable 接口

什么是 Runnable

LangChain 从 0.1 版本开始引入了 Runnable 接口，这是整个框架最核心的抽象。几乎所有组件（模型、提示词模板、解析器、链）都实现了这个接口。

Runnable 定义了四个核心方法：

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

# 1. invoke — 单次调用
result = llm.invoke("什么是 Python？")
print(type(result))  # <class 'langchain_core.messages.ai.AIMessage'>

# 2. batch — 批量调用
results = llm.batch(["什么是 Python？", "什么是 Java？"])
print(len(results))  # 2

# 3. stream — 流式输出
for chunk in llm.stream("写一首诗"):
    print(chunk.content, end="")

# 4. ainvoke — 异步调用
import asyncio

async def main():
    result = await llm.ainvoke("什么是 Python？")
    print(result.content)

asyncio.run(main())

Runnable 的组合操作符

Runnable 接口最强大的特性是支持组合（Composition），通过操作符将多个 Runnable 串联起来：

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

# 创建组件
prompt = ChatPromptTemplate.from_template("用一句话解释：{topic}")
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

# 使用管道操作符组合（LCEL 语法）
chain = prompt | llm | parser

# 调用
result = chain.invoke({"topic": "机器学习"})
print(result)
# 输出：机器学习是让计算机通过数据自动学习规律和模式的技术。

Message 消息体系

消息类型

LangChain 使用不同的消息类型来表示对话中的不同角色：

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

# 直接构造消息列表
messages = [
    SystemMessage(content="你是一个资深 Python 开发工程师，擅长简洁清晰地解答技术问题。"),
    HumanMessage(content="Python 中的装饰器是什么？"),
    AIMessage(content="装饰器是 Python 中一种语法糖，它可以在不修改原函数代码的情况下，为函数添加额外的功能。"),
    HumanMessage(content="能写个简单的例子吗？"),
]

# 传入模型
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke(messages)
print(response.content)

多模态消息

LangChain 支持在消息中嵌入图片、文件等多模态内容：

from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

# 图片 URL 方式
message = HumanMessage(
    content=[
        {"type": "text", "text": "这张图片里有什么？"},
        {
            "type": "image_url",
            "image_url": {
                "url": "https://example.com/photo.jpg",
                "detail": "low",  # low / high / auto
            },
        },
    ]
)

response = llm.invoke([message])
print(response.content)

Callbacks 回调系统

回调机制

LangChain 的回调系统允许你在链执行的不同阶段插入自定义逻辑，比如日志记录、进度显示、数据收集等：

from langchain_core.callbacks import BaseCallbackHandler
from langchain_openai import ChatOpenAI

class MyCallbackHandler(BaseCallbackHandler):
    """自定义回调处理器"""

    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"[LLM 开始] 模型: {serialized.get('name', 'unknown')}")
        print(f"[LLM 开始] 输入: {prompts[0][:50]}...")

    def on_llm_end(self, response, **kwargs):
        print(f"[LLM 结束] 输出: {response.generations[0][0].text[:50]}...")

    def on_llm_error(self, error, **kwargs):
        print(f"[LLM 错误] {error}")

    def on_chain_start(self, serialized, inputs, **kwargs):
        print(f"[链开始] {serialized.get('name', 'unnamed')}")

    def on_chain_end(self, outputs, **kwargs):
        print(f"[链结束] 输出: {str(outputs)[:50]}...")


# 使用回调
llm = ChatOpenAI(
    model="gpt-4o-mini",
    callbacks=[MyCallbackHandler()],
)

response = llm.invoke("什么是 Python？")

LCEL（LangChain Expression Language）

什么是 LCEL

LCEL 是 LangChain 的声明式语法，使用管道操作符 | 将组件串联起来。它是构建 LangChain 应用的核心方式：

# 传统写法（不推荐）
prompt_text = f"用一句话解释：{topic}"
response = llm.invoke(prompt_text)
result = parser.parse(response.content)

# LCEL 写法（推荐）
chain = prompt | llm | parser
result = chain.invoke({"topic": "机器学习"})

LCEL 的优势：

特性说明
简洁一行代码定义完整工作流
统一接口所有组件都支持 invoke / stream / batch
流式优先自动支持流式输出，无需额外处理
可追踪自动记录执行过程，方便调试
可组合组件可以自由嵌套和组合

LCEL 组合示例

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

llm = ChatOpenAI(model="gpt-4o-mini")

# 基础管道
basic_chain = (
    ChatPromptTemplate.from_template("告诉我一个关于{topic}的笑话")
    | llm
    | StrOutputParser()
)

# 并行处理
parallel_chain = RunnableParallel(
    joke=ChatPromptTemplate.from_template("讲一个关于{topic}的笑话") | llm | StrOutputParser(),
    fact=ChatPromptTemplate.from_template("说一个关于{topic}的事实") | llm | StrOutputParser(),
)

result = parallel_chain.invoke({"topic": "Python"})
print(result["joke"])
print(result["fact"])

# 带备用方案的链
from langchain_anthropic import ChatAnthropic

fallback_chain = (
    ChatPromptTemplate.from_template("解释{concept}")
    | ChatOpenAI(model="gpt-4o-mini").with_fallbacks(
        [ChatAnthropic(model="claude-sonnet-4-20250514")]
    )
    | StrOutputParser()
)

Schema 数据模型

LangChain 使用一套统一的 Schema 来描述数据，核心类型如下：

from langchain_core.documents import Document

# Document 是 RAG 场景的核心数据结构
doc = Document(
    page_content="LangChain 是一个用于开发 LLM 应用的框架。",
    metadata={
        "source": "langchain_docs.pdf",
        "page": 1,
        "author": "LangChain Team",
    },
)

print(doc.page_content)  # 文档内容
print(doc.metadata)      # 元数据（用于检索和过滤）

本章小结

核心要点

  1. 六大模块：Model I/O、Prompts、Output Parsers、Chains、Memory、Retrieval、Agents 构成 LangChain 的完整功能体系
  2. Runnable 接口：所有组件的统一抽象，支持 invoke / stream / batch 三种调用方式
  3. LCEL 语法：用管道操作符 | 声明式地组合组件，简洁且强大
  4. Message 体系：System / Human / AI / Tool 四种消息类型，构建对话的基石
  5. Callback 系统：全链路回调机制，实现日志、追踪、调试等横切关注点

概念速查表

概念一句话说明
Runnable所有组件的统一接口
LCEL用 | 连接组件的声明式语法
Message对话消息的标准化结构
Document文档数据的标准化结构
Callback链执行过程的钩子函数
Chain多步骤工作流的编排
Agent能自主决策和调用工具的模型
Default footer
Copyright © 2026 HSXISAWD