![[LangChain之链]RunnableCallable——将“自由定义”的函数变成标准组件](images/news1.jpg)
在众多针对Runnable的继承类型中个人认为最重要的莫过于RunnableCallable。有过LangChain开发经验的人都知道LangChain的很多组件都可以写成函数的形式比如工具、中间件、LangGraph的节点等而且它们的签名相对“自由”其参数不仅仅用于提供外部输入还可以用来注入一些运行时对象比如Runtime、BaseStore、StreamWriter和RunnableConfig等。它们很多都会转换成RunnableCallable对象来使用。1. 构造函数和RunnableLambda类似RunnableCallable用于封装提供的同步或者异步函数将它们转换成支持LCEL链的标准组件。从如下所示的代码片段可以看出我们提供的函数签名对输入参数和返回值都没有限制。classRunnableCallable(Runnable):def__init__(self,func:Callable[...,Any|Runnable]|None,afunc:Callable[...,Awaitable[Any|Runnable]]|NoneNone,*,name:str|NoneNone,tags:Sequence[str]|NoneNone,trace:boolTrue,recurse:boolTrue,explode_args:boolFalse,**kwargs:Any,)-None除了提供的用于执行目标操作的func和afunc参数外RunnableCallable的构造函数还提供了其他的关键字参数name给这个节点起个名方便在跟踪记录或日志里识别。如果没有指定func或者afunc的名称会作为其名称tags附加到跟踪记录的标签trace开启LangSmitch跟踪的开关。如果设为False这个单元的执行过程将不会出现在监控链路中适合高频、简单的逻辑以节省资源;recurse: 递归开关。如果函数返回的依然是一个Runnable是否继续自动执行它;explode_args:参数解包。如果设为True且输入是一个字典它会将字典解包为关键字参数传给func;kwargs: 预先填充的参数2. 针对Runtime和RunnableConfig的自动注入如果提供的函数包含Runtime和RunnableConfig类型的参数并且参数名分别设置为runtime和config调用invoke/aninvoke方发并指定config参数为一个RunnableConfig对象该对象将绑定为func/afunc的config参数RunnableConfig中保存的Runtime对应的Key为“__pregel_runtime”可以通过常量langgraph._internal._constants.CONFIG_KEY_RUNTIME得到它将绑定为runtime参数。这也是为什么config和runtime在很多地方作为保留参数的原因。如下的程序演示了针对这两个核心对象的参数注入。fromlanggraph._internal._runnableimportRunnableCallablefromlanggraph._internal._constantsimportCONFIG_KEY_RUNTIMEfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.runtimeimportRuntime global_config:RunnableConfig{configurable:{CONFIG_KEY_RUNTIME:Runtime()}}deftest_func(input:dict,config:RunnableConfig,runtime:Runtime)-dict:assertconfigisglobal_configassertruntimeisglobal_config.get(configurable,{}).get(CONFIG_KEY_RUNTIME)returninputrunnableRunnableCallable(functest_func)resultrunnable.invoke(input{foo:bar},configglobal_config)assertresult{foo:bar}3. 针对其他参数的手工注入除了针对Runtime和RunnableConfig的自动注入封装函数的其他参数也可以通过在构造函数预填充的参数或者调用invoke/aninvoke方法指定的关键字参数进行绑定。在如下这个演示实例中test_func中的stream_writer和store参数就是分别通过这两种方式填充的。当我们调用StateGraph的add_conditional_edges方法添加“条件边”path参数指定的用于解析分支路径的函数中的store和stream_writer参数就是采用这种方式注入的。fromlanggraph._internal._runnableimportRunnableCallablefromlanggraph._internal._constantsimportCONFIG_KEY_RUNTIMEfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.runtimeimportRuntimefromlanggraph.store.baseimportBaseStorefromlanggraph.typesimportStreamWriterfromtypingimportcast global_config:RunnableConfig{configurable:{CONFIG_KEY_RUNTIME:Runtime()}}deftest_func(input:dict,config:RunnableConfig,stream_writer:StreamWriter,store:BaseStore)-dict:runtimecast(Runtime,config.get(configurable,{}).get(CONFIG_KEY_RUNTIME))assertstream_writerisruntime.stream_writerassertstoreisruntime.storereturninputruntimecast(Runtime,global_config.get(configurable,{}).get(CONFIG_KEY_RUNTIME))runnableRunnableCallable(functest_func,stream_writerruntime.stream_writer)resultrunnable.invoke(input{foo:bar},configglobal_config,storeruntime.store)assertresult{foo:bar}4. 递归执行在默认情况下如果指定函数返回一个Runnable执行RunnableCallable时会以递归的防止执行它。我们可以在创建RunnableCallable时利用recurse参数关闭这一特性。RunnableCallable递归执行Runnable的能力体现在如下的演示程序中。fromlanggraph._internal._runnableimportRunnableCallablefromlangchain_core.runnablesimportRunnable log:list[str][]deffoo(input:dict)-Runnable:log.append(foo)returnRunnableCallable(bar)defbar(input:dict)-Runnable:log.append(bar)returnRunnableCallable(baz)defbaz(input:dict)-dict:log.append(baz)returninputrunnableRunnableCallable(funcfoo)resultrunnable.invoke(input{foo:bar})assertresult{foo:bar}assertlog[foo,bar,baz]log.clear()runnableRunnableCallable(funcfoo,recurseFalse)resultrunnable.invoke(input{foo:bar})assertisinstance(result,Runnable)assertlog[foo]5. 参数拆包如果调用构造函数时将explode_args参数设置为True意味着调用invoke/ainvoke时指定的输入参数在传入封装的函数前会先进行拆包。如下的程序演示了这一点fromlanggraph._internal._runnableimportRunnableCallabledeffoobar(data:str,*,foo:str,bar:str)-dict:return{data:data,foo:foo,bar:bar}input((foobar,),{foo:123,bar:456})runnableRunnableCallable(funcfoobar,explode_argsTrue)resultrunnable.invoke(input)assertresult{data:foobar,foo:123,bar:456}runnableRunnableCallable(foobar)try:resultrunnable.invoke(input)assertFalse,Expected TypeError due to missing argumentsexceptExceptionase:assertisinstance(e,TypeError)assertstr(e)foobar() missing 2 required keyword-only arguments: foo and bar
[LangChain之链]RunnableCallable——将“自由定义”的函数变成标准组件
在众多针对Runnable的继承类型中个人认为最重要的莫过于RunnableCallable。有过LangChain开发经验的人都知道LangChain的很多组件都可以写成函数的形式比如工具、中间件、LangGraph的节点等而且它们的签名相对“自由”其参数不仅仅用于提供外部输入还可以用来注入一些运行时对象比如Runtime、BaseStore、StreamWriter和RunnableConfig等。它们很多都会转换成RunnableCallable对象来使用。1. 构造函数和RunnableLambda类似RunnableCallable用于封装提供的同步或者异步函数将它们转换成支持LCEL链的标准组件。从如下所示的代码片段可以看出我们提供的函数签名对输入参数和返回值都没有限制。classRunnableCallable(Runnable):def__init__(self,func:Callable[...,Any|Runnable]|None,afunc:Callable[...,Awaitable[Any|Runnable]]|NoneNone,*,name:str|NoneNone,tags:Sequence[str]|NoneNone,trace:boolTrue,recurse:boolTrue,explode_args:boolFalse,**kwargs:Any,)-None除了提供的用于执行目标操作的func和afunc参数外RunnableCallable的构造函数还提供了其他的关键字参数name给这个节点起个名方便在跟踪记录或日志里识别。如果没有指定func或者afunc的名称会作为其名称tags附加到跟踪记录的标签trace开启LangSmitch跟踪的开关。如果设为False这个单元的执行过程将不会出现在监控链路中适合高频、简单的逻辑以节省资源;recurse: 递归开关。如果函数返回的依然是一个Runnable是否继续自动执行它;explode_args:参数解包。如果设为True且输入是一个字典它会将字典解包为关键字参数传给func;kwargs: 预先填充的参数2. 针对Runtime和RunnableConfig的自动注入如果提供的函数包含Runtime和RunnableConfig类型的参数并且参数名分别设置为runtime和config调用invoke/aninvoke方发并指定config参数为一个RunnableConfig对象该对象将绑定为func/afunc的config参数RunnableConfig中保存的Runtime对应的Key为“__pregel_runtime”可以通过常量langgraph._internal._constants.CONFIG_KEY_RUNTIME得到它将绑定为runtime参数。这也是为什么config和runtime在很多地方作为保留参数的原因。如下的程序演示了针对这两个核心对象的参数注入。fromlanggraph._internal._runnableimportRunnableCallablefromlanggraph._internal._constantsimportCONFIG_KEY_RUNTIMEfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.runtimeimportRuntime global_config:RunnableConfig{configurable:{CONFIG_KEY_RUNTIME:Runtime()}}deftest_func(input:dict,config:RunnableConfig,runtime:Runtime)-dict:assertconfigisglobal_configassertruntimeisglobal_config.get(configurable,{}).get(CONFIG_KEY_RUNTIME)returninputrunnableRunnableCallable(functest_func)resultrunnable.invoke(input{foo:bar},configglobal_config)assertresult{foo:bar}3. 针对其他参数的手工注入除了针对Runtime和RunnableConfig的自动注入封装函数的其他参数也可以通过在构造函数预填充的参数或者调用invoke/aninvoke方法指定的关键字参数进行绑定。在如下这个演示实例中test_func中的stream_writer和store参数就是分别通过这两种方式填充的。当我们调用StateGraph的add_conditional_edges方法添加“条件边”path参数指定的用于解析分支路径的函数中的store和stream_writer参数就是采用这种方式注入的。fromlanggraph._internal._runnableimportRunnableCallablefromlanggraph._internal._constantsimportCONFIG_KEY_RUNTIMEfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.runtimeimportRuntimefromlanggraph.store.baseimportBaseStorefromlanggraph.typesimportStreamWriterfromtypingimportcast global_config:RunnableConfig{configurable:{CONFIG_KEY_RUNTIME:Runtime()}}deftest_func(input:dict,config:RunnableConfig,stream_writer:StreamWriter,store:BaseStore)-dict:runtimecast(Runtime,config.get(configurable,{}).get(CONFIG_KEY_RUNTIME))assertstream_writerisruntime.stream_writerassertstoreisruntime.storereturninputruntimecast(Runtime,global_config.get(configurable,{}).get(CONFIG_KEY_RUNTIME))runnableRunnableCallable(functest_func,stream_writerruntime.stream_writer)resultrunnable.invoke(input{foo:bar},configglobal_config,storeruntime.store)assertresult{foo:bar}4. 递归执行在默认情况下如果指定函数返回一个Runnable执行RunnableCallable时会以递归的防止执行它。我们可以在创建RunnableCallable时利用recurse参数关闭这一特性。RunnableCallable递归执行Runnable的能力体现在如下的演示程序中。fromlanggraph._internal._runnableimportRunnableCallablefromlangchain_core.runnablesimportRunnable log:list[str][]deffoo(input:dict)-Runnable:log.append(foo)returnRunnableCallable(bar)defbar(input:dict)-Runnable:log.append(bar)returnRunnableCallable(baz)defbaz(input:dict)-dict:log.append(baz)returninputrunnableRunnableCallable(funcfoo)resultrunnable.invoke(input{foo:bar})assertresult{foo:bar}assertlog[foo,bar,baz]log.clear()runnableRunnableCallable(funcfoo,recurseFalse)resultrunnable.invoke(input{foo:bar})assertisinstance(result,Runnable)assertlog[foo]5. 参数拆包如果调用构造函数时将explode_args参数设置为True意味着调用invoke/ainvoke时指定的输入参数在传入封装的函数前会先进行拆包。如下的程序演示了这一点fromlanggraph._internal._runnableimportRunnableCallabledeffoobar(data:str,*,foo:str,bar:str)-dict:return{data:data,foo:foo,bar:bar}input((foobar,),{foo:123,bar:456})runnableRunnableCallable(funcfoobar,explode_argsTrue)resultrunnable.invoke(input)assertresult{data:foobar,foo:123,bar:456}runnableRunnableCallable(foobar)try:resultrunnable.invoke(input)assertFalse,Expected TypeError due to missing argumentsexceptExceptionase:assertisinstance(e,TypeError)assertstr(e)foobar() missing 2 required keyword-only arguments: foo and bar
