ToolsNode 源码:LLM 返回 tool_calls 之后发生了什么(第53篇-E39)

系列「企业级 AI Agent 实现拆解」E39 篇，Part 9 源码深度篇第四章。上一篇拆了 Callback 系统。这篇拆工具执行：LLM 返回一条包含 tool_calls 字段的消息之后，Eino 是怎么把它变成实际的 Go 函数调用，最终返回 ToolMessage 的。

读完这篇你会知道

Tool 接口体系：BaseTool / InvokableTool / StreamableTool 的分工
ToolsNode
构建期做了什么：convTools → toolsTuple
genToolCallTasks
：从 LLM 输出提取任务列表的细节
并行执行：parallelRunToolCall 怎么用 goroutine
InferTool
：从 Go struct 自动推导 JSON Schema
HITL 中断：tool 可以抛出 interrupt 挂起等待人工审核

全景

LLM 响应（AssistantMessage）    ├── Role: "assistant"    └── ToolCalls: [            {ID:"call_1", Function:{Name:"search", Arguments:"{\"q\":\"eino\"}"}},            {ID:"call_2", Function:{Name:"calc",   Arguments:"{\"expr\":\"2+2\"}"}},        ]          │          ▼ToolsNode.Invoke(ctx, msg)    ├── genToolCallTasks()  ← 提取任务列表，处理别名和参数    ├── parallelRunToolCall()  ← goroutine 并行执行（默认）    │       ├── goroutine: search.InvokableRun(ctx, `{"q":"eino"}`)    │       └── 主goroutine: calc.InvokableRun(ctx, `{"expr":"2+2"}`)    └── 收集结果 → []*schema.Message{            ToolMessage("search result", "call_1"),            ToolMessage("4", "call_2"),        }          │          ▼追加到 messages 列表，继续下一轮 LLM 调用

Tool 接口体系

Eino 把 Tool 分成四个接口，按功能正交叠加：

// 最小接口：只提供元数据（给 LLM 看 schema 用）type BaseTool interface {    Info(ctx context.Context) (*schema.ToolInfo, error)}// 可执行工具（返回字符串）type InvokableTool interface {    BaseTool    InvokableRun(ctx context.Context, argumentsInJSON string, opts ...Option) (string, error)}// 流式工具（分 chunk 返回字符串）type StreamableTool interface {    BaseTool    StreamableRun(ctx context.Context, argumentsInJSON string, opts ...Option) (*schema.StreamReader[string], error)}// 增强工具（返回多模态内容：文本/图片/音频/文件）type EnhancedInvokableTool interface {    BaseTool    InvokableRun(ctx context.Context, toolArgument *schema.ToolArgument, opts ...Option) (*schema.ToolResult, error)}

优先级规则：ToolsNode 在执行时，如果一个 tool 同时实现了 Enhanced 和普通接口，Enhanced 优先。没有实现 StreamableTool 但实现了 InvokableTool 的，框架自动把它包装成流式（反之亦然）——所以你实现任意一个，两种执行模式都能用。

构建期：toolsTuple 是核心

NewToolNode(ctx, conf) 调用 convTools()，构建一个 toolsTuple：

