LangChain教程：RunnabLeLambda组件让普通函数能融入 LangChain 的链式调用-马育民老师

# 介绍

`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函数

# 加载 .env 文件，并放入到环境变量中
load_dotenv()

# 创建 RunnableLambda 对象，传入 lambda 函数
rl = RunnableLambda(lambda resp: resp.content)

model = ChatTongyi(model="qwen3-max")

# 串联时，最后使用 rl
chain = prompt | model | rl

```

# 案例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` 会自动适配，无需额外处理：

```python
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. 错误处理
可在包装的函数中添加异常捕获，保证链式调用的稳定性：
```python
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` 参数获取：

```python
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] 测试上下文
```

# 总结

1. `RunnableLambda` 的核心作用是**将普通Python函数包装为LangChain的Runnable组件**，使其能融入链式调用；
2. 主要用于链式流程中的「自定义中间处理」（如数据清洗、格式转换、数据提取）；
3. 支持同步/异步函数，兼容任意输入输出类型，可添加错误处理保证链的稳定性。

简单来说，`RunnableLambda` 是「普通Python函数」和「LangChain链式调用」之间的桥梁，让你能灵活插入自定义逻辑，而不用受限于LangChain内置组件。

原文出处：http://malaoshi.top/show_1GW2qz81mZAA.html