专栏第7篇：第六篇我们讲了 MCP 协议如何让工具调用标准化。但 MCP 解决的是"工具怎么被调用"的问题，Agent 框架解决的是"怎么把 LLM、Prompt、工具、记忆串起来"的问题。今天我们来拆解 LangChain 最核心的概念——Chain，以及它背后的 LCEL（LangChain Expression Language）设计哲学。

目录

一、Chain 到底是什么？

第一次看 LangChain 文档时，我以为 Chain 就是"链"——像链表一样把东西串起来。结果写代码时踩了一堆坑，才发现 Chain 的本质是"数据流水线"。

1.1 从概念到代码

普通调用方式：
  prompt = f"解释什么是{concept}"
  response = llm(prompt)
  result = response.content
  # 每一步都手动管理输入输出

Chain 方式：
  chain = prompt_template | llm | output_parser
  result = chain.invoke({"concept": "区块链"})
  # 声明式流水线，数据自动流转

核心区别：普通方式是"命令式编程"（告诉程序怎么做），Chain 是"声明式编程"（告诉程序想要什么结果）。

1.2 Chain 的四个标准接口

每个 Chain（或者说每个 Runnable 组件）都支持四个调用方式：

方法 作用 适用场景
invoke(input) 单次同步调用 常规交互
stream(input) 流式输出 需要实时显示
batch(inputs) 批量处理 一次处理多条
ainvoke(input) 异步调用 高并发场景 
chain = prompt | llm | parser

# 四种调用方式
result = chain.invoke({"topic": "AI"})           # 单次
for chunk in chain.stream({"topic": "AI"}):      # 流式
results = chain.batch([{"topic": "A"}, {"topic": "B"}])  # 批量
result = await chain.ainvoke({"topic": "AI"})    # 异步

二、LCEL：用管道符 | 写 Chain

LCEL（LangChain Expression Language）是 LangChain 的核心语法创新。它的设计灵感来自 Unix 管道的 | 操作符。

2.1 LCEL 是什么？

LCEL = LangChain Expression Language，直译是"LangChain 表达式语言"。

为什么叫"语言"而不是"API"？因为它有一套自己的语法规则：

prompt | llm | parser

| 就是 LCEL 的核心语法符号，规则很简单：左边组件的输出，自动变成右边组件的输入。

类比理解：

  • SQL 是声明式查询语言——你说"要什么数据"，数据库帮你算
  • LCEL 是声明式流水线语言——你说"数据怎么流"，框架帮你连

背后的设计哲学：

  1. 组合优于继承：不用写类继承，用 | 把组件拼起来
  2. 统一接口：所有组件都实现 Runnable，所以可以任意拼接
  3. 延迟执行：chain = A | B 只是定义流水线，真正执行是 chain.invoke() 时

2.2 管道符的本质

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

# 三个独立组件
prompt = ChatPromptTemplate.from_template("用一句话解释{concept}")
llm = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()

# 用 | 连接成 Chain
chain = prompt | llm | parser

# 等价于：
# result = parser.parse(llm.invoke(prompt.format(concept="区块链")))

| 做的事情很简单：把左边组件的输出，作为右边组件的输入。

2.3 数据流向图

输入字典 concept=区块链

PromptTemplate

格式化后的Prompt

LLM

LLM输出对象 AIMessage

OutputParser

最终输出 纯文本

2.4 为什么不用函数嵌套？

# 函数嵌套方式（难以阅读）
result = parser.parse(llm.invoke(prompt.format(concept="区块链")))

# 管道方式（一目了然）
chain = prompt | llm | parser
result = chain.invoke({"concept": "区块链"})

管道方式的优势：

  1. 可读性：数据流向从左到右，符合阅读习惯
  2. 可组合性：随时可以在中间插入新组件
  3. 可调试性：可以单独测试每个组件

三、PromptTemplate：不只是字符串拼接

PromptTemplate 是 Chain 的起点，它的作用远不止是"把变量插进字符串"。

3.1 基础模板

from langchain_core.prompts import ChatPromptTemplate

# 单变量模板
prompt = ChatPromptTemplate.from_template("用{style}风格解释{concept}")

# 多变量输入
result = prompt.invoke({
    "style": "浪漫主义",
    "concept": "爱情"
})
# 输出：ChatPromptValue（包含 System + Human 消息）

3.2 ChatPromptTemplate：区分角色

from langchain_core.prompts import ChatPromptTemplate

# 定义多角色对话模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位{role}，擅长用{style}风格回答"),
    ("human", "请解释{concept}")
])

