Skip to content

Commit d0c6613

Browse files
committed
feat(loop): parallel tool execution with bounded concurrency
Add MaxToolParallel config to run tool calls concurrently when the LLM returns multiple tool calls in one response (supported by Claude 3.5+, GPT-4o, DeepSeek V4). Changes: - loop.go: Replace sequential for-loop with 3-phase parallel execution: Phase 1 — fire all tool_call events synchronously (rendering) Phase 2 — execute tools in goroutines bounded by a channel semaphore (default cap 4, configurable via MaxToolParallel) Phase 3 — process results in order (compress, wrap, append to messages) - odek.go: Add MaxToolParallel to public Config, wire to engine - loader.go: Add max_tool_parallel to FileConfig/ResolvedConfig, merge chain - telegram.go, serve.go, main.go: Wire MaxToolParallel from resolved config I/O-bound tools (read_file, search_files, web_search) benefit most from concurrent execution, reducing per-iteration latency from sum-of-latencies to max-latency for parallel tool calls.
1 parent 54fc206 commit d0c6613

6 files changed

Lines changed: 90 additions & 23 deletions

File tree

cmd/odek/main.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -953,6 +953,7 @@ func run(args []string) error {
953953
BaseURL: resolved.BaseURL,
954954
APIKey: resolved.APIKey,
955955
MaxIterations: resolved.MaxIter,
956+
MaxToolParallel: resolved.MaxToolParallel,
956957
SystemMessage: systemMessage,
957958
NoProjectFile: resolved.NoAgents,
958959
Thinking: resolved.Thinking,
@@ -1864,6 +1865,7 @@ func continueCmd(args []string) error {
18641865
BaseURL: resolved.BaseURL,
18651866
APIKey: resolved.APIKey,
18661867
MaxIterations: resolved.MaxIter,
1868+
MaxToolParallel: resolved.MaxToolParallel,
18671869
SystemMessage: systemMessage,
18681870
NoProjectFile: resolved.NoAgents,
18691871
Thinking: resolved.Thinking,

cmd/odek/serve.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -245,6 +245,7 @@ func newServeAgent(resolved config.ResolvedConfig, system string, sendFn func(v
245245
BaseURL: resolved.BaseURL,
246246
APIKey: resolved.APIKey,
247247
MaxIterations: resolved.MaxIter,
248+
MaxToolParallel: resolved.MaxToolParallel,
248249
SystemMessage: system,
249250
NoProjectFile: resolved.NoAgents,
250251
Thinking: resolved.Thinking,

cmd/odek/telegram.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1116,6 +1116,7 @@ func handleChatMessage(
11161116
BaseURL: resolved.BaseURL,
11171117
APIKey: resolved.APIKey,
11181118
MaxIterations: resolved.MaxIter,
1119+
MaxToolParallel: resolved.MaxToolParallel,
11191120
SystemMessage: systemMessage,
11201121
RuntimeContext: odek.BuildRuntimeContext("telegram"),
11211122
InteractionMode: resolved.InteractionMode,

internal/config/loader.go

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -145,6 +145,11 @@ type FileConfig struct {
145145
// Default: 3.
146146
MaxConcurrency int `json:"max_concurrency,omitempty"`
147147

148+
// MaxToolParallel limits how many tool calls run concurrently per
149+
// agent iteration. Config: max_tool_parallel.
150+
// Default: 0 (loop uses default of 4).
151+
MaxToolParallel int `json:"max_tool_parallel,omitempty"`
152+
148153
// Telegram configures the Telegram bot integration.
149154
Telegram *telegram.TelegramConfig `json:"telegram,omitempty"`
150155

@@ -240,6 +245,11 @@ type ResolvedConfig struct {
240245
// Default: 3.
241246
MaxConcurrency int
242247

248+
// MaxToolParallel limits how many tool calls run concurrently per
249+
// agent iteration. Config: max_tool_parallel.
250+
// Default: 0 (loop uses default of 4).
251+
MaxToolParallel int
252+
243253
// Telegram is the resolved Telegram bot configuration.
244254
Telegram telegram.TelegramConfig
245255

@@ -566,6 +576,9 @@ func LoadConfig(cli CLIFlags) ResolvedConfig {
566576
resolved.MaxConcurrency = 3
567577
}
568578

579+
// MaxToolParallel: 0 = use loop engine default (4)
580+
resolved.MaxToolParallel = cfg.MaxToolParallel
581+
569582
// Booleans: default to false if not set
570583
if cfg.Sandbox != nil {
571584
resolved.Sandbox = *cfg.Sandbox

internal/loop/loop.go

Lines changed: 64 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,14 @@ type Engine struct {
9191
// dedicated "system" field for Anthropic compatibility.
9292
PromptCaching bool
9393

94+
// MaxToolParallel controls how many tool calls run concurrently per
95+
// iteration. 0 = use default (4). Models that support parallel tool
96+
// calling (Claude 3.5+, GPT-4o, DeepSeek V4) can emit multiple tool
97+
// calls in one response — this setting bounds concurrency so tools
98+
// like read_file, search_files, and web_search run in parallel while
99+
// avoiding resource exhaustion.
100+
MaxToolParallel int
101+
94102
// Token accounting — accumulated across all iterations of the most recent run.
95103
// Reset on each Run/RunWithMessages call and read by callers (e.g. WebUI).
96104
TotalInputTokens int
@@ -151,6 +159,10 @@ func (e *Engine) SetNarrator(n *narrate.Narrator) { e.narrator = n }
151159
// If nil, no callback is fired.
152160
func (e *Engine) SetIterationCallback(cb IterationCallback) { e.iterationCallback = cb }
153161

162+
// SetMaxToolParallel sets the maximum concurrency for tool execution per
163+
// iteration. 0 or negative = use default (4).
164+
func (e *Engine) SetMaxToolParallel(n int) { e.MaxToolParallel = n }
165+
154166
// ── Token Estimation ─────────────────────────────────────────────────
155167
//
156168
// Zero-dependency heuristic: 1 token ≈ 4 chars for English text.
@@ -552,12 +564,13 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [
552564
}
553565
messages = append(messages, assistantMsg)
554566

555-
// ACT: execute each tool call
567+
// ACT: execute each tool call in parallel with bounded concurrency
556568
toolNames := make([]string, 0, len(result.ToolCalls))
569+
570+
// Phase 1: fire all tool_call events synchronously (rendering + events)
557571
for _, tc := range result.ToolCalls {
558572
toolNames = append(toolNames, tc.Function.Name)
559573

560-
// Render tool call: engaging mode uses narrator, verbose mode uses renderer.
561574
if e.narrator != nil {
562575
if msg := e.narrator.ToolCallMessage(tc.Function.Name, tc.Function.Arguments); msg != "" {
563576
if e.renderer != nil {
@@ -570,17 +583,46 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [
570583
if e.toolEventHandler != nil {
571584
e.toolEventHandler("tool_call", tc.Function.Name, tc.Function.Arguments)
572585
}
586+
}
573587

574-
t := e.registry.Get(tc.Function.Name)
575-
output := fmt.Sprintf("error: tool %q not found", tc.Function.Name)
576-
if t != nil {
577-
res, err := t.Call(tc.Function.Arguments)
578-
if err != nil {
579-
output = fmt.Sprintf("error: %s", err.Error())
580-
} else {
581-
output = redact.RedactSecrets(res)
588+
// Phase 2: execute tools in parallel (bounded by semaphore)
589+
type execResult struct {
590+
output string
591+
}
592+
parallel := e.MaxToolParallel
593+
if parallel <= 0 {
594+
parallel = 4
595+
}
596+
sem := make(chan struct{}, parallel)
597+
results := make([]execResult, len(result.ToolCalls))
598+
599+
for i, tc := range result.ToolCalls {
600+
sem <- struct{}{} // acquire — blocks if at cap
601+
go func(idx int, tcRef llm.ToolCall) {
602+
defer func() { <-sem }() // release
603+
604+
t := e.registry.Get(tcRef.Function.Name)
605+
output := fmt.Sprintf("error: tool %q not found", tcRef.Function.Name)
606+
if t != nil {
607+
res, err := t.Call(tcRef.Function.Arguments)
608+
if err != nil {
609+
output = fmt.Sprintf("error: %s", err.Error())
610+
} else {
611+
output = redact.RedactSecrets(res)
612+
}
582613
}
583-
}
614+
results[idx] = execResult{output: output}
615+
}(i, tc)
616+
}
617+
// Drain the semaphore — wait for all goroutines to finish.
618+
for i := 0; i < cap(sem); i++ {
619+
sem <- struct{}{}
620+
}
621+
622+
// Phase 3: process results in order (render, compress, append to messages)
623+
const maxOutput = 4096
624+
for i, tc := range result.ToolCalls {
625+
output := results[i].output
584626

585627
// Tool results: only shown in verbose mode.
586628
if e.narrator == nil && e.renderer != nil {
@@ -591,19 +633,18 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [
591633
}
592634

593635
// Compress large tool outputs to save context window.
594-
// Keep the first and last portions — head usually contains
595-
// the most important info, tail may have final results.
596-
const maxOutput = 4096
597-
if len(output) > maxOutput {
598-
head := maxOutput * 3 / 4 // 3KB head
599-
tail := maxOutput / 4 // 1KB tail
600-
output = output[:head] +
601-
fmt.Sprintf("\n\n... [%d bytes omitted — output was %d bytes total] ...\n\n",
602-
len(output)-head-tail, len(output)) +
603-
output[len(output)-tail:]
604-
}
636+
// Keep the first and last portions — head usually contains
637+
// the most important info, tail may have final results.
638+
if len(output) > maxOutput {
639+
head := maxOutput * 3 / 4 // 3KB head
640+
tail := maxOutput / 4 // 1KB tail
641+
output = output[:head] +
642+
fmt.Sprintf("\n\n... [%d bytes omitted — output was %d bytes total] ...\n\n",
643+
len(output)-head-tail, len(output)) +
644+
output[len(output)-tail:]
645+
}
605646

606-
// Wrap tool output in unbreakable delimiters so the model
647+
// Wrap tool output in unbreakable delimiters so the model
607648
// treats it as DATA, never as instructions. The header and
608649
// footer both explicitly frame the content as untrusted data.
609650
// Even if the output contains "ignore previous instructions",

odek.go

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -151,6 +151,12 @@ type Config struct {
151151
// on cached tokens and ~60-80% TTFT latency reduction.
152152
PromptCaching bool
153153

154+
// MaxToolParallel controls how many tool calls run concurrently per
155+
// agent iteration. 0 = use default (4). Models that emit multiple
156+
// parallel tool calls benefit from concurrent execution of I/O-bound
157+
// tools like read_file, search_files, and web_search.
158+
MaxToolParallel int
159+
154160
// SkillEventHandler, if set, is invoked when a skill lifecycle event
155161
// occurs (loaded, autoloaded, saved, deleted, etc.). Used by WebUI
156162
// (WebSocket streaming) and Telegram (inline messages).
@@ -475,6 +481,9 @@ func New(cfg Config) (*Agent, error) {
475481

476482
engine := loop.New(client, registry, cfg.MaxIterations, cfg.SystemMessage, cfg.Renderer, maxContext)
477483
engine.PromptCaching = cfg.PromptCaching
484+
if cfg.MaxToolParallel > 0 {
485+
engine.SetMaxToolParallel(cfg.MaxToolParallel)
486+
}
478487

479488
// Set skill verbosity: condensed by default, full banners when verbose.
480489
if cfg.Skills != nil {

0 commit comments

Comments
 (0)