diff --git a/runtime/loop.go b/runtime/loop.go index 72426ac..980a52f 100644 --- a/runtime/loop.go +++ b/runtime/loop.go @@ -69,6 +69,9 @@ type Config struct { Trace func(name string, attrs map[string]string) func() MaxTurns int ToolExecution string + // SerialTools 是并行执行下的顺序敏感工具名集合:这些调用按模型给出的 + // 原顺序在调用方 goroutine 逐个执行(互斥且保序),其余调用并发执行。 + SerialTools map[string]bool GetSteering func() []llm.Message GetFollowUp func() []llm.Message // OnEvent 观察循环事件。同一次 Run 内不会并发调用 @@ -597,7 +600,10 @@ func serializeHostCallbacks(cfg *Config) { func executeTools(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message { if cfg.ToolExecution == ExecParallel && len(assistant.ToolCalls) > 1 { - return executeParallel(ctx, cfg, assistant) + if len(cfg.SerialTools) == 0 { + return executeParallel(ctx, cfg, assistant) + } + return executeMixed(ctx, cfg, assistant) } out := make([]llm.Message, 0, len(assistant.ToolCalls)) for _, call := range assistant.ToolCalls { @@ -610,6 +616,35 @@ func executeTools(ctx context.Context, cfg Config, assistant llm.Message) []llm. return out } +// executeMixed 并行执行非串行调用,串行名(SerialTools)按模型给出的原顺序 +// 在调用方 goroutine 逐个执行——互斥且保序(卡片流等有先后依赖的工具不能只 +// 靠互斥:并发派发下先抢到锁的不一定是先声明的调用)。 +func executeMixed(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message { + out := make([]llm.Message, len(assistant.ToolCalls)) + var wg sync.WaitGroup + serialIdx := make([]int, 0, len(assistant.ToolCalls)) + for i, call := range assistant.ToolCalls { + if cfg.SerialTools[call.Name] { + serialIdx = append(serialIdx, i) + continue + } + wg.Add(1) + go func(i int, call llm.ToolCall) { + defer wg.Done() + out[i] = runOne(ctx, cfg, call) + }(i, call) + } + for _, i := range serialIdx { + if err := ctx.Err(); err != nil { + out[i] = cancelledToolMessage(assistant.ToolCalls[i], err) + continue + } + out[i] = runOne(ctx, cfg, assistant.ToolCalls[i]) + } + wg.Wait() + return out +} + func executeParallel(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message { out := make([]llm.Message, len(assistant.ToolCalls)) var wg sync.WaitGroup diff --git a/runtime/loop_serial_partition_test.go b/runtime/loop_serial_partition_test.go new file mode 100644 index 0000000..280a3fa --- /dev/null +++ b/runtime/loop_serial_partition_test.go @@ -0,0 +1,72 @@ +package runtime + +import ( + "context" + "encoding/json" + "sync" + "testing" + + "agent-native/runtime/llm" + "agent-native/runtime/tools" +) + +// orderTool 记录自己的执行序号(并发安全)。 +type orderTool struct { + name string + mu sync.Mutex + seq []string +} + +func (o *orderTool) Name() string { return o.name } +func (o *orderTool) Description() string { return o.name } +func (o *orderTool) Parameters() json.RawMessage { return json.RawMessage(`{"type":"object"}`) } +func (o *orderTool) Execute(ctx context.Context, _ json.RawMessage) (tools.Result, error) { + id := o.name + if call, ok := ToolCallFromContext(ctx); ok { + id = call.ID + } + o.mu.Lock() + o.seq = append(o.seq, id) + o.mu.Unlock() + return tools.Result{Content: "ran " + id}, nil +} + +// TestLoopSerialToolsKeepOrder 并行模式下 SerialTools 命中的调用按声明顺序执行 +// (互斥不保序不够:卡片流等有先后依赖),未命中的并发执行。 +func TestLoopSerialToolsKeepOrder(t *testing.T) { + cards := &orderTool{name: "card"} + reads := &orderTool{name: "read"} + reg := tools.NewRegistry(t.TempDir()) + reg.Register(cards) + reg.Register(reads) + res, err := Run(context.Background(), Config{ + Model: "fake", Tools: reg, + ToolExecution: ExecParallel, + SerialTools: map[string]bool{"card": true}, + Stream: script( + llm.Message{Role: llm.RoleAssistant, StopReason: llm.StopToolCalls, ToolCalls: []llm.ToolCall{ + {ID: "s1", Name: "card", Arguments: `{}`}, + {ID: "p1", Name: "read", Arguments: `{}`}, + {ID: "s2", Name: "card", Arguments: `{}`}, + {ID: "s3", Name: "card", Arguments: `{}`}, + }}, + llm.Message{Role: llm.RoleAssistant, Content: "done", StopReason: llm.StopEnd}, + ), + }, llm.Message{Role: llm.RoleUser, Content: "go"}) + if err != nil { + t.Fatal(err) + } + cards.mu.Lock() + got := append([]string(nil), cards.seq...) + cards.mu.Unlock() + if len(got) != 3 || got[0] != "s1" || got[1] != "s2" || got[2] != "s3" { + t.Fatalf("card executions must follow declaration order s1→s2→s3, got %v", got) + } + reads.mu.Lock() + nRead := len(reads.seq) + reads.mu.Unlock() + if nRead != 1 { + t.Fatalf("read executions = %d", nRead) + } + _ = res +}