LCEL和Runnable在Langchain 1.2.0中如何？
摘要：LangChain LCEL学习笔记 本来第三章应该学一下Tools和Agent的，后来发现这两个东西是核心，越学东西越多，于是就让G老师帮忙重新列了一下学习路线，按照G老师的路线，这一段应该学习LCEL相关的信息。回归正题，本文将带您深入
LangChain LCEL学习笔记
本来第三章应该学一下Tools和Agent的，后来发现这两个东西是核心，越学东西越多，于是就让G老师帮忙重新列了一下学习路线，按照G老师的路线，这一段应该学习LCEL相关的信息。回归正题，本文将带您深入探索 LCEL 与 Runnable 接口的奥秘。
1. 引言：从“链”说起
1.1 为什么要了解 LCEL？
在 LangChain 的世界里，有一个词贯穿始终——Chain（链）。
想象一下：用户输入一段文字，你需要先做情感分析，再根据情感结果选择不同的回复策略，最后将结果格式化输出。如果不用 LCEL，你可能需要写一堆嵌套函数、回调逻辑，代码像意大利面条一样纠缠不清。
但有了 LCEL，一切变得清晰优雅：
chain = (
sentiment_prompt # 第一步：情感分析
| llm # 第二步：LLM 处理
| output_parser # 第三步：格式化输出
)
这就是 LCEL 的魅力所在——用管道操作符声明式地组合数据流，让复杂的 AI 工作流变得像搭积木一样简单。
1.2 LCEL 的核心优势
LCEL（LangChain Expression Language）不仅仅是一个语法糖，它为 LangChain 带来了质的飞跃：
优势
说明
声明式
用管道操作符清晰表达数据流，阅读代码如读业务流程
组合性
轻松组合提示词、模型、输出解析器构建复杂链
流式支持
原生支持流式输出，用户体验更佳
并行执行
支持批量和并行处理，性能与效率兼得
错误处理
内置错误处理机制，链更健壮
追踪支持
与 LangSmith 无缝集成，调试不再抓狂
2. LCEL 基础：一切的开始
2.1 核心概念：Runnable 接口
LCEL 的核心是 Runnable 接口。你可以把它想象成 LangChain 世界的“统一身份证”——无论是提示词模板、LLM 模型，还是输出解析器，都实现了这个接口。
这意味着它们都有共同的方法：
# 基本调用
chain.invoke(input) # 同步调用
chain.batch(inputs) # 批量调用
chain.stream(input) # 流式调用
# 异步调用
await chain.ainvoke(input) # 异步调用
await chain.abatch(inputs) # 异步批量调用
async for chunk in chain.astream(input): # 异步流式调用
这就是 Runnable 的核心方法三剑客：invoke、batch、stream。无论你组合什么样的链，都用这几种方式调用。
2.2 第一个 LCEL 链
让我们用一个完整的例子来感受 LCEL 的魅力：
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
# 初始化模型
base_url = "https://api.minimax.chat/v1"
api_key = "your-api-key"
model = ChatOpenAI(
model="MiniMax-M2.7",
base_url=base_url,
api_key=api_key,
)
# 构建链
prompt = ChatPromptTemplate.from_template("用{language}翻译：{text}")
output_parser = StrOutputParser()
chain = prompt | model | output_parser
# 调用链
result = chain.invoke({
"language": "中文",
"text": "Hello, World!"
})
print(result) # 你好，世界！
这就是一个完整的 LCEL 链。数据像水流一样经过三个处理单元：prompt → model → output_parser。
数据流向是这样的：
输入 {"language": "中文", "text": "Hello, World!"}
↓
prompt.invoke() → ChatMessage
↓
model.invoke() → AIMessage
↓
output_parser.invoke() → String
↓
输出 "你好，世界！"
3. Runnable 家族详解
这是本文的重点之一。LangChain 提供了丰富的 Runnable 组件，它们像乐高积木一样可以任意组合。本章节深入了解每一个组件。
3.1 基础构建块
3.1.1 RunnableLambda：将函数变为可组合的组件
RunnableLambda 是最基础的工具，它的作用只有一个——把普通 Python 函数转换成 Runnable 对象。
from langchain_core.runnables import RunnableLambda
# 定义一个普通函数
def add_ten(x: int) -> int:
return x + 10
# 转换为 Runnable
r = RunnableLambda(add_ten)
# 调用
result = r.invoke(5) # 15
你也可以使用匿名函数：
r = RunnableLambda(lambda x: x + 10)
result = r.invoke(5) # 15
为什么要转换为 Runnable？
因为转换后，这个函数就可以和其他 Runnable 用管道操作符连接了
def multiply_ten(x: int) -> int:
return x * 10
# 链接两个 Runnable
chain = RunnableLambda(add_ten) | RunnableLambda(multiply_ten)
result = chain.invoke(1) # (1 + 10) * 10 = 110
3.1.2 管道操作符 |：像搭积木一样组装链
管道操作符是 LCEL 的灵魂。它的作用很直观：把左侧的输出传给右侧作为输入。
# 等价于：result = step2(step1(input))
chain = step1 | step2
# 多步管道
chain = step1 | step2 | step3 | step4
除了 | 操作符，还有 .pipe() 方法可以实现同样的效果：
chain = r1.pipe(r2).pipe(r3)
两者完全等价，看个人喜好选择。
3.1.3 三种调用方式：invoke / batch / stream
方法
说明
适用场景
invoke
单个输入 → 单个输出
实时交互、简单场景
batch
列表输入 → 列表输出
批量处理、效率优先
stream
单个输入 → 迭代器输出
流式输出、用户体验
# invoke: 单个输入 -> 单个输出
result = chain.invoke(5)
# batch: 批量输入 -> 列表输出（并行执行）
results = chain.batch([1, 2, 3, 4, 5])
# stream: 流式输出（迭代器）
for item in chain.stream(5):
print(item) # 逐步输出
batch 会自动并行执行，非常适合需要处理大量数据的场景。如果担心 API 限流，可以用config={"max_concurrency": 3} 参数控制并发数，后面会讲一下config操作
3.2 并行与条件
3.2.1 RunnableParallel：多任务并行处理
有时候我们需要同时执行多个任务，比如同时获取翻译、总结、情感分析结果。RunnableParallel 就是为此而生。
from langchain_core.runnables import RunnableParallel, RunnableLambda
def add_ten(x: int) -> int:
return x + 10
def multiply_ten(x: int) -> int:
return x * 10
def square(x: int) -> int:
return x * x
# 字典形式定义并行任务
parallel = RunnableParallel({
"add": RunnableLambda(add_ten),
"mul": RunnableLambda(multiply_ten),
"square": RunnableLambda(square),
})
# 调用 - 所有任务并行执行
result = parallel.invoke(5)
# {'add': 15, 'mul': 50, 'square': 25}
也可以用关键字参数的形式：
parallel2 = RunnableParallel(
add=RunnableLambda(add_ten),
mul=RunnableLambda(multiply_ten),
)
实际应用场景：在 Agent 开发中，你可以用 RunnableParallel 同时调用多个工具（比如同时查询天气、股票、新闻），然后汇总结果。
3.2.2 RunnableBranch：条件分支（if-elif-else）
RunnableBranch 允许根据条件选择不同的处理路径。它的工作方式类似 if-elif-else 逻辑。
from langchain_core.runnables import RunnableBranch, RunnableLambda
def to_positive(x: int) -> str:
return f"{x} 是正数"
def to_negative(x: int) -> str:
return f"{x} 是负数"
def to_zero(x: int) -> str:
return f"{x} 是零"
# 条件分支 - 最后一个参数作为默认分支
branch = RunnableBranch(
(lambda x: x > 0, RunnableLambda(to_positive)),
(lambda x: x < 0, RunnableLambda(to_negative)),
RunnableLambda(to_zero), # 默认分支
)
# 调用
print(branch.invoke(5)) # '5 是正数'
print(branch.invoke(-3)) # '-3 是负数'
print(branch.invoke(0)) # '0 是零'
在 Agent 开发中的应用：这是 Agent 决策的核心组件！根据用户输入或中间结果，Branch 可以决定：
使用哪个工具
调用哪个模型
走哪个处理流程
RunnableBranch 的多种路由场景
下面展示 6+ 种实用的路由场景：
场景一：分数等级路由
score_branch = RunnableBranch(
(lambda x: x >= 90, RunnableLambda(lambda x: f"等级A: {x}分")),
(lambda x: x >= 80, RunnableLambda(lambda x: f"等级B: {x}分")),
(lambda x: x >= 70, RunnableLambda(lambda x: f"等级C: {x}分")),
(lambda x: x >= 60, RunnableLambda(lambda x: f"等级D: {x}分")),
RunnableLambda(lambda x: f"等级F: {x}分"), # 默认分支
)
scores = [95, 82, 75, 63, 45]
for s in scores:
print(f" {s}分 -> {score_branch.invoke(s)}")
# 输出：
# 95分 -> 等级A: 95分
# 82分 -> 等级B: 82分
# 75分 -> 等级C: 75分
# 63分 -> 等级D: 63分
# 45分 -> 等级F: 45分
场景二：类型路由
user_chain = RunnableLambda(lambda x: f"用户处理: {x['name']}")
product_chain = RunnableLambda(lambda x: f"商品处理: {x['name']}")
order_chain = RunnableLambda(lambda x: f"订单处理: {x['name']}")
type_router = RunnableBranch(
(lambda x: x.get("type") == "user", user_chain),
(lambda x: x.get("type") == "product", product_chain),
(lambda x: x.get("type") == "order", order_chain),
RunnableLambda(lambda x: f"默认处理: {x}"),
)
print(type_router.invoke({"type": "user", "name": "Alex"}))
# 输出：用户处理: Alex
场景三：数值范围路由
high_chain = RunnableLambda(lambda x: f"高值处理: {x * 10}")
medium_chain = RunnableLambda(lambda x: f"中值处理: {x * 5}")
low_chain = RunnableLambda(lambda x: f"低值处理: {x * 2}")
value_router = RunnableBranch(
(lambda x: x > 100, high_chain),
(lambda x: x > 50, medium_chain),
low_chain,
)
print(value_router.invoke(60)) # 中值处理: 300
print(value_router.invoke(150)) # 高值处理: 1500
场景四：字符串前缀路由
string_processor = RunnableBranch(
(lambda x: x.startswith("upper:"),
RunnableLambda(lambda x: x[6:].upper())),
(lambda x: x.startswith("lower:"),
RunnableLambda(lambda x: x[6:].lower())),
(lambda x: x.startswith("cap:"),
RunnableLambda(lambda x: x[4:].capitalize())),
RunnableLambda(lambda x: f"[默认] {x}"),
)
print(string_processor.invoke("upper:hello")) # HELLO
print(string_processor.invoke("lower:HELLO")) # hello
print(string_processor.invoke("cap:hello")) # Hello
3.2.3 RouterRunnable：简单 key→handler 映射
有时候只需要简单的 key 映射，不需要复杂的条件判断。RouterRunnable 就是为这种情况设计的。
from langchain_core.runnables.router import RouterRunnable
# 创建路由表：key -> Runnable
router = RouterRunnable(runnables={
"add": RunnableLambda(lambda x: x + 10),
"multiply": RunnableLambda(lambda x: x * 10),
"square": RunnableLambda(lambda x: x ** 2),
})
# 输入必须是字典，包含 "key" 和 "input"
result = router.invoke({"key": "add", "input": 5})
print(result) # 15
RouterRunnable vs RunnableBranch 对比
特性
RouterRunnable
RunnableBranch
路由方式
根据 key 字段选择
根据条件函数返回 True/False
输入格式
必须是 {"key": "...", "input": ...}
任意类型
灵活性
低，key 必须是字符串
高，可自定义任意条件
复杂度
简单，适合 handler 映射
复杂，适合 if-elif-else
选择建议：
需要简单的 key → handler 映射？用 RouterRunnable
需要复杂的条件判断？用 RunnableBranch
3.3 数据流转神器
3.3.1 RunnablePassthrough：透传与字段选择
RunnablePassthrough 有两个主要用途：透传输入和字段操作。
基本透传
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
def multiply_ten(x: int) -> int:
return x * 10
# 透传当前值，同时进行其他处理
chain = RunnableLambda(multiply_ten) | {
"original": RunnablePassthrough(), # 透传原始输入
"processed": RunnableLambda(lambda x: x * 10),
}
result = chain.invoke(5)
# {'original': 5, 'processed': 50}
使用 .pick() 选择字段
# 从字典中选择单个字段
pick_name = RunnablePassthrough().pick("name")
result = pick_name.invoke({"name": "Alex", "age": 30})
# 'Alex'
# 从字典中选择多个字段
pick_multi = RunnablePassthrough().pick(["name", "age"])
result = pick_multi.invoke({"name": "Alex", "age": 30})
# {'name': 'Alex', 'age': 30}
使用 .assign() 添加字段
# assign - 添加字段
chain = (
RunnableLambda(lambda x: {"input": x})
| RunnablePassthrough.assign(added=lambda x: x["input"] + 10)
)
result = chain.invoke(5)
# {'input': 5, 'added': 15}
# assign 多个字段
chain_multi_assign = (
RunnableLambda(lambda x: {"original": x})
| RunnablePassthrough.assign(
doubled=lambda x: x["original"] * 2,
squared=lambda x: x["original"] ** 2,
)
)
result = chain_multi_assign.invoke(5)
# {'original': 5, 'doubled': 10, 'squared': 25}
3.3.2 RunnableSequence：显式序列创建
RunnableSequence 用于显式创建有序的 Runnable 序列，等同于使用 | 管道操作符。
from langchain_core.runnables import RunnableSequence, RunnableLambda
def add_ten(x: int) -> int:
return x + 10
def multiply_ten(x: int) -> int:
return x * 10
def square(x: int) -> int:
return x * x
def to_string(x: int) -> str:
return f"结果是: {x}"
# 显式创建序列
sequence = RunnableSequence(
first=RunnableLambda(add_ten),
middle=[RunnableLambda(multiply_ten), RunnableLambda(square)],
last=RunnableLambda(to_string),
)
result = sequence.invoke(5)
# 5 -> add_ten -> 15 -> multiply_ten -> 150 -> square -> 22500 -> to_string -> "结果是: 22500"
3.4 批量处理
3.4.1 RunnableEach：序列元素逐个处理
RunnableEach 对输入序列中的每个元素执行同一个 Runnable。
from langchain_core.runnables import RunnableLambda, RunnableEach
def add_ten(x: int) -> int:
return x + 10
def multiply_ten(x: int) -> int:
return x * 10
# 基本用法 - 对每个元素执行 add_ten
each_add = RunnableEach(bound=RunnableLambda(add_ten))
result = each_add.invoke([1, 2, 3])
# [11, 12, 13]
# 链式组合
chain_each = (
RunnableEach(bound=RunnableLambda(add_ten))
| RunnableEach(bound=RunnableLambda(multiply_ten))
)
result = chain_each.invoke([1, 2, 3])
# [1, 2, 3] -> add_ten -> [11, 12, 13] -> multiply_ten -> [110, 120, 130]
3.4.2 batch_as_completed：完成即返回
batch_as_completed 会在任务完成时立即返回，而不是等待所有任务完成。
from langchain_core.runnables import RunnableLambda
def slow_op(x):
import time
time.sleep(x * 0.1)
return x * 10
chain = RunnableLambda(slow_op)
# 普通 batch - 按顺序返回
results = chain.batch([1, 2, 3, 4])
# batch_as_completed - 哪个完成返回哪个
for result in chain.batch_as_completed([1, 2, 3, 4]):
print(result) # 按完成顺序打印结果
4. 进阶功能：让链更健壮
这一章是让 LCEL 链真正用于生产环境的关键。想象一下：你的链在凌晨三点因为网络波动失败了，你希望它自动重试吗？你希望优雅地降级到备用模型吗？这一章告诉你怎么做。
4.1 容错机制
4.1.1 with_fallbacks：备用方案
with_fallbacks 为 Runnable 添加备用方案。当主 Runnable 抛出异常时，自动切换到备用方案。
from langchain_core.runnables import RunnableLambda
def always_fail(x):
raise RuntimeError(f"处理 {x} 时出错")
def fallback_handler(x):
return f"备用方案处理: {x}"
def call_gpt4(x):
return f"GPT-4: {x}"
def call_gpt35(x):
return f"GPT-3.5: {x}"
def call_local(x):
return f"Local: {x}"
# 基本用法 - 单个备用
chain = RunnableLambda(always_fail).with_fallbacks(
fallbacks=[RunnableLambda(fallback_handler)]
)
result = chain.invoke(5) # '备用方案处理: 5'
# 模型降级示例 - 多级备用
model_chain = (
RunnableLambda(call_gpt4)
.with_fallbacks(fallbacks=[RunnableLambda(call_gpt35)])
.with_fallbacks(fallbacks=[RunnableLambda(call_local)])
)
result = model_chain.invoke("Hello")
# 依次尝试 GPT-4 -> GPT-3.5 -> Local
4.1.2 with_retry：重试机制
with_retry 为 Runnable 添加重试机制，适用于处理网络请求临时故障等场景。
from langchain_core.runnables import RunnableLambda
def might_fail_once(x: int) -> int:
"""模拟可能失败一次的函数"""
if not hasattr(might_fail_once, 'called'):
might_fail_once.called = True
print("失败一次")
raise RuntimeError("第一次调用失败")
return x + 10
def selective_fail(x):
if x < 0:
raise ConnectionError("连接失败")
return x * 2
# 基本用法
retry_once = RunnableLambda(might_fail_once).with_retry()
result = retry_once.invoke(5) # 10
# 指定重试次数和异常类型
retry_connection = RunnableLambda(selective_fail).with_retry(
retry_if_exception_type=(ConnectionError,),
stop_after_attempt=3,
)
# 指数退避 + 随机抖动
retry_with_jitter = RunnableLambda(might_fail_once).with_retry(
wait_exponential_jitter=True,
stop_after_attempt=5,
)
# 链式组合中的重试
chain_with_retry = (
RunnableLambda(add_ten)
| RunnableLambda(might_fail_once).with_retry()
| RunnableLambda(multiply_ten)
)
参数详解：
retry_if_exception_type：指定需要重试的异常类型
stop_after_attempt：最大重试次数
wait_exponential_jitter：指数退避+随机抖动，避免惊群效应
4.2 配置与监控
4.2.1 RunnableConfig：执行配置
RunnableConfig 用于控制 Runnable 的执行行为，可以传递给 invoke、batch、stream 等方法。
from langchain_core.runnables import RunnableLambda, RunnableConfig
def add_ten(x: int) -> int:
return x + 10
chain = RunnableLambda(add_ten)
# tags - 标签管理
config_tags = {"tags": ["user", "premium", "urgent"]}
result = chain.invoke(5, config=config_tags)
# max_concurrency - 最大并发数
config_concurrency = {"max_concurrency": 2}
results = chain.batch([1, 2, 3, 4], config=config_concurrency)
# with_config 方法 - 在链中设置配置
chain_with_config = (
RunnableLambda(add_ten).with_config(run_name="add_ten_step")
| RunnableLambda(lambda x: x * 10).with_config(run_name="multiply_step")
)
RunnableConfig 常用参数：
参数
说明
典型用法
tags
标签管理
区分生产/开发/测试环境
metadata
元数据
追踪请求来源、用户ID
max_concurrency
最大并发数
控制 API 调用的并发度
recursion_limit
递归深度限制
防止无限循环
callbacks
回调函数
集成监控、日志
run_name
运行名称
方便调试时识别
4.2.2 with_listeners：生命周期监听器
with_listeners 允许在 Runnable 执行的不同阶段（开始，结束，失败）插入自定义逻辑。
from langchain_core.runnables import RunnableLambda
def on_start_callback(run):
print(f"[on_start] 开始执行 run_id={run.id}")
def on_end_callback(run):
print(f"[on_end] 执行完成 run_id={run.id}")
def on_error_callback(run):
print(f"[on_error] 出错: run_id={run.id}")
chain = RunnableLambda(add_ten).with_listeners(
on_start=on_start_callback,
on_end=on_end_callback,
on_error=on_error_callback,
)
result = chain.invoke(5)
# 输出：
# [on_start] 开始执行 run_id=...
# [on_end] 执行完成 run_id=...
5. Runnable 在 Agent 开发中的应用
这一章我们来聊聊 Runnable 在 LangChain Agent 中的应用。Agent 是 LangChain 最强大的特性之一，而它背后的核心正是 Runnable 接口。
5.1 Agent 与 Runnable 的关系
想象一下 Agent 的工作流程：感知用户输入 → 决策下一步做什么 → 执行动作 → 获取结果 → 继续决策...
这个流程中的每一步，都可以用 Runnable 来表示：
# Agent 的简化工作流
agent_chain = (
prompt # 1. 感知：解析用户输入
| llm.with_tools(tools) # 2. 决策：选择要执行的工具
| output_parser # 3. 解析：解析工具调用
| execute_tools # 4. 执行：运行选择的工具
| ... # 循环往复
)
本质上，Agent 就是一条复杂的 Runnable 链。它之所以能灵活地处理各种任务，正是因为它建立在 Runnable 的组合能力之上。
5.2 动态路由在 Agent 中的应用
在 Agent 中，最常见的场景是根据条件决定下一步操作。
5.2.1 工具选择的条件判断
from langchain_core.runnables import RunnableBranch, RunnableLambda
# 根据用户意图选择不同工具
def classify_intent(x):
return x.get("intent", "unknown")
intent_router = RunnableBranch(
(lambda x: x.get("intent") == "weather",
RunnableLambda(lambda x: "调用天气API")),
(lambda x: x.get("intent") == "search",
RunnableLambda(lambda x: "调用搜索API")),
(lambda x: x.get("intent") == "calculator",
RunnableLambda(lambda x: "调用计算器")),
RunnableLambda(lambda x: "调用通用对话模型"),
)
5.2.2 多模型切换
# 根据任务类型选择不同模型
def select_model(x):
task_type = x.get("task_type", "general")
if task_type == "code":
return "gpt-4"
elif task_type == "creative":
return "claude-3"
return "gpt-3.5-turbo"
model_router = RunnableBranch(
(lambda x: x.get("task_type") == "code",
llm.bind(model="gpt-4")),
(lambda x: x.get("task_type") == "creative",
llm.bind(model="claude-3")),
llm.bind(model="gpt-3.5-turbo"),
)
5.3 工具调用的容错处理
Agent 免不了要和外部工具打交道，而外部工具可能会失败。Runnable 的容错机制在这里大显身手。
5.3.1 工具失败时的备用方案
from langchain_core.runnables import RunnableLambda
def call_primary_api(x):
# 可能失败的 API 调用
raise ConnectionError("主API不可用")
def call_backup_api(x):
return f"备用API返回: {x}"
# 使用 with_fallbacks 处理工具失败
safe_tool_chain = (
RunnableLambda(call_primary_api)
.with_fallbacks(fallbacks=[RunnableLambda(call_backup_api)])
)
result = safe_tool_chain.invoke("test")
# 输出：备用API返回: test
5.3.2 网络/API 临时故障重试
def call_unreliable_api(x):
import random
if random.random() < 0.3:
raise ConnectionError("网络波动")
return f"成功: {x}"
# 使用 with_retry 处理临时故障
retryable_tool = (
RunnableLambda(call_unreliable_api)
.with_retry(
retry_if_exception_type=(ConnectionError,),
stop_after_attempt=3,
wait_exponential_jitter=True,
)
)
6. 实战案例
让我们通过几个实际案例，把学到的知识串联起来。
6.1 多语言翻译助手
这个案例展示如何使用 .batch() 批量处理多个翻译请求。
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
# 初始化模型
base_url = "https://api.minimax.chat/v1"
api_key = "your-api-key"
model = ChatOpenAI(
model="MiniMax-M2.7",
base_url=base_url,
api_key=api_key,
)
# 创建翻译链
translation_prompt = ChatPromptTemplate.from_template(
"将以下句子翻译成{language}：{text}"
)
output_parser = StrOutputParser()
translation_chain = translation_prompt | model | output_parser
# 批量翻译
batch_inputs = [
{"language": "中文", "text": "Hello, how are you?"},
{"language": "日语", "text": "Hello, how are you?"},
{"language": "法语", "text": "Hello, how are you?"},
{"language": "德语", "text": "Hello, how are you?"},
]
results = translation_chain.batch(batch_inputs)
# 返回: ["你好，你怎么样？", "こんにちは、お元気ですか？", "Bonjour", "Wie geht es dir?"]
6.2 多功能消息处理系统
这个案例展示如何使用 RunnableParallel 同时执行多个任务。
from langchain_core.runnables import RunnableParallel
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
# 初始化模型
model = ChatOpenAI(model="MiniMax-M2.7", base_url=base_url, api_key=api_key)
output_parser = StrOutputParser()
# 创建多任务链
multi_task_chain = RunnableParallel({
"translation": (
ChatPromptTemplate.from_template("翻译成中文：{text}")
| model
| StrOutputParser()
),
"summary": (
ChatPromptTemplate.from_template("用一句话总结：{text}")
| model
| StrOutputParser()
),
"sentiment": (
ChatPromptTemplate.from_template("分析情感（正面/负面/中性）：{text}")
| model
| StrOutputParser()
),
})
# 一次调用，同时获得翻译、总结和情感分析
result = multi_task_chain.invoke({
"text": "LangChain is an amazing framework for building LLM applications!"
})
# 返回:
# {
# "translation": "LangChain 是一个用于构建 LLM 应用的惊人框架！",
# "summary": "介绍 LangChain 框架的强大功能",
# "sentiment": "正面"
# }
6.3 带验证的 AI 对话链
这个案例展示如何使用 RunnableLambda 添加输入验证。
from langchain_core.runnables import RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
model = ChatOpenAI(model="MiniMax-M2.7", base_url=base_url, api_key=api_key)
def validate_input(input_dict):
if not input_dict.get("topic"):
raise ValueError("topic 不能为空")
if len(input_dict["topic"]) > 100:
raise ValueError("topic 长度不能超过100")
return input_dict
validation_chain = (
RunnableLambda(validate_input)
| ChatPromptTemplate.from_template("写一首关于{topic}的诗")
| model
| StrOutputParser()
)
# 正常调用
result = validation_chain.invoke({"topic": "春天"})
# 验证失败会抛出 ValueError
# validation_chain.invoke({"topic": ""}) # ValueError: topic 不能为空
7. 总结
7.1 Langchain中内置Runnable 类型
核心 Runnable 类
类/方法
功能
RunnableLambda
将普通函数转为 Runnable
RunnableParallel
并行执行多个 Runnable
RunnablePassthrough
透传输入或选择字段
RunnableBranch
条件分支与动态路由
RouterRunnable
基于 key 的简单路由
RunnableSequence
显式创建有序序列
RunnableEach
对序列中每个元素执行 Runnable
.with_fallbacks()
添加容错备用方案
.with_retry()
添加重试机制
.with_listeners()
添加生命周期监听器
.with_config()
设置执行配置
.pick()
从字典中选择字段
.assign()
向字典添加字段
提示模板与输出解析器
类型
说明
导入方式
ChatPromptTemplate
聊天提示模板
from langchain_core.prompts import ChatPromptTemplate
PromptTemplate
文本提示模板
from langchain_core.prompts import PromptTemplate
StrOutputParser
字符串输出解析器
from langchain_core.output_parsers import StrOutputParser
JsonOutputParser
JSON 输出解析器
from langchain_core.output_parsers import JsonOutputParser
XMLOutputParser
XML 输出解析器
from langchain_core.output_parsers import XMLOutputParser
LLM 模型
类型
说明
导入方式
ChatOpenAI
OpenAI 聊天模型
from langchain_openai import ChatOpenAI
OpenAI
OpenAI 文本模型
from langchain_openai import OpenAI
ChatAnthropic
Anthropic 聊天模型
from langchain_anthropic import ChatAnthropic
ChatVertexAI
Google Vertex AI 模型
from langchain_google_vertexai import ChatVertexAI
📝 个人笔记：本文档基于 LangChain 1.2.0 版本测试实验，碎碎念一下，今天和同事聊天，我感觉AI大模型时代其实应该了解Agent开发，同事说现在CC，opencode结合大模型这么厉害，Agent的开发也可以使用CC，opencode生成。使得我有点道心破碎不知所措，最近在工作中感觉JAVA的开发慢慢可以使用大模型代替导致有点看不到未来的路，后来想想既然学习了就学到底吧，有始有终，但行好事莫问前程。