type toolsTuple struct {    indexes        map[string]int              // tool 名 → 数组下标（O(1) 查找）    meta           []*executorMeta             // callback 元信息    endpoints      []InvokableToolEndpoint     // 普通执行入口    streamEndpoints []StreamableToolEndpoint   // 流式执行入口    // ... Enhanced 变种    canonicalNames []string  // 规范名称列表（支持 alias 时用）    toolInfos      []*schema.ToolInfo  // JSON Schema（用于 alias 验证）}

每个 tool 在构建时被封装成 endpoint 函数：

funcwrapToolCall(it InvokableTool, middlewares []Middleware, needCallback bool) InvokableToolEndpoint {    // 1. 如果需要 Callback，套一层 invokableToolWithCallback    if needCallback {        it = &invokableToolWithCallback{it: it}    }    // 2. 中间件逆序 wrap（最先注册的最外层）    return middleware(func(ctx, input) (*ToolOutput, error) {        result, err := it.InvokableRun(ctx, input.Arguments, input.CallOptions...)        return &ToolOutput{Result: result}, err    })}

关键点： 中间件在这里完成 wrap，每次执行只需直接调 endpoint(ctx, input)，不需要每次都遍历中间件列表。

执行期：从 ToolCalls 到任务列表

Invoke 核心流程：

func(tn *ToolsNode) Invoke(ctx, input *schema.Message, opts...) ([]*schema.Message, error) {    // 1. 从 AssistantMessage 提取任务列表    tasks, err := tn.genToolCallTasks(ctx, tuple, input, ...)    // 2. 并行或顺序执行    if tn.executeSequentially {        sequentialRunToolCall(ctx, runToolCallTaskByInvoke, tasks, ...)    } else {        parallelRunToolCall(ctx, runToolCallTaskByInvoke, tasks, ...)    }    // 3. 收集结果，构建 ToolMessage 列表    output := make([]*schema.Message, n)    for i := 0; i < n; i++ {        output[i] = schema.ToolMessage(tasks[i].output, tasks[i].callID, ...)    }    return output, nil}

genToolCallTasks 负责从 input.ToolCalls 生成任务列表，顺带处理：

参数别名
：remapArgs(args, aliasMap) 把模型返回的参数名 alias 映射到规范名
参数预处理
：toolArgumentsHandler(ctx, name, args) 允许在执行前改写参数
未知工具
：模型幻觉调了不存在的 tool → 走 unknownToolHandler（不设置则直接报错）

并行执行：goroutine 策略

LLM 一次可以返回多个 tool_calls（比如同时搜索 + 计算）。Eino 默认并行执行：

funcparallelRunToolCall(ctx, run, tasks, opts) {    if len(tasks) == 1 {        run(ctx, &tasks[0], opts...)  // 只有一个 → 不开 goroutine        return    }    var wg sync.WaitGroup    for i := 1; i < len(tasks); i++ {  // i=1 起，注意        wg.Add(1)        go func(t *toolCallTask) {            defer wg.Done()            defer func() {                if panicErr := recover(); panicErr != nil {                    t.err = safe.NewPanicErr(panicErr, debug.Stack())                }            }()            run(ctx, t, opts...)        }(ctx, &tasks[i], opts...)    }    run(ctx, &tasks[0], opts...)  // tasks[0] 在主 goroutine 执行    wg.Wait()}

设计细节：

tasks[0]
在主 goroutine 跑，tasks[1:] 各开一个 goroutine——减少一次 goroutine 创建开销
goroutine 里有 recover()——单个 tool panic 不会崩进程，错误存进 task.err
每个 task 独立写 task.output / task.err，不共享写，无锁

InferTool：从 Go struct 到 JSON Schema

最常用的 tool 创建方式——你只写业务逻辑，JSON Schema 自动推导：

type SearchInput struct {    Query  string `json:"q"    jsonschema:"description=搜索关键词,required"`    Limit  int    `json:"limit" jsonschema:"description=返回条数,default=10"`}type SearchOutput struct {    Results []string `json:"results"`}searchTool, err := utils.InferTool(    "web_search",    "搜索互联网内容，返回相关文档列表",    func(ctx context.Context, input SearchInput) (SearchOutput, error) {        // 实际业务逻辑        results := doSearch(input.Query, input.Limit)        return SearchOutput{Results: results}, nil    },)

InferTool 内部：

goStruct2ToolInfo[T]()
用反射 + jsonschema 标签把 SearchInput 转成 JSON Schema → 存进 schema.ToolInfo
返回的 InvokableTool 实现里，InvokableRun 会把 LLM 的 JSON 字符串 sonic.Unmarshal 成 SearchInput，再调你的函数，最后把 SearchOutputsonic.Marshal 成字符串返回

HITL 中断：tool 可以挂起

工具执行时可以抛出"需要人工审核"的 interrupt，让 Agent 暂停等待：

func(t *MyTool) InvokableRun(ctx context.Context, args string, opts ...tool.Option) (string, error) {    input := parseArgs(args)    if input.Amount > 10000 {        // 超额转账需要人工审批 → 抛出 interrupt，不是普通 error        return "", tool.NewInterruptError("需要人工审核：转账金额超过限额")    }    return doTransfer(input), nil}

ToolsNode 在收集结果时检测：

for i := 0; i < n; i++ {    if tasks[i].err != nil {        info, ok := IsInterruptRerunError(tasks[i].err)        if !ok {            return nil, fmt.Errorf("tool failed: %w", tasks[i].err)  // 普通错误        }        rerunExtra.RerunTools = append(rerunExtra.RerunTools, tasks[i].callID)        errs = append(errs, WrapInterruptAndRerunIfNeeded(...))  // HITL 错误    }}if len(errs) > 0 {    return nil, CompositeInterrupt(ctx, rerunExtra, rerunState, errs...)    // ↑ 已执行成功的 tool 结果存进 rerunState，恢复时不重复执行}

恢复时 ToolsNode 从 rerunState 读出已执行成功的结果，只重新执行被 interrupt 的那些——部分完成的 batch 不会全部重跑。

小结

ToolsNode 的核心设计：

构建期做重活
：convTools 把 tool 列表转成 name→index map + endpoint 函数，中间件在构建时 wrap，执行时零开销
执行时三步走
：genToolCallTasks（解析 LLM 输出）→ parallelRunToolCall（goroutine 并发）→ 收集结果成 ToolMessage 列表
自动互补
：只实现 InvokableTool 自动得到 StreamableTool，反之亦然
HITL 内置
：tool 抛 interrupt → 保存已执行结果 → Agent 暂停等待人工 → 恢复时只重跑 interrupt 的那部分

代码来源：eino/compose/tool_node.go · eino/components/tool/interface.go