# 自动构建消息列表
messages = prompt.invoke({
    "role": "诗人",
    "style": "浪漫主义",
    "concept": "爱情"
})
# 输出：[SystemMessage, HumanMessage]

关键点：ChatPromptTemplate 不是返回字符串，而是返回结构化的消息列表。这保证了 LLM 能正确区分 System 指令和 User 输入。

3.3 FewShotPromptTemplate：给 LLM 举例子

from langchain_core.prompts import FewShotChatMessagePromptTemplate

# 定义示例
examples = [
    {"input": "开心", "output": "喜悦如春日暖阳，融化了冬日的寒冰"},
    {"input": "难过", "output": "悲伤似秋雨连绵，打湿了离人的心"}
]

# 构建 Few-Shot 模板
example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}")
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples
)

# 组合到完整 Prompt
final_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个情感丰富的诗人"),
    few_shot_prompt,
    ("human", "{input}")
])

Few-Shot 的价值：通过示例告诉 LLM “我想要什么格式的输出”，比单纯用文字描述更直观。

四、Output Parser：LLM 输出从文本到结构化数据

LLM 的输出本质上是文本，但程序需要结构化数据。Output Parser 就是负责这个转换的组件。

4.1 为什么需要 Output Parser？

LLM 原始输出：
  "这部电影太棒了！视觉效果震撼，配乐动人。
   评分：9/10。优点：特效、配乐、剧情。
   缺点：前半段节奏慢。"

程序需要：
  {
    "rating": 9,
    "pros": ["特效", "配乐", "剧情"],
    "cons": ["前半段节奏慢"]
  }

4.2 三种常用 Parser

Parser 作用 适用场景
StrOutputParser 提取纯文本 简单问答
JsonOutputParser 解析 JSON 字符串 需要结构化数据
PydanticOutputParser 解析为 Pydantic 对象 需要类型校验（推荐）

4.3 PydanticOutputParser：最推荐的方式

from pydantic import BaseModel, Field
from typing import List
from langchain_core.output_parsers import PydanticOutputParser

# 1. 定义输出结构
class MovieReview(BaseModel):
    title: str = Field(description="电影名称")
    rating: int = Field(description="评分 1-10", ge=1, le=10)
    pros: List[str] = Field(description="优点列表")
    cons: List[str] = Field(description="缺点列表")

# 2. 创建 Parser
parser = PydanticOutputParser(pydantic_object=MovieReview)

# 3. 获取格式说明（自动生成的 JSON Schema）
format_instructions = parser.get_format_instructions()

# 4. 把格式说明注入 Prompt
prompt = ChatPromptTemplate.from_template("""
分析以下电影评论：

{review}

{format_instructions}
""")
prompt = prompt.partial(format_instructions=format_instructions)

# 5. 构建 Chain
chain = prompt | llm | parser

# 6. 调用
result = chain.invoke({"review": "《星际穿越》..."})
# result 是 MovieReview 对象，可以直接点属性
print(result.rating)  # 9
print(result.pros)    # ["视觉效果", "配乐"]

Pydantic Parser 的优势：

  • 自动将 LLM 输出转换为强类型对象
  • 自带 JSON Schema 生成，Prompt 中无需手写格式要求
  • 类型校验失败时自动重试（可选配置）

五、Runnable 接口：为什么所有组件都能用 | 连接

你也许会好奇：为什么 PromptTemplate、LLM、OutputParser 这些完全不同的东西，都能用 | 连接？

答案是：它们都实现了 Runnable 接口。

5.1 Runnable 的核心方法

from langchain_core.runnables import Runnable

# 所有 Runnable 组件都支持：
runnable.invoke(input)     # 同步调用
runnable.stream(input)     # 流式输出
runnable.batch(inputs)     # 批量处理
runnable.ainvoke(input)    # 异步调用

5.2 自定义 Runnable

如果内置组件不够用，你可以自定义 Runnable：

from langchain_core.runnables import RunnableLambda

# 用 lambda 创建自定义组件
def add_context(input_dict):
    return {
        "context": f"这是关于{input_dict['topic']}的背景信息",
        "question": input_dict["question"]
    }

custom_step = RunnableLambda(add_context)

# 自定义组件可以无缝接入 Chain
chain = custom_step | prompt | llm | parser

5.3 RunnablePassthrough：透传数据

有时候你需要把输入同时传给多个下游组件：

from langchain_core.runnables import RunnablePassthrough

# 同时执行两个 Chain，保留原始输入
chain = (
    RunnablePassthrough.assign(
        summary=lambda x: summary_chain.invoke({"text": x["text"]}),
        keywords=lambda x: keyword_chain.invoke({"text": x["text"]})
    )
    | final_prompt
    | llm
)

