RunnableLambda 是 LangChain 中自定義邏輯的核心組件，它允許你將任意 Python 函數(shù)（匿名函數(shù)/普通函數(shù)）封裝成 LangChain 的 Runnable 接口，從而無縫融入 LangChain 的鏈（Chain）體系中。簡單來說，它是“普通 Python 函數(shù)”和“LangChain 組件”之間的“適配器”。

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

可以把 RunnableLambda 理解成一個“函數(shù)包裝器”：

• 你寫的普通 Python 函數(shù)（比如數(shù)據(jù)處理、計算、格式轉換），本身不支持 LangChain 的 invoke()/stream() 等方法；
• 用 RunnableLambda 包裝后，這個函數(shù)就變成了標準的 LangChain 組件，能和 RunnablePassthrough、Prompt、LLM 等組件通過 | 運算符組合成鏈。

它的核心價值是自定義化——當 LangChain 內(nèi)置組件滿足不了你的需求時（比如特殊的數(shù)據(jù)處理邏輯），用 RunnableLambda 可以快速接入自定義代碼。

二、基礎用法（包裝簡單函數(shù)）

1. 包裝匿名函數(shù)（lambda）

最基礎的用法，適合簡單的單行邏輯（比如數(shù)據(jù)格式轉換、字段提取）：

from langchain_core.runnables import RunnableLambda

# 定義一個簡單的匿名函數(shù)：將輸入的question字段轉為大寫
uppercase_question = RunnableLambda(lambda x: x["question"].upper())

# 調(diào)用測試
input_data = {"question": "什么是RunnableLambda？"}
output = uppercase_question.invoke(input_data)
print(output)
# 輸出結果：什么是RUNNABLELAMBDA？

2. 包裝普通函數(shù)（多行邏輯）

適合復雜的自定義邏輯（比如數(shù)據(jù)清洗、計算、調(diào)用外部接口）：

from langchain_core.runnables import RunnableLambda

# 定義普通Python函數(shù)（復雜邏輯示例：處理問題并補充前綴）
def process_question(input_data):
    # 1. 提取原始問題
    question = input_data.get("question", "")
    # 2. 自定義處理：補充前綴 + 去空格
    processed = f"用戶提問：{question.strip()}"
    # 3. 返回處理后的數(shù)據(jù)（可以是任意格式：字符串、字典、列表）
    return {"processed_question": processed}

# 用RunnableLambda包裝函數(shù)
custom_processor = RunnableLambda(process_question)

# 調(diào)用測試
input_data = {"question": "  LangChain的RunnableLambda怎么用？  "}
output = custom_processor.invoke(input_data)
print(output)
# 輸出結果：{'processed_question': '用戶提問：LangChain的RunnableLambda怎么用？'}

三、進階用法（融入LangChain鏈）

RunnableLambda 的核心價值是和其他組件組合，下面是兩個實戰(zhàn)場景：

場景1：自定義數(shù)據(jù)處理 → 調(diào)用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. 自定義處理函數(shù)：補充上下文
def add_context(input_data):
    question = input_data["question"]
    # 自定義邏輯：根據(jù)問題類型補充上下文
    if "Python" in question:
        context = "Python是一門解釋型編程語言，適合新手學習"
    else:
        context = "通用知識，無需特殊上下文"
    return {
        "question": question,
        "context": context  # 新增上下文字段
    }

# 3. 包裝函數(shù)為Runnable組件
context_processor = RunnableLambda(add_context)

# 4. 定義Prompt模板（使用question和context字段）
prompt = ChatPromptTemplate.from_template(
    "根據(jù)上下文回答問題：\n上下文：{context}\n問題：{question}"
)

# 5. 組合鏈：自定義處理 → Prompt → LLM
chain = context_processor | prompt | llm

# 6. 調(diào)用測試
result = chain.invoke({"question": "Python的優(yōu)點是什么？"})
print(result.content)
# 輸出示例：Python的優(yōu)點包括語法簡潔易讀、生態(tài)豐富（有大量第三方庫）、跨平臺、適合快速開發(fā)...

場景2：LLM輸出后 → 自定義解析

用 RunnableLambda 處理模型的輸出（比如格式校驗、數(shù)據(jù)提?。?/p>

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的網(wǎng)站，用逗號分隔")
str_parser = StrOutputParser()  # 先把模型輸出轉為字符串

# 2. 自定義解析函數(shù)：將字符串轉為列表
def parse_website_list(text):
    # 自定義邏輯：分割字符串 + 去空格 + 去空值
    websites = [item.strip() for item in text.split(",") if item.strip()]
    return {"websites": websites}

# 3. 包裝解析函數(shù)
parse_processor = RunnableLambda(parse_website_list)

# 4. 組合鏈：透傳輸入 → Prompt → LLM → 轉字符串 → 自定義解析
chain = RunnablePassthrough() | prompt | llm | str_parser | parse_processor

# 5. 調(diào)用測試
result = chain.invoke({})
print(result)
# 輸出示例：{'websites': ['菜鳥教程', 'W3School', 'Python官方文檔']}

四、關鍵注意事項（新手避坑）

1. 輸入輸出格式：
   • RunnableLambda 包裝的函數(shù)，輸入默認是鏈的上游輸出（可以是字典、字符串、列表等）；
   • 函數(shù)的返回值會作為下游組件的輸入，需確保格式和下游組件匹配（比如下游Prompt需要字典，就返回字典）。
2. 異常處理：
   建議在自定義函數(shù)中添加異常處理，避免鏈執(zhí)行中斷：

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

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

   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 函數(shù)封裝為標準 Runnable 組件，支持融入鏈體系；
2. 使用場景：處理內(nèi)置組件滿足不了的自定義邏輯（數(shù)據(jù)處理、格式轉換、外部接口調(diào)用等）；
3. 關鍵技巧：確保函數(shù)的輸入輸出格式與上下游組件匹配，復雜邏輯建議加異常處理。

它和 RunnablePassthrough 是互補的：RunnablePassthrough 負責通用的透傳/簡單加工，RunnableLambda 負責完全自定義的邏輯，結合使用能覆蓋絕大多數(shù)數(shù)據(jù)處理場景。