diff --git a/AGENT_LOOP_IMPROVEMENTS.md b/AGENT_LOOP_IMPROVEMENTS.md index 856422849..3ca571a7d 100644 --- a/AGENT_LOOP_IMPROVEMENTS.md +++ b/AGENT_LOOP_IMPROVEMENTS.md @@ -13,7 +13,7 @@ This document outlines a series of tasks to improve the main loop of the agentic - [ ] **Granular Error Classification:** Update `LLMProvider` interfaces to return structured, typed errors (e.g., `providers.ErrContextLengthExceeded`) instead of relying on fragile string matching. ## Phase 3: Performance & Latency -- [ ] **Concurrent Message Processing:** Evaluate introducing a worker pool or goroutines to process independent user requests concurrently without blocking the entire agent instance. +- [x] **Concurrent Message Processing:** Evaluate introducing a worker pool or goroutines to process independent user requests concurrently without blocking the entire agent instance. - [ ] **Background Summarization:** Offload `maybeSummarize` and context compression to a background worker to unblock the main thread and respond to the user faster. - [ ] **Streaming Responses:** Implement streaming LLM token generation directly to the `bus.PublishOutbound` instead of waiting for full generation. diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index fe0bbe77e..89a49155c 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -27,6 +27,7 @@ type AgentLoop struct { state *state.Manager running atomic.Bool summarizing sync.Map + wg sync.WaitGroup fallback *providers.FallbackChain channelManager *channels.Manager mediaStore media.MediaStore diff --git a/pkg/agent/loop_init.go b/pkg/agent/loop_init.go index 5cdce0c6d..8cda5a790 100644 --- a/pkg/agent/loop_init.go +++ b/pkg/agent/loop_init.go @@ -198,6 +198,7 @@ func (al *AgentLoop) Run(ctx context.Context) error { for al.running.Load() { select { case <-ctx.Done(): + al.wg.Wait() return nil default: msg, ok := al.bus.ConsumeInbound(ctx) @@ -205,8 +206,10 @@ func (al *AgentLoop) Run(ctx context.Context) error { continue } + al.wg.Add(1) // Process message - func() { + go func(msg bus.InboundMessage) { + defer al.wg.Done() // TODO: Re-enable media cleanup after inbound media is properly consumed by the agent. // Currently disabled because files are deleted before the LLM can access their content. // defer func() { @@ -259,10 +262,11 @@ func (al *AgentLoop) Run(ctx context.Context) error { ) } } - }() + }(msg) } } + al.wg.Wait() return nil }