# 输入 {"text": "长文本"}
# 下游可以同时拿到 text、summary、keywords

六、顺序与并行：复杂工作流怎么搭

当业务逻辑变复杂时，简单的 A | B | C 不够用了。LangChain 提供了顺序和并行两种编排方式。

6.1 顺序 Chain：前一个的输出是后一个的输入

# Chain 1：生成标题
title_chain = title_prompt | llm | parser

# Chain 2：基于标题生成大纲
outline_chain = outline_prompt | llm | parser

# 手动串联（Sequential Chain）
def generate_article(inputs):
    title = title_chain.invoke(inputs)
    outline = outline_chain.invoke({"title": title})
    return {"title": title, "outline": outline}

6.2 并行 Chain：同时执行多个任务

from langchain_core.runnables import RunnableParallel

# 定义两个并行的 Chain
positive_chain = positive_prompt | llm | parser  # 分析优点
negative_chain = negative_prompt | llm | parser  # 分析缺点

# 用 RunnableParallel 同时执行
parallel_chain = RunnableParallel(
    advantages=positive_chain,
    disadvantages=negative_chain
)

# 一次调用，两个 Chain 同时执行
result = parallel_chain.invoke({"topic": "远程工作"})
# result["advantages"] 和 result["disadvantages"] 同时拿到

输入 topic=远程工作

并行执行

优点分析 Chain

缺点分析 Chain

结果合并

返回字典 advantages+disadvantages

七、AgentExecutor：Chain 和 Agent 的交汇点

前面六篇我们一直在讲 Agent（ReAct、记忆、工具调用、MCP），现在回到 LangChain，看看它怎么把 Agent 能力封装成 Chain。

7.1 AgentExecutor 的工作流程

是

否

用户输入

AgentExecutor

LLM 决定行动

需要工具？

执行工具

结果返回 LLM

生成最终回答

AgentExecutor 本质上是一个循环 Chain：

  1. 把用户输入传给 LLM
  2. LLM 决定是调用工具还是直接回答
  3. 如果需要工具，执行工具并把结果传回 LLM
  4. 重复步骤 2-3，直到 LLM 决定直接回答

7.2 AgentExecutor vs 普通 Chain

维度 普通 Chain AgentExecutor
执行流程 固定流水线 动态循环（可能多次调用工具）
LLM 角色 生成内容 决策（调用工具还是回答）
工具使用 无或固定 动态选择
适用场景 确定性任务 需要推理和工具调用的复杂任务

7.3 代码示例

from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import Tool

# 定义工具
tools = [
    Tool(name="search", func=search_api, description="搜索网络信息"),
    Tool(name="calc", func=calculator, description="执行数学计算")
]

# 创建 ReAct Agent
agent = create_react_agent(llm, tools, prompt)

# 用 AgentExecutor 包装
agent_executor = AgentExecutor(agent=agent, tools=tools)

# 调用（内部自动处理循环）
result = agent_executor.invoke({"input": "北京今天的天气怎样？"})

💡 关键点：AgentExecutor 把 Agent 的复杂决策流程封装成了标准的 Runnable 接口。你可以把它当作一个黑盒 Chain 来用，也可以深入定制每一步的行为。

八、总结

本文从 Chain 的本质出发，梳理了：

  1. Chain 的本质：不是"链"，是"声明式数据流水线"
  2. LCEL 语法：用 | 管道符连接组件，数据从左到右自动流转
  3. PromptTemplate：不只是字符串拼接，支持多角色消息和 Few-Shot 示例
  4. Output Parser：把 LLM 文本输出转为结构化数据，Pydantic Parser 最推荐
  5. Runnable 接口：所有组件统一的标准接口，支持 invoke/stream/batch/ainvoke
  6. 顺序与并行：RunnableParallel 实现多任务并发执行
  7. AgentExecutor：把 Agent 的动态决策流程封装为标准 Chain 接口

参考资源：

  • LangChain Documentation: LCEL Overview
  • LangChain Core: Runnable Interface
  • LangChain Cookbook: Chains and Agents
# AI
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

WalkingLab 组织入驻 AtomGit，释放高校智能体开源科研成果

avatar AtomGit开源社区
cover

海外动态代理行业彻底换代，AI 数据时代新标准已到来

avatar AtomGit开源社区
cover

全球首个！京东全栈开源JoyAI-VL-Interaction，让大模型从“一问一答”走向“边看边说”

avatar AtomGit开源社区