Bound parallel tool dispatch
This commit is contained in:
parent
6a8e897fa5
commit
062642dfd3
|
|
@ -15,7 +15,10 @@ import (
|
||||||
"ollie/util"
|
"ollie/util"
|
||||||
)
|
)
|
||||||
|
|
||||||
const maxTransientRetries = 3
|
const (
|
||||||
|
maxTransientRetries = 3
|
||||||
|
maxParallelToolCalls = 16
|
||||||
|
)
|
||||||
|
|
||||||
// retryBaseDelay is the base delay for rate-limit retries (5<<attempt seconds).
|
// retryBaseDelay is the base delay for rate-limit retries (5<<attempt seconds).
|
||||||
// Tests can override this to speed up retry tests.
|
// Tests can override this to speed up retry tests.
|
||||||
|
|
@ -162,31 +165,31 @@ func (ag *Agent) run(ctx context.Context) error {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
msg := backend.Message{ID: responseID, Role: "assistant", Content: content, Reasoning: reasoning, ToolCalls: toolCalls}
|
msg := backend.Message{ID: responseID, Role: "assistant", Content: content, Reasoning: reasoning, ToolCalls: toolCalls}
|
||||||
results, interrupted := ag.execToolCalls(ctx, toolCalls)
|
results, interrupted := ag.execToolCalls(ctx, toolCalls)
|
||||||
|
|
||||||
h.update(msg, results)
|
h.update(msg, results)
|
||||||
ag.save()
|
ag.save()
|
||||||
|
|
||||||
if interrupted {
|
if interrupted {
|
||||||
return ctx.Err()
|
return ctx.Err()
|
||||||
}
|
|
||||||
if len(toolCalls) == 0 {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
|
|
||||||
if abort := ag.trackErrors(ctx, results, &errs, step); abort != nil {
|
|
||||||
return abort
|
|
||||||
}
|
|
||||||
|
|
||||||
_, compactErr := ag.autoCompact(ctx)
|
|
||||||
if compactErr != nil {
|
|
||||||
ag.emit(Event{Role: "error", Content: compactErr.Error()})
|
|
||||||
break
|
|
||||||
}
|
|
||||||
|
|
||||||
step++
|
|
||||||
}
|
}
|
||||||
|
if len(toolCalls) == 0 {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
|
if abort := ag.trackErrors(ctx, results, &errs, step); abort != nil {
|
||||||
|
return abort
|
||||||
|
}
|
||||||
|
|
||||||
|
_, compactErr := ag.autoCompact(ctx)
|
||||||
|
if compactErr != nil {
|
||||||
|
ag.emit(Event{Role: "error", Content: compactErr.Error()})
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
|
step++
|
||||||
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
@ -467,6 +470,7 @@ func (ag *Agent) execBatch(ctx context.Context, batch []backend.ToolCall) ([]too
|
||||||
uniqueResults := make([]inflightResult, len(batch))
|
uniqueResults := make([]inflightResult, len(batch))
|
||||||
|
|
||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
|
sem := make(chan struct{}, maxParallelToolCalls)
|
||||||
for k, call := range batch {
|
for k, call := range batch {
|
||||||
key := call.Name + "\x00" + string(call.Arguments)
|
key := call.Name + "\x00" + string(call.Arguments)
|
||||||
if _, dup := inflight[key]; dup {
|
if _, dup := inflight[key]; dup {
|
||||||
|
|
@ -476,6 +480,13 @@ func (ag *Agent) execBatch(ctx context.Context, batch []backend.ToolCall) ([]too
|
||||||
wg.Add(1)
|
wg.Add(1)
|
||||||
go func(k int, call backend.ToolCall) {
|
go func(k int, call backend.ToolCall) {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
|
select {
|
||||||
|
case sem <- struct{}{}:
|
||||||
|
case <-ctx.Done():
|
||||||
|
uniqueResults[k].wasInt = true
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer func() { <-sem }()
|
||||||
uniqueResults[k].tr, uniqueResults[k].wasInt = ag.execOne(ctx, call)
|
uniqueResults[k].tr, uniqueResults[k].wasInt = ag.execOne(ctx, call)
|
||||||
}(k, call)
|
}(k, call)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue