LangChain 框架深入

Chain / Agent / Tool / Memory / Callback、LCEL 表达式语言、LangGraph 图工作流

LangChain 架构总览

LangChain 是一个面向大语言模型（LLM）应用开发的框架，其核心设计目标是通过模块化组合降低 LLM 应用开发复杂度。框架由六大核心模块构成：

模块	职责	核心抽象
Model I/O	管理模型输入输出	BaseLanguageModel、PromptTemplate、OutputParser
Retrieval	外部知识检索	DocumentLoader、TextSplitter、VectorStore、Retriever
Chain	构建调用序列	LLMChain、SequentialChain、RouterChain
Agent	自主决策与行动	Agent、AgentExecutor、Tool
Memory	对话状态持久化	BaseMemory、ConversationBufferMemory、SummaryMemory
Callback	事件观测与拦截	BaseCallbackHandler、CallbackManager

这六个模块既可独立使用，也可通过 LCEL 表达式语言组合为复杂的执行流水线。更高阶的 LangGraph 则在 Chain 的基础上引入图结构，支持循环、分支和条件跳转。

Chain 链系统

Chain 是 LangChain 中最核心的组合单元，它将 Prompt、Model 和 OutputParser 封装为可复用的调用链。

LLMChain

LLMChain 是最基础的链类型，封装了 "PromptTemplate + LLM + OutputParser" 的标准调用流程：

python

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

prompt = PromptTemplate(
    template="请用{tone}的语气回答：{question}",
    input_variables=["tone", "question"]
)
chain = LLMChain(llm=ChatOpenAI(), prompt=prompt)
result = chain.run(tone="正式", question="什么是RAG？")

其源码核心逻辑在 __call__ 方法中依次执行：格式化 Prompt -> 调用 LLM -> 解析输出。Chain 内部通过 CallbackManager 在每一步触发对应事件。

SequentialChain

SequentialChain 将多个 Chain 串联为执行管道，前一个 Chain 的输出作为后一个 Chain 的输入：

python

from langchain.chains import SequentialChain

chain1 = LLMChain(llm=llm, prompt=prompt1, output_key="summary")
chain2 = LLMChain(llm=llm, prompt=prompt2, output_key="translation")
pipeline = SequentialChain(chains=[chain1, chain2], input_variables=["text"])
pipeline.run(text="...")

RouterChain 与 TransformChain

RouterChain：根据输入内容动态选择下游子链。内部维护一个路由条件列表，匹配到对应条件后转发给目标链执行。适用于多意图分类场景。
TransformChain：允许在链间插入自定义数据转换逻辑，对中间结果做清洗、过滤或格式转换，不涉及 LLM 调用。

Agent 与 Tool

Agent 系统赋予 LLM 调用外部工具的能力，是 LLM 从"对话模型"迈向"行动模型"的关键机制。

Agent 类型

Agent 类型	底层模型要求	特点
OpenAI Tools	GPT-4 / GPT-4o 等	原生 function calling，稳定高效
ReAct	通用	思考-行动-观察循环，适用任何 LLM
Structured Chat	通用	支持结构化参数传递的工具调用
XML Agent	通用	使用 XML 格式组织工具调用指令

Tool 定义与注册

Tool 是 Agent 能力的原子单元。LangChain 提供了 @tool 装饰器将任意 Python 函数注册为工具：

python

from langchain.tools import tool

@tool
def search_weather(city: str) -> str:
    """查询指定城市的天气信息"""
    return f"{city}今日天气：晴，25-30°C"

tools = [search_weather]

每个 Tool 包含 name、description、args_schema 三个核心属性。description 会注入 Agent 的 System Prompt，指导 LLM 在何时调用该工具。

AgentExecutor 执行循环

AgentExecutor 实现了 Agent 的标准执行循环：

接收用户输入，调用 Agent（LLM）生成下一步动作
若 LLM 返回 Final Answer，结束循环
若 LLM 返回 Tool Call（工具名 + 参数），执行对应工具
将工具执行结果（Observation）注入上下文
重复步骤 1，直到达到最大迭代次数

python

from langchain.agents import create_react_agent, AgentExecutor

agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=5)
executor.invoke({"input": "北京今天天气怎么样？"})

max_iterations 防止死循环，early_stopping_method 控制超时后的兜底策略。

Memory 记忆系统

Memory 模块解决 LLM 的无状态问题，为对话注入历史上下文。

记忆类型对比

记忆类型	存储方式	适用场景	缺点
ConversationBufferMemory	保存完整对话列表	简短对话	Token 消耗随轮次线性增长
ConversationBufferWindowMemory	只保留最近 k 轮	窗口对话	丢失早期关键信息
SummaryMemory	LLM 自动摘要历史	长对话	额外 LLM 调用开销
VectorStoreMemory	向量检索 + 语义匹配	超长对话/知识库	需要 Embedding 模型
CombinedMemory	组合多种记忆策略	复杂场景	配置复杂

实现原理

Memory 通过 load_memory_variables 加载历史到 Prompt 上下文，通过 save_context 在每轮对话结束后记录输入输出：

python

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(return_messages=True)
memory.save_context({"input": "你好"}, {"output": "你好！有什么可以帮助你的？"})
memory.save_context({"input": "RAG是什么？"}, {"output": "检索增强生成..."})

# 加载到上下文
variables = memory.load_memory_variables({})
print(variables["history"])  # 包含两轮对话的 messages

对于多会话场景，可以使用 ConversationSummaryMemory 配合 chat_memory 参数连接到外部数据库（如 Redis、PostgreSQL），实现持久化存储。

Callback 事件系统

Callback 模块提供了对 LLM 应用运行过程的细粒度观测能力。

事件类型

Callback 覆盖了链执行全生命周期的关键节点：

事件方法	触发时机	典型用途
on_llm_start/end	LLM 调用前后	记录 Token 用量、计算延迟
on_chain_start/end	Chain 执行前后	日志链路追踪
on_tool_start/end	Tool 调用前后	审计工具调用
on_agent_action/finish	Agent 动作/结束	监控 Agent 决策过程
on_retriever_start/end	检索器调用前后	评估检索质量
on_chat_model_start	Chat 模型调用	流式输出拦截

自定义 Handler

python

from langchain.callbacks.base import BaseCallbackHandler

class LogCallbackHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"[LLM 开始] prompts: {prompts}")

    def on_llm_end(self, response, **kwargs):
        print(f"[LLM 结束] token 用量: {response.llm_output}")

    def on_chain_error(self, error, **kwargs):
        print(f"[Chain 异常] {error}")

通过 callbacks 参数传入 Chain 或 Agent，即可启用事件监听。多个 Handler 可同时注册，互不干扰。

LCEL 表达式语言

LCEL（LangChain Expression Language）是 LangChain 推出的声明式编程范式，基于统一的 Runnable 接口实现链式组合。

Runnable 接口

所有 LCEL 组件都实现 Runnable 协议，提供 invoke、batch、stream、astream 等统一调用方法：

组件	功能	等效运算符
RunnableSequence	顺序执行	`pipe` / `
RunnableParallel	并行执行	字典 `{}`
RunnablePassthrough	透传输入	`itemgetter`
RunnableLambda	自定义函数	无
RunnableBranch	条件路由	无

管道操作符链式调用

| 操作符是 LCEL 的语法核心，将多个 Runnable 串联为一条执行流水线：

python

from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

# LCEL 链式调用示例
prompt = ChatPromptTemplate.from_template("用{language}回答：{question}")
model = ChatOpenAI(model="gpt-4o")

chain = (
    RunnablePassthrough.assign(
        language=lambda x: x.get("language", "中文")
    )
    | prompt
    | model
)

result = chain.invoke({"question": "什么是向量数据库？"})

上述链的执行顺序：RunnablePassthrough 注入默认参数 -> PromptTemplate 格式化 -> LLM 生成。这种管道风格使得代码可读性大幅提升，且天然支持流式输出。

RunnableParallel 并行

通过字典语法实现多路并行执行，结果合并后传递给下游：

python

from langchain_core.runnables import RunnableParallel

parallel = RunnableParallel(
    summary=summary_chain,
    keywords=keyword_chain,
)
combined = parallel | merge_chain

LangGraph 图工作流

LangGraph 在 Chain 的单向流水线基础上引入图结构，支持循环、条件分支和有状态的工作流编排。

核心概念

概念	说明	类比
StateGraph	带状态管理的图结构	有限状态机
Node	图中的执行节点	Chain / Function
Edge	节点间的连接	数据流
Conditional Edge	带条件的边	if/else 分支

构建图工作流

python

from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal

class AgentState(TypedDict):
    messages: list
    next_action: str

# 定义节点函数
def call_model(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

def should_continue(state: AgentState) -> Literal["tools", "end"]:
    last_msg = state["messages"][-1]
    return "tools" if last_msg.tool_calls else "end"

def call_tool(state: AgentState) -> AgentState:
    tool_result = execute_tool(state["messages"][-1].tool_calls)
    return {"messages": state["messages"] + [tool_result]}

# 构建图
graph = StateGraph(AgentState)

graph.add_node("model", call_model)
graph.add_node("tools", call_tool)

graph.set_entry_point("model")
graph.add_conditional_edges(
    "model",
    should_continue,
    {"tools": "tools", "end": END}
)
graph.add_edge("tools", "model")

app = graph.compile()

LangGraph 与 Chain 的对比

维度	Chain	LangGraph
拓扑结构	线性/树状	有向图
循环支持	不支持（需外部循环）	原生支持
状态管理	隐式传递	显式 State 对象
条件分支	RouterChain	Conditional Edge
适用场景	固定流程	复杂对话、多步骤推理
调试难度	低	中

LangGraph 适合需要循环推理的 Agent 场景（如 ReAct 的思考-行动-观察循环），而传统 Chain 适合确定性的串行任务。LangGraph 的 StateGraph 在每个节点执行后更新状态对象，使得整个执行链路的状态变更可追溯、可回放。

总结

LangChain 从 Chain 的线性组合到 LCEL 的声明式管道，再到 LangGraph 的图工作流，演进路径体现了 LLM 应用开发从简单到复杂、从确定到动态的趋势。实际项目中应根据需求层次选择合适抽象：固定流程用 Chain，多工具调用用 Agent + LCEL，复杂有状态工作流用 LangGraph。

LangChain 框架深入 ​

LangChain 架构总览 ​

Chain 链系统 ​

LLMChain ​

SequentialChain ​

RouterChain 与 TransformChain ​

Agent 与 Tool ​

Agent 类型 ​

Tool 定义与注册 ​

AgentExecutor 执行循环 ​

Memory 记忆系统 ​

记忆类型对比 ​

实现原理 ​

Callback 事件系统 ​

事件类型 ​

自定义 Handler ​

LCEL 表达式语言 ​

Runnable 接口 ​

管道操作符链式调用 ​

RunnableParallel 并行 ​

LangGraph 图工作流 ​

核心概念 ​

构建图工作流 ​

LangGraph 与 Chain 的对比 ​

总结 ​

LangChain 框架深入

LangChain 架构总览

Chain 链系统

LLMChain

SequentialChain

RouterChain 与 TransformChain

Agent 与 Tool

Agent 类型

Tool 定义与注册

AgentExecutor 执行循环

Memory 记忆系统

记忆类型对比

实现原理

Callback 事件系统

事件类型

自定义 Handler

LCEL 表达式语言

Runnable 接口

管道操作符链式调用

RunnableParallel 并行

LangGraph 图工作流

核心概念

构建图工作流

LangGraph 与 Chain 的对比

总结