自定义函数加入链

RunnnamebeLambda

chain = first_prompt | model | json_parser | second_prompt | model | str_parser

前文我们根据JsonOutputParser完成了多模型执行链条的构建。

除了JsonOutputParser这类固定功能的解析器之外
我们也可以自己编写Lambda匿名函数来完成自定义逻辑的数据转换，想怎么转换就怎么转换，更自由。

想要完成这个功能，可以基于RunnableLambda类实现。

RunnableLambda类是LangChain内置的，将普通函数等转换为Runnable接口实例，方便自定义函数加入chain。

语法:

RunnableLambda(函数对象或lambda匿名函数)

补充

Lambda 函数基础语法：

lambda 参数列表: 表达式

代码示例

import os

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi
# from langchain_core.runnables import RunnableLambda
from dotenv import load_dotenv

load_dotenv()
LLM_API_KEY = os.getenv("LLM_API_KEY")

model = ChatTongyi(model="qwen3-max",api_key=LLM_API_KEY)
str_parser = StrOutputParser()

first_prompt = PromptTemplate.from_template(
    "我邻居姓：{lastname}，刚生了{gender}，请帮忙起名字，仅告知我一个名字，不要额外信息。"
)

second_prompt = PromptTemplate.from_template(
    "姓名：{name}，请帮我解析含义。"
)

# 函数的入参: AIMessage -> dict ({"name": "xxx"})
# my_func = RunnableLambda(lambda ai_msg: {"name": ai_msg.content})

chain = first_prompt | model | (lambda ai_msg: {"name": ai_msg.content}) | second_prompt | model | str_parser

for chunk in chain.stream({"lastname": "张", "gender": "女孩"}):
    print(chunk, end="", flush=True)

直接使用函数

chain = first_prompt | model | (lambda ai_msg: {"name": ai_msg.content}) | second_prompt | model | str_parser

跳过RunnableLambda类，直接让函数加入链也是可以的。因为Runnable接口类在实现__or__的时候，支持Callable接口的实例。

函数就是Callable接口的实例

def __or__(
    self,
    other: Runnable[Any, Other]
    | Callable[[Iterator[Any]], Iterator[Other]]
    | Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
    | Callable[[Any], Other]
    | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]:

如上代码示例，|符号（底层是调用__or__）组链，是支持函数加入的。

其本质是将函数自动转换为RunnableLambda

总结

如果像要在链中加入自定义函数，可以选择：

将函数封装入RunnableLambda类对象，其是Runnable接口实例，可以直接入链
直接将函数入链，函数会自动转换为RunnableLambda对象

RunnnamebeLambda​

代码示例​

直接使用函数​

总结​

RunnnamebeLambda

代码示例

直接使用函数

总结