RunnableLambda

RunnableLambda 是 LangChain 中自定义逻辑的核心组件，它允许你将任意 Python 函数（匿名函数/普通函数）封装成 LangChain 的 Runnable 接口，从而无缝融入 LangChain 的链（Chain）体系中。简单来说，它是“普通 Python 函数”和“LangChain 组件”之间的“适配器”。

一、核心概念（新手友好版）

可以把 RunnableLambda 理解成一个“函数包装器”：

• 你写的普通 Python 函数（比如数据处理、计算、格式转换），本身不支持 LangChain 的 invoke()/stream() 等方法；
• 用 RunnableLambda 包装后，这个函数就变成了标准的 LangChain 组件，能和 RunnablePassthrough、Prompt、LLM 等组件通过 | 运算符组合成链。

它的核心价值是自定义化——当 LangChain 内置组件满足不了你的需求时（比如特殊的数据处理逻辑），用 RunnableLambda 可以快速接入自定义代码。

二、基础用法（包装简单函数）

1. 包装匿名函数（lambda）

最基础的用法，适合简单的单行逻辑（比如数据格式转换、字段提取）：

from langchain_core.runnables import RunnableLambda

# 定义一个简单的匿名函数：将输入的question字段转为大写
uppercase_question = RunnableLambda(lambda x: x["question"].upper())

# 调用测试
input_data = {"question": "什么是RunnableLambda？"}
output = uppercase_question.invoke(input_data)
print(output)
# 输出结果：什么是RUNNABLELAMBDA？

2. 包装普通函数（多行逻辑）

适合复杂的自定义逻辑（比如数据清洗、计算、调用外部接口）：

from langchain_core.runnables import RunnableLambda

# 定义普通Python函数（复杂逻辑示例：处理问题并补充前缀）
def process_question(input_data):
    # 1. 提取原始问题
    question = input_data.get("question", "")
    # 2. 自定义处理：补充前缀 + 去空格
    processed = f"用户提问：{question.strip()}"
    # 3. 返回处理后的数据（可以是任意格式：字符串、字典、列表）
    return {"processed_question": processed}

# 用RunnableLambda包装函数
custom_processor = RunnableLambda(process_question)

# 调用测试
input_data = {"question": "  LangChain的RunnableLambda怎么用？  "}
output = custom_processor.invoke(input_data)
print(output)
# 输出结果：{'processed_question': '用户提问：LangChain的RunnableLambda怎么用？'}

三、进阶用法（融入LangChain链）

RunnableLambda 的核心价值是和其他组件组合，下面是两个实战场景：

场景1：自定义数据处理 → 调用LLM

先通过 RunnableLambda 处理输入，再传给 Prompt + LLM 生成回答：

from langchain_core.runnables import RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import os

# 1. 配置模型（需替换为你的API Key）
os.environ["OPENAI_API_KEY"] = "你的OpenAI API Key"
llm = ChatOpenAI(model="gpt-3.5-turbo")

# 2. 自定义处理函数：补充上下文
def add_context(input_data):
    question = input_data["question"]
    # 自定义逻辑：根据问题类型补充上下文
    if "Python" in question:
        context = "Python是一门解释型编程语言，适合新手学习"
    else:
        context = "通用知识，无需特殊上下文"
    return {
        "question": question,
        "context": context  # 新增上下文字段
    }

# 3. 包装函数为Runnable组件
context_processor = RunnableLambda(add_context)

# 4. 定义Prompt模板（使用question和context字段）
prompt = ChatPromptTemplate.from_template(
    "根据上下文回答问题：\n上下文：{context}\n问题：{question}"
)

# 5. 组合链：自定义处理 → Prompt → LLM
chain = context_processor | prompt | llm

# 6. 调用测试
result = chain.invoke({"question": "Python的优点是什么？"})
print(result.content)
# 输出示例：Python的优点包括语法简洁易读、生态丰富（有大量第三方库）、跨平台、适合快速开发...

场景2：LLM输出后 → 自定义解析

用 RunnableLambda 处理模型的输出（比如格式校验、数据提取）：

from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# 1. 初始化组件
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("列出3个学习Python的网站，用逗号分隔")
str_parser = StrOutputParser()  # 先把模型输出转为字符串

# 2. 自定义解析函数：将字符串转为列表
def parse_website_list(text):
    # 自定义逻辑：分割字符串 + 去空格 + 去空值
    websites = [item.strip() for item in text.split(",") if item.strip()]
    return {"websites": websites}

# 3. 包装解析函数
parse_processor = RunnableLambda(parse_website_list)

# 4. 组合链：透传输入 → Prompt → LLM → 转字符串 → 自定义解析
chain = RunnablePassthrough() | prompt | llm | str_parser | parse_processor

# 5. 调用测试
result = chain.invoke({})
print(result)
# 输出示例：{'websites': ['菜鸟教程', 'W3School', 'Python官方文档']}

四、关键注意事项（新手避坑）

1. 输入输出格式：
   • RunnableLambda 包装的函数，输入默认是链的上游输出（可以是字典、字符串、列表等）；
   • 函数的返回值会作为下游组件的输入，需确保格式和下游组件匹配（比如下游Prompt需要字典，就返回字典）。
2. 异常处理：
   建议在自定义函数中添加异常处理，避免链执行中断：

   def safe_processor(input_data):
       try:
           return input_data["question"].upper()
       except KeyError:
           return "未找到问题字段"
   

3. 流式输出：
   RunnableLambda 也支持流式输出（stream() 方法），但函数需返回可迭代对象（比如生成器）：

   def stream_processor(input_data):
       for char in input_data["question"]:
           yield char.upper()
   
   stream_runnable = RunnableLambda(stream_processor)
   for chunk in stream_runnable.stream({"question": "hello"}):
       print(chunk, end="")  # 输出：H E L L O
   

总结

1. 核心作用：RunnableLambda 是 LangChain 的“自定义逻辑适配器”，将普通 Python 函数封装为标准 Runnable 组件，支持融入链体系；
2. 使用场景：处理内置组件满足不了的自定义逻辑（数据处理、格式转换、外部接口调用等）；
3. 关键技巧：确保函数的输入输出格式与上下游组件匹配，复杂逻辑建议加异常处理。

它和 RunnablePassthrough 是互补的：RunnablePassthrough 负责通用的透传/简单加工，RunnableLambda 负责完全自定义的逻辑，结合使用能覆盖绝大多数数据处理场景。