[ToolNode在LangGraph中的运用-03]定义在ToolNode的众多工具是如何执行的？

从“LangChain和LangGraph两种编程模式的同一性”和“工具调用的拦截和异常处理”提供的实例可以看出当我们传入携带ToolCall的AIMessage列表作为参数执行ToolNode后它会返回一个ToolMessage的列表。其实返回类型不仅于输入参数的类型有关还与工具函数的返回类型有关。1. ToolNode的输出类型作为一个Runnable对象ToolNode接收如下四种输入类型消息列表ToolCall列表字典且可以通过messages_key配置得到消息列表一个Pydantic模型且具有一个以messages_key配置命名的字段返回消息列表而ToolNode的则具有如下两种返回类型返回Command利用其update字段实现状态更新并利用goto字段实现跳转返回其他内容用于生成ToolMessage确定ToolNode的输出类型的规则主要有如下几条1.1 返回类型规则一在没有任何一个工具函数返回Command如果输入是一个消息列表那么ToolNode返回一个ToolMessage列表否则返回一个字典。如下的代码体现了这一规则fromtypingimportAnnotatedfromlangchain_core.toolsimporttoolfromlangchain_core.messagesimportAIMessage,ToolCall,AnyMessagefromlanggraph.prebuiltimportToolNodefromlanggraph.runtimeimportRuntimefromlangchain_core.runnablesimportRunnableConfigfromlanggraph._internal._constantsimportCONFIG_KEY_RUNTIMEfrompydanticimportBaseModelimportoperatorclassState(BaseModel):messages:Annotated[list[AnyMessage],operator.add]tooldeffoo(x:int,y:int)-str:A test tool foo.returnffoo({x},{y})tooldefbar(x:int,y:int)-str:A test tool bar.returnfbar({x},{y})tool_nodeToolNode(tools[foo,bar])tool_calls[ToolCall(namefoo,args{x:1,y:2},id001,typetool_call),ToolCall(namebar,args{x:3,y:4},id002,typetool_call),]messages[AIMessage(content,tool_callstool_calls)]config:RunnableConfig{configurable:{CONFIG_KEY_RUNTIME:Runtime(),}}resulttool_node.invoke(messages,configconfig)print(finput-messages:{result})resulttool_node.invoke(tool_calls,configconfig)print(finput-tool_calls:{result})resulttool_node.invoke({messages:messages},configconfig)print(finput-dict:{result})resulttool_node.invoke(State(messagesmessages),configconfig)# type: ignoreprint(finput-state:{result})输出input-messages: [ToolMessage(contentfoo(1, 2), namefoo, tool_call_id001), ToolMessage(contentbar(3, 4), namebar, tool_call_id002)] input-tool_calls: {messages: [ToolMessage(contentfoo(1, 2), namefoo, tool_call_id001), ToolMessage(contentbar(3, 4), namebar, tool_call_id002)]} input-dict: {messages: [ToolMessage(contentfoo(1, 2), namefoo, tool_call_id001), ToolMessage(contentbar(3, 4), namebar, tool_call_id002)]} input-state: {messages: [ToolMessage(contentfoo(1, 2), namefoo, tool_call_id001), ToolMessage(contentbar(3, 4), namebar, tool_call_id002)]}对于LangGraph来说节点返回的节点代表的是针对状态的增量更新所以字典携带的消息列表对应Key为messages_key配置默认为messages会自动添加到状态的消息列表中。如果返回的是一个ToolMessage列表:如果当前状态就是一个消息列表那么每个ToolMessage都会被添加到此列表中;否则会根据构造时设置的messages_key定位消息历史成员并将每个ToolMessage添加其中。对于返回的Command若它的update字段被设置成消息列表也是一样处理。1.1.2 返回类型规则二如果存在某一个或者多个工具函数返回Command对象如果update字段被设置为一个ToolMessage列表此时作为输入的只能时消息列表如果update字段被设置成一个字典此时对应的输入类型为其余三种在这种情况下ToolNode最终返回一个列表, 具体的工具函数根据返回值按照如下的规则添加结果如果Command将graph字段设置为Command.PARENT并将goto设置成一组Send实现针对父Agent某个节点的跳转这些Command将会被合并成一个这些Command的goto字段中的Send对象将会合并不去重合并后的Send列表作为最终Command对象的goto字段。合并后的Command会被添加到列表中。这一规则背后的含义是可以实现在父Agent中的跳转但不能改变器状态Command的update字段被丢弃其他的Command直接添加到列表中返回类型不是Command的工具函数会按照规则一生成ToolMessage列表或者字典并添加到列表中在如下的演示程序中我们让工具函数返回一个常规的Command对象其update字段被设置为字典最终返回的列表中将包含此Command对象。fromtypingimportAnnotatedfromlangchain_core.toolsimporttoolfromlanggraph.typesimportCommandfromlangchain_core.messagesimportAIMessage,ToolCall,ToolMessage,AnyMessagefromlanggraph.prebuiltimportToolNodefromlangchain_core.runnablesimportRunnableConfigfromlanggraph.runtimeimportRuntimefromlanggraph._internal._constantsimportCONFIG_KEY_RUNTIMEfrompydanticimportBaseModelimportoperatorclassState(BaseModel):messages:Annotated[list[AnyMessage],operator.add]tooldeffoo(x:int,y:int)-Command:A test tool foo.update{messages:[ToolMessage(contentffoo({x},{y}),tool_call_id001)]}returnCommand(updateupdate,gotonext)tooldefbar(x:int,y:int)-str:A test tool bar.returnfbar({x},{y})tool_nodeToolNode(tools[foo,bar])tool_calls[ToolCall(namefoo,args{x:1,y:2},id001,typetool_call),ToolCall(namebar,args{x:3,y:4},id002,typetool_call),]messages[AIMessage(content,tool_callstool_calls)]config:RunnableConfig{configurable:{CONFIG_KEY_RUNTIME:Runtime(),}}resulttool_node.invoke(tool_calls,configconfig)print(finput-tool_calls:{result})resulttool_node.invoke({messages:messages},configconfig)print(finput-dict:{result})resulttool_node.invoke(State(messagesmessages),configconfig)# type: ignoreprint(finput-state:{result})输出input-tool_calls: [Command(update{messages: [ToolMessage(contentfoo(1, 2), namefoo, tool_call_id001)]}, gotonext), {messages: [ToolMessage(contentbar(3, 4), namebar, tool_call_id002)]}] input-dict: [Command(update{messages: [ToolMessage(contentfoo(1, 2), namefoo, tool_call_id001)]}, gotonext), {messages: [ToolMessage(contentbar(3, 4), namebar, tool_call_id002)]}] input-state: [Command(update{messages: [ToolMessage(contentfoo(1, 2), namefoo, tool_call_id001)]}, gotonext), {messages: [ToolMessage(contentbar(3, 4), namebar, tool_call_id002)]}]然后我们让foo和bar这两个工具函数返回一个graph被设置成Command.PARENT的Command并将它们的goto字段设置成包含两个Send对象的列表最终我们会发现ToolNode的输出只有一个Command对象它的goto字段包含四个Send对象进行不去重合并。fromtypingimportAnnotatedfromlangchain_core.toolsimporttoolfromlanggraph.typesimportCommand,Sendfromlangchain_core.messagesimportAIMessage,ToolCall,ToolMessage,AnyMessagefromlanggraph.prebuiltimportToolNodefromlangchain_core.runnablesimportRunnableConfigfromlanggraph.runtimeimportRuntimefromlanggraph._internal._constantsimportCONFIG_KEY_RUNTIMEfrompydanticimportBaseModelimportoperatorclassState(BaseModel):messages:Annotated[list[AnyMessage],operator.add]tooldeffoo(x:int,y:int)-Command:A test tool foo.update{messages:[ToolMessage(contentffoo({x},{y}),tool_call_id001)]}goto[Send(nodeA,argNone),Send(nodeB,argNone)]returnCommand(updateupdate,gotogoto,graphCommand.PARENT)tooldefbar(x:int,y:int)-Command:A test tool bar.update{messages:[ToolMessage(contentffoo({x},{y}),tool_call_id001)]}goto[Send(nodeB,argNone),Send(nodeC,argNone)]returnCommand(updateupdate,gotogoto,graphCommand.PARENT)tool_nodeToolNode(tools[foo,bar])tool_calls[ToolCall(namefoo,args{x:1,y:2},id001,typetool_call),ToolCall(namebar,args{x:3,y:4},id002,typetool_call),]messages[AIMessage(content,tool_callstool_calls)]config:RunnableConfig{configurable:{CONFIG_KEY_RUNTIME:Runtime(),}}resulttool_node.invoke(tool_calls,configconfig)print(finput-tool_calls:{result})resulttool_node.invoke({messages:messages},configconfig)print(finput-dict:{result})resulttool_node.invoke(State(messagesmessages),configconfig)# type: ignoreprint(finput-state:{result})输出input-tool_calls: [Command(graph__parent__, goto[Send(nodeA, argNone), Send(nodeB, argNone), Send(nodeB, argNone), Send(nodeC, argNone)])] input-dict: [Command(graph__parent__, goto[Send(nodeA, argNone), Send(nodeB, argNone), Send(nodeB, argNone), Send(nodeC, argNone)])] input-state: [Command(graph__parent__, goto[Send(nodeA, argNone), Send(nodeB, argNone), Send(nodeB, argNone), Send(nodeC, argNone)])]2. 工具函数的并发执行从构造函数中针对基类RunnableCallable__init__方法的调用不难看出ToolNode的执行逻辑体现在它的_func和afunc方法上。classToolNode(RunnableCallable):name:strtoolsdef__init__(self,tools:Sequence[BaseTool|Callable],*,name:strtools,tags:list[str]|NoneNone,handle_tool_errors:bool|str|Callable[...,str]|type[Exception]|tuple[type[Exception],...]_default_handle_tool_errors,messages_key:strmessages,wrap_tool_call:ToolCallWrapper|NoneNone,awrap_tool_call:AsyncToolCallWrapper|NoneNone,)-None:super().__init__(self._func,self._afunc,namename,tagstags,traceFalse)同步执行工具节点的_func方法定义如下_afunc方法实现与之类似, 它首先会调用_parse_input提取最后一条AIMessage消息中携带的所有ToolCall对象以及当前输入的类型“list”, dict或者 “tool_calls”就是上面介绍的消息列表、ToolCall列表和字典含Pydantic模型。classToolNode(RunnableCallable):def_func(self,input:list[AnyMessage]|dict[str,Any]|BaseModel,config:RunnableConfig,runtime:Runtime,)-Any:tool_calls,input_typeself._parse_input(input)config_listget_config_list(config,len(tool_calls))tool_runtimes[]forcall,cfginzip(tool_calls,config_list,strictFalse):stateself._extract_state(input)tool_runtimeToolRuntime(statestate,tool_call_idcall[id],configcfg,contextruntime.context,storeruntime.store,stream_writerruntime.stream_writer,)tool_runtimes.append(tool_runtime)input_types[input_type]*len(tool_calls)withget_executor_for_config(config)asexecutor:outputslist(executor.map(self._run_one,tool_calls,input_types,tool_runtimes))returnself._combine_tool_outputs(outputs,input_type)接下来此方法为每个Tool_Call创建RunnableConfig配置和作为工具运行时的ToolRuntime对象然后从配置中提取预先设置的ThreadPoolExecutor对象以池线程池的方式并发调用每个工具。针对每个工具的调用实现在_run_one方法中它的执行逻辑很简单利用Tool_Call携带的工具名称找到对应的BaseTool对象调用其invoke方法完成对应工具函数的执行;工具函数返回的Command将直接作为_run_one方法的返回值返回的其他对象将被封装成返回的ToolMessage对象;最后调用_combine_tool_outputs方法按照上述的规则合并这些Command和ToolMessage并返回列表或者字典;