Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7f60baeaf8 | ||
|
|
ef2fa75006 |
+36
-1
@@ -69,6 +69,9 @@ type Config struct {
|
|||||||
Trace func(name string, attrs map[string]string) func()
|
Trace func(name string, attrs map[string]string) func()
|
||||||
MaxTurns int
|
MaxTurns int
|
||||||
ToolExecution string
|
ToolExecution string
|
||||||
|
// SerialTools 是并行执行下的顺序敏感工具名集合:这些调用按模型给出的
|
||||||
|
// 原顺序在调用方 goroutine 逐个执行(互斥且保序),其余调用并发执行。
|
||||||
|
SerialTools map[string]bool
|
||||||
GetSteering func() []llm.Message
|
GetSteering func() []llm.Message
|
||||||
GetFollowUp func() []llm.Message
|
GetFollowUp func() []llm.Message
|
||||||
// OnEvent 观察循环事件。同一次 Run 内不会并发调用
|
// OnEvent 观察循环事件。同一次 Run 内不会并发调用
|
||||||
@@ -597,7 +600,10 @@ func serializeHostCallbacks(cfg *Config) {
|
|||||||
|
|
||||||
func executeTools(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message {
|
func executeTools(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message {
|
||||||
if cfg.ToolExecution == ExecParallel && len(assistant.ToolCalls) > 1 {
|
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))
|
out := make([]llm.Message, 0, len(assistant.ToolCalls))
|
||||||
for _, call := range assistant.ToolCalls {
|
for _, call := range assistant.ToolCalls {
|
||||||
@@ -610,6 +616,35 @@ func executeTools(ctx context.Context, cfg Config, assistant llm.Message) []llm.
|
|||||||
return out
|
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 {
|
func executeParallel(ctx context.Context, cfg Config, assistant llm.Message) []llm.Message {
|
||||||
out := make([]llm.Message, len(assistant.ToolCalls))
|
out := make([]llm.Message, len(assistant.ToolCalls))
|
||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user