From 1f250494920a789bb139d9502401a9d1da0e9565 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 10 Oct 2026 08:33:53 +0000 Subject: [PATCH] Carry tool CallID in loop events and add reasoning_split to the chat client Why: Parallel tool execution makes same-turn tool events interleave, so consumers can no longer pair starts and ends by order; the kernel Event needs the call ID. MiniMax interleaved-thinking models stream their reasoning inside content blocks unless the OpenAI-compatible request opts in with reasoning_split=true, which forces reasoning into the dedicated reasoning_content field the SSE layer already captures. Evidence: go test ./... green; new tests TestLoopToolEventsCarryCallID and TestOpenAIChatBodyReasoningSplit pin both behaviors (regression anchors for the backend parallel-work change). Impact: Additive Event.CallID (empty for non-tool events; existing consumers unaffected) and an opt-in Client.ReasoningSplit flag that only adds a request field when set. No default behavior change. Refs: Agent-System-Dev proposal 2026-10-10-agent-delegation-routing-sandbox.md (F2/F5), Dev task feature/agent-e1-parallel-fixes. --- runtime/llm/client.go | 9 ++++++ runtime/llm/reasoning_split_test.go | 47 +++++++++++++++++++++++++++ runtime/loop.go | 17 +++++----- runtime/loop_callid_test.go | 49 +++++++++++++++++++++++++++++ 4 files changed, 115 insertions(+), 7 deletions(-) create mode 100644 runtime/llm/reasoning_split_test.go create mode 100644 runtime/loop_callid_test.go diff --git a/runtime/llm/client.go b/runtime/llm/client.go index 9783bbc..8771124 100644 --- a/runtime/llm/client.go +++ b/runtime/llm/client.go @@ -20,6 +20,12 @@ type Client struct { HTTPClient *http.Client // Protocol 空 = openai-chat（零值兼容所有直接构造点）。 Protocol Protocol + // ReasoningSplit 在 openai-chat 请求体加 "reasoning_split": true： + // MiniMax 等 interleaved-thinking 提供商据此把思考放进独立的 + // reasoning_content 字段而不是混在 content 的 块里（分流后 + // SSE 的 reasoning_content 捕获照常生效）。对不认识该字段的服务器 + // 属未知字段，多数按忽略处理；宿主按模型名自行决定是否置位。 + ReasoningSplit bool } // Stream implements StreamFunc. Transport and provider failures are encoded as @@ -67,6 +73,9 @@ func (c *Client) run(ctx context.Context, model string, messages []Message, tool if effort := OpenAIReasoningEffort(opts.ThinkingLevel); effort != "" { body["reasoning_effort"] = effort } + if c.ReasoningSplit { + body["reasoning_split"] = true + } if len(tools) > 0 { body["tools"] = toProviderTools(tools) body["tool_choice"] = "auto" diff --git a/runtime/llm/reasoning_split_test.go b/runtime/llm/reasoning_split_test.go new file mode 100644 index 0000000..440a8c5 --- /dev/null +++ b/runtime/llm/reasoning_split_test.go @@ -0,0 +1,47 @@ +package llm + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// TestOpenAIChatBodyReasoningSplit 验证 Client.ReasoningSplit 置位时 +// openai-chat 请求体带 "reasoning_split": true，未置位时不带（未知字段 +// 不外溢给别的提供商）。 +func TestOpenAIChatBodyReasoningSplit(t *testing.T) { + var bodies []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + raw, _ := io.ReadAll(r.Body) + bodies = append(bodies, string(raw)) + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, "data: [DONE]\n\n") + })) + defer srv.Close() + + drain := func(c *Client) { + for range c.Stream(context.Background(), "MiniMax-M3", + []Message{{Role: RoleUser, Content: "hi"}}, nil, StreamOptions{}) { + } + } + + drain(&Client{BaseURL: srv.URL, APIKey: "k"}) + drain(&Client{BaseURL: srv.URL, APIKey: "k", ReasoningSplit: true}) + if len(bodies) != 2 { + t.Fatalf("requests = %d", len(bodies)) + } + if strings.Contains(bodies[0], "reasoning_split") { + t.Fatalf("flag off must not leak the field: %s", bodies[0]) + } + var got map[string]any + if err := json.Unmarshal([]byte(bodies[1]), &got); err != nil { + t.Fatal(err) + } + if v, ok := got["reasoning_split"].(bool); !ok || !v { + t.Fatalf("flag on must set reasoning_split=true: %s", bodies[1]) + } +} diff --git a/runtime/loop.go b/runtime/loop.go index 5155cab..c6f898e 100644 --- a/runtime/loop.go +++ b/runtime/loop.go @@ -38,6 +38,9 @@ type Event struct { Message llm.Message Tool string Text string + // CallID 在 tool_execution_start / tool_execution_end 时携带工具调用的 + // ID：并行执行下同回合多个调用并发进出，消费方按 ID 配对而非按顺序。 + CallID string // Detail carries tool arguments/results or reasoning deltas for // process-visualization consumers; log writers may ignore it. Detail string @@ -712,16 +715,16 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message { } if why := rejectIncompleteToolCall(call); why != "" { msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: why, IsError: true} - emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, Detail: call.Arguments}) - emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: why}) + emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, CallID: call.ID, Detail: call.Arguments}) + emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: why}) if cfg.AfterTool != nil { cfg.AfterTool(call, msg) } return msg } - emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, Detail: call.Arguments}) + emit(cfg, Event{Type: "tool_execution_start", Tool: call.Name, CallID: call.ID, Detail: call.Arguments}) if gated, msg, blocked := gateTool(ctx, cfg, call); blocked { - emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content}) + emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: msg.Content}) noteToolHook(cfg, call, msg) if cfg.AfterTool != nil { cfg.AfterTool(call, msg) @@ -733,7 +736,7 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message { if cfg.BeforeTool != nil { if err := cfg.BeforeTool(call); err != nil { msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: err.Error(), IsError: true} - emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content}) + emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: msg.Content}) noteToolHook(cfg, call, msg) if cfg.AfterTool != nil { cfg.AfterTool(call, msg) @@ -744,7 +747,7 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message { emit(cfg, Event{Type: "tool_execution_update", Tool: call.Name, Detail: call.Arguments}) if err := commitToolIntent(cfg, call); err != nil { msg := llm.Message{Role: llm.RoleTool, Name: call.Name, ToolCallID: call.ID, Content: err.Error(), IsError: true} - emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: msg.Content}) + emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: msg.Content}) return msg } endSpan := startSpan(cfg, "execute_tool "+call.Name, map[string]string{ @@ -769,7 +772,7 @@ func runOne(ctx context.Context, cfg Config, call llm.ToolCall) llm.Message { Content: res.Content, IsError: res.IsError, } - emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, Text: res.Content}) + emit(cfg, Event{Type: "tool_execution_end", Tool: call.Name, CallID: call.ID, Text: res.Content}) noteToolHook(cfg, call, msg) if cfg.AfterTool != nil { cfg.AfterTool(call, msg) diff --git a/runtime/loop_callid_test.go b/runtime/loop_callid_test.go new file mode 100644 index 0000000..cf72ca7 --- /dev/null +++ b/runtime/loop_callid_test.go @@ -0,0 +1,49 @@ +package runtime + +import ( + "context" + "testing" + + "agent-native/runtime/llm" + "agent-native/runtime/tools" +) + +// TestLoopToolEventsCarryCallID 验证 tool_execution_start/end 事件带工具调用 +// ID：并行执行下消费方按 ID 配对（回归锚点，防止字段被删后顺序假设回归）。 +func TestLoopToolEventsCarryCallID(t *testing.T) { + hits := 0 + reg := tools.NewRegistry(t.TempDir()) + reg.Register(hitTool{hits: &hits}) + var starts, ends []Event + res, err := Run(context.Background(), Config{ + Model: "fake", + Tools: reg, + Stream: script( + llm.Message{Role: llm.RoleAssistant, StopReason: llm.StopToolCalls, ToolCalls: []llm.ToolCall{{ + ID: "call-7", Name: "echo", Arguments: `{}`, + }}}, + llm.Message{Role: llm.RoleAssistant, Content: "done", StopReason: llm.StopEnd}, + ), + OnEvent: func(ev Event) { + switch ev.Type { + case "tool_execution_start": + starts = append(starts, ev) + case "tool_execution_end": + ends = append(ends, ev) + } + }, + }, llm.Message{Role: llm.RoleUser, Content: "go"}) + if err != nil { + t.Fatal(err) + } + if hits != 1 || len(starts) != 1 || len(ends) != 1 { + t.Fatalf("hits=%d starts=%d ends=%d", hits, len(starts), len(ends)) + } + if starts[0].CallID != "call-7" || starts[0].Tool != "echo" { + t.Fatalf("start = %+v", starts[0]) + } + if ends[0].CallID != "call-7" || ends[0].Tool != "echo" { + t.Fatalf("end = %+v", ends[0]) + } + _ = res +} -- 2.54.0