介绍
RunnableLambda 是 LangChain 中用于 将普通 Python 函数包装成「可运行组件（Runnable）」 的工具，让普通函数能无缝融入 LangChain 的链式调用（| 语法）中。
简单来说：
- 普通函数无法直接用
|和 Prompt、LLM、解析器等组件串联； - 用
RunnableLambda包装后，函数就变成了 Runnable 类型，能和其他组件一起组成链式流程。
用法
方式1
传入普通函数
RunnableLambda(fun)
参数解释
- fun：普通函数
方式2
传入lambda函数
RunnableLambda(lambda 函数)
例子1-传入普通函数
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.runnables import RunnableLambda
from dotenv import load_dotenv
# 加载 .env 文件，并放入到环境变量中
load_dotenv()
# 定义普通 Python 函数，提取LLM返回的文本内容
def process_resp(resp) -> str:
"""
调用 .content，直接返回字符串
"""
return resp.content
# 用 RunnableLambda 包装函数
rl = RunnableLambda(process_resp)
# 定义模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个温柔的小学数学老师，回答问题时用“亲”、“小朋友”，并且简短回答问题，回答问题不要用markdown格式"),
("human", "计算过程：{msg}")
])
model = ChatTongyi(model="qwen3-max")
# 串联时，最后使用 rl
chain = prompt | model | rl
while True:
"""
输入：3x+5=20，x 等于多少？
"""
question = input("请输入：")
# 调用stream向模型提问，返回 generator类型
res = chain.stream({
"msg": question
})
# 流式输出
for chunk in res:
# ※※※ 不需要调用 .content ※※※
print(chunk, end="", flush=True)
print("\n----回答结束-------\n")
例子2-传入lambda函数
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.runnables import RunnableLambda
from dotenv import load_dotenv
# 加载 .env 文件，并放入到环境变量中
load_dotenv()
# 创建 RunnableLambda 对象，传入 lambda 函数
rl = RunnableLambda(lambda resp: resp.content)
# 定义模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个温柔的小学数学老师，回答问题时用“亲”、“小朋友”，并且简短回答问题，回答问题不要用markdown格式"),
("human", "计算过程：{msg}")
])
model = ChatTongyi(model="qwen3-max")
# 串联时，最后使用 rl
chain = prompt | model | rl
while True:
"""
输入：3x+5=20，x 等于多少？
"""
question = input("请输入：")
# 调用stream向模型提问，返回 generator类型
res = chain.stream({
"msg": question
})
# 流式输出
for chunk in res:
# ※※※ 不需要调用 .content ※※※
print(chunk, end="", flush=True)
print("\n----回答结束-------\n")
案例3-转成JSON
通过定义函数，将返回结果转成 JSON：
from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.runnables import RunnableLambda
from dotenv import load_dotenv
import json
# 定义普通 Python 函数，提取LLM返回的文本内容
def json_parse(resp) -> dict:
"""安全的JSON解析函数：捕获解析错误"""
text = resp.content
try:
return json.loads(text)
except json.JSONDecodeError:
print(f"❌ JSON解析失败，原始文本：{text}")
return {"code": 500} # 返回默认值，避免链中断
# 用 RunnableLambda 包装函数
rl = RunnableLambda(json_parse)
# 第一步：定义示例模板（告诉 AI 示例的格式）
example_template = """
名称：{name}
屏幕：{screen}
CPU：{cpu}
5G：{5G}
"""
example_prompt = PromptTemplate.from_template(example_template)
# 第二步：准备示例（列表+字典格式，每个字典是一个示例）
examples = [
{"name": "iPhone 17 Pro Max", "screen": "6.9英寸", "cpu": "A19 Pro（3纳米制程）", "5G": "支持"}
]
# 第三步：1.定义开头的引导语
prefix = "请按照以下格式返回指定手机的配置（参考示例格式）："
# 第三步：2.定义问AI的问题
# ※※※必须明确指出输出格式※※※
# suffix = "现在请生成 {name} 的手机配置，必须严格按照指定格式输出JSON："
suffix = "现在请生成 {name} 的手机配置，必须严格按照指定格式输出JSON，不要markdown："
# 第四步：组装 FewShotPromptTemplate
few_shot_prompt = FewShotPromptTemplate(
prefix=prefix, # 开头的引导语
example_prompt=example_prompt, # 示例的格式模板
examples=examples, # 要展示给 AI 的示例
suffix=suffix, # 问AI的问题（包含动态变量）
input_variables=["name"], # 最终要传入的动态变量（这里是城市名）
)
# 加载 .env 文件，并放入到环境变量中
load_dotenv()
# 创建模型对象
model = ChatTongyi(model="qwen-max")
# 串联模板、模型、json输出解析器（| 运算符）
chain = few_shot_prompt | model | rl
# 因为要将结果转成json，所以要获取到全部返回结果才能转换，所以不能用流式输出，否则格式错乱
resp = chain.invoke({"name": "iphone11"})
print("---------大模型回答--------------------")
print("res类型", type(resp))
print(resp)
代码解释1
def json_parse(resp) -> dict:
"""安全的JSON解析函数：捕获解析错误"""
text = resp.content
try:
return json.loads(text)
except json.JSONDecodeError:
print(f"❌ JSON解析失败，原始文本：{text}")
return {"code": 500} # 返回默认值，避免链中断
代码解释2
执行下面代码，返回的是 markdown 格式，转成JSON会报错：
suffix = "现在请生成 {name} 的手机配置，必须严格按照指定格式输出JSON："
报错如下：
❌ JSON解析失败，原始文本：``json
{
"名称": "iPhone 11",
"屏幕": "6.1英寸",
"CPU": "A13 Bionic",
"5G": "不支持"
}
``
---------大模型回答--------------------
res类型 <class 'dict'>
{'code': 500}
Process finished with exit code 0
代码解释3
执行下面代码，返回的是标准 JSON 格式，正常转成JSON：
suffix = "现在请生成 {name} 的手机配置，必须严格按照指定格式输出JSON，不要markdown："
执行结果：
---------大模型回答--------------------
res类型 <class 'dict'>
{'名称': 'iPhone 11', '屏幕': '6.1英寸', 'CPU': 'A13 Bionic（7纳米制程）', '5G': '不支持'}
关键特性与使用场景
1. 支持不同类型的输入输出
RunnableLambda 不限制函数的输入输出类型，可处理：
- 字符串、数字、字典等基础类型；
- LangChain 内置类型（如
ChatMessage、BaseMessage）； - 自定义类实例。
2. 处理链式调用中的「中间加工」
这是最常用的场景：在链式流程中插入自定义逻辑，比如：
- 清洗 LLM 返回的文本（去除多余字符、修正格式）；
- 转换数据类型（如将字符串转为字典/列表）；
- 过滤/提取数据（如从结构化结果中取指定字段）；
- 记录日志、统计耗时等辅助操作。
3. 支持异步函数
如果你的函数是异步的（async def 定义），RunnableLambda 会自动适配，无需额外处理：
import asyncio
# 定义异步函数
async def async_process(text: str) -> str:
await asyncio.sleep(0.1) # 模拟异步操作（如网络请求）
return text.lower()
# 包装异步函数
async_runnable = RunnableLambda(async_process)
# 异步调用
async def main():
result = await async_runnable.ainvoke("HELLO ASYNC")
print("✅ 异步调用结果：", result) # 输出：hello async
asyncio.run(main())
4. 错误处理
可在包装的函数中添加异常捕获，保证链式调用的稳定性：
def safe_json_parse(text: str) -> dict:
"""安全的JSON解析函数：捕获解析错误"""
try:
return json.loads(text)
except json.JSONDecodeError:
print(f"❌ JSON解析失败，原始文本：{text}")
return {"keywords": []} # 返回默认值，避免链中断
# 包装后融入链，即使解析失败也不会崩溃
safe_parse_runnable = RunnableLambda(safe_json_parse)
进阶：带上下文的函数（RunnableLambda + Config）
如果需要在函数中获取链式调用的上下文（如 run_id、tags），可通过 config 参数获取：
def with_context(text: str, config: dict) -> str:
"""函数接收第二个参数：config（链式调用的上下文）"""
run_id = config.get("run_id", "未知ID")
return f"[运行ID：{run_id}] {text}"
# 包装时指定接收config
runnable_with_context = RunnableLambda(with_context, takes_config=True)
# 调用时传入config
result = runnable_with_context.invoke(
"测试上下文",
config={"run_id": "123456"}
)
print("✅ 带上下文的函数结果：", result)
# 输出：[运行ID：123456] 测试上下文
总结
RunnableLambda的核心作用是将普通Python函数包装为LangChain的Runnable组件，使其能融入链式调用；- 主要用于链式流程中的「自定义中间处理」（如数据清洗、格式转换、数据提取）；
- 支持同步/异步函数，兼容任意输入输出类型，可添加错误处理保证链的稳定性。
简单来说，RunnableLambda 是「普通Python函数」和「LangChain链式调用」之间的桥梁，让你能灵活插入自定义逻辑，而不用受限于LangChain内置组件。