[智能体-237]：LCEL 多节点各自独立调用工具实现方案

核心结论：

bind_tools依附 LLM，哪个节点用 LLM，就在当前节点的 LLM 单独绑定本节点专属工具；不同链路节点可以挂载不同工具集，各自隔离、互不干扰。 类比：CPU 不同运算流水线（不同节点）按需挂载不同外设驱动，A 流水线挂计算器网卡、B 流水线挂数据库网卡。

一、底层原理

1. bind_tools()生成全新 LLM 实例，互不污染原始 base_llm；
2. LCEL 每个节点是独立 Runnable，每个节点内的 LLM 可以单独绑定专属 tool_list；
3. 工具执行器（查表执行 tool）按节点隔离 / 全局共用二选一；
4. 节点 A LLM 绑定工具 A 组、节点 B LLM 绑定工具 B 组，各自只认识自己的工具。

二、完整示例：双节点链路，各自绑定不同工具

链路：输入预处理 → 节点A(LLM+计算器工具) → 节点B(LLM+字符串工具) → 结果组装

python

运行

from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import HumanMessage,AIMessage,ToolMessage from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableLambda # 1. 定义两套隔离工具 ## 节点A专用：数学计算工具 @tool def calc(a:float,b:float,opt:str): """加减乘除运算""" if opt=="+":return a+b elif opt=="-":return a-b elif opt=="*":return a*b elif opt=="/":return a/b ## 节点B专用：字符串工具 @tool def str_count(s:str): """统计字符长度""" return len(s) # 全局原始裸LLM，不绑定任何工具 base_llm = ChatOpenAI(model="gpt-3.5-turbo",temperature=0) # 2. 两个节点分别绑定各自工具，生成独立带工具LLM llm_nodeA = base_llm.bind_tools([calc]) # A节点只认识calc llm_nodeB = base_llm.bind_tools([str_count])# B节点只认识str_count # 工具注册表（全局统一执行） tool_dict = {"calc":calc,"str_count":str_count} # 通用工具执行封装Runnable：收到AIMessage就执行tool，回灌消息二次推理 def tool_run(msg:AIMessage): if not msg.tool_calls: return msg.content msg_list = [] for item in msg.tool_calls: fn = tool_dict[item["name"]] res = fn.invoke(item["args"]) msg_list.append(ToolMessage(str(res),tool_call_id=item["id"])) # 二次调用当前节点绑定过tools的LLM final = llm_nodeA.invoke([HumanMessage(""),msg]+msg_list) if "calc" in [i["name"] for i in msg.tool_calls] else llm_nodeB.invoke([HumanMessage(""),msg]+msg_list) return final.content # 定义所有chain的所有节点共享的可执行工具节点！！！ tool_exec = RunnableLambda(tool_run) # 3. 构造两段独立prompt+节点链路 promptA = ChatPromptTemplate.from_messages([("user","请计算：{query}")]) nodeA = promptA | llm_nodeA | tool_exec # 把工具节点插入到chainA中 promptB = ChatPromptTemplate.from_messages([("user","统计字符：{resA}")]) nodeB = promptB | llm_nodeB | tool_exec # 把工具节点插入到chainB中 # 完整串联全链路 full_chain = {"resA":nodeA} | nodeB # 调用：先算10+20，再统计结果字符串长度 res = full_chain.invoke({"query":"10+20"}) print(res)

运行逻辑：

1. 节点 A LLM 仅绑定 calc，自动调用加法得到30；
2. 结果流入节点 B，节点 B LLM 只绑定 str_count，统计"30"字符长度 = 2。

三、三种工程落地模式

模式 1：各节点独立绑定独立工具（隔离最强，推荐多业务场景）

• 原始 base_llm 不变；
• 每个节点内部：llm_n = base_llm.bind_tools(tools_n)；
• 优点：工具域隔离，A 节点不能调用 B 节点工具；
• 适用：分段业务、不同环节权限 / 工具不同。

模式 2：全局统一工具 LLM，全节点共用一套工具

全链路所有 LLM 节点共用同一个llm_all=base_llm.bind_tools(all_tools)； 所有节点都能调用全部工具，最简开发。

python

运行

llm_all = base_llm.bind_tools([calc,str_count]) nodeA = promptA|llm_all|tool_exec nodeB = promptB|llm_all|tool_exec

模式 3：动态按需临时绑定（运行时切换工具）

用Runnable.bind(tools=xxx)动态在链路局部临时挂载工具，不污染全局 LLM：

python

运行

# 仅当前Runnable临时附加工具配置 temp_llm = base_llm.bind(tools=[calc])

四、关键细节：为什么能做到节点隔离

1. bind_tools不修改原对象，每次返回新实例，各个节点的 llm 是不同内存对象、kwargs 里 tools 配置互相独立；
2. 模型侧：每次请求 HTTP body 里tools数组随各自 llm 配置变化，A 请求带 calc、B 请求带 str_count；
3. 执行侧：工具字典全局统一，或者按节点拆分 tool_map。

五、复杂 DAG 场景（RunnableMap 并行多分支，每分支独立工具）

python

运行

from langchain_core.runnables import RunnableMap # 两条并行分支，各自LLM+专属工具 branch_calc = promptA|llm_nodeA|tool_exec branch_str = promptB|llm_nodeB|tool_exec dag_chain = RunnableMap({ "计算结果":branch_calc, "字符统计":branch_str })

并行两个分支独立使用各自绑定的工具，互不干涉。

六、总结

1. LCEL 实现多节点差异化工具核心：每个节点内部单独对 base_llm 执行 bind_tools，生成本节点私有 LLM；
2. bind_tools 生成新实例是隔离的关键，原始裸 LLM 复用；
3. 工具决策在各节点 LLM，工具执行统一由后置 RunnableLambda 完成；
4. 类比硬件：同一块 CPU (base_llm)，不同流水线节点动态挂载不同外设驱动（tools）。