diff --git a/AGENT_LOOP_IMPROVEMENTS.md b/AGENT_LOOP_IMPROVEMENTS.md index 6d135055c..0f48a7e9f 100644 --- a/AGENT_LOOP_IMPROVEMENTS.md +++ b/AGENT_LOOP_IMPROVEMENTS.md @@ -21,7 +21,7 @@ This document outlines a series of tasks to improve the main loop of the agentic ## Phase 3: Performance & Latency * [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 an asynchronous worker. Instead of blocking the main thread, the worker outputs its execution trace and compressed context to a structured `/logs/{session}/{subagent}/` directory to provide a persistent audit trail while keeping the loop responsive. +* [x] **Background Summarization:** Offload `maybeSummarize` and context compression to an asynchronous worker. Instead of blocking the main thread, the worker outputs its execution trace and compressed context to a structured `/logs/{session}/{subagent}/` directory to provide a persistent audit trail while keeping the loop responsive. * [ ] **Streaming Responses:** Implement streaming LLM token generation directly to the `bus.PublishOutbound` instead of waiting for full generation. ## Phase 4: Features diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 89a49155c..d3932fc88 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 + summaryJobs chan summaryJob wg sync.WaitGroup fallback *providers.FallbackChain channelManager *channels.Manager @@ -37,6 +38,11 @@ type AgentLoop struct { } // processOptions configures how a message is processed +type summaryJob struct { + agent *AgentInstance + sessionKey string +} + type processOptions struct { SessionKey string // Session identifier for history/context Channel string // Target channel for tool execution diff --git a/pkg/agent/loop_init.go b/pkg/agent/loop_init.go index 8cda5a790..5df39adb2 100644 --- a/pkg/agent/loop_init.go +++ b/pkg/agent/loop_init.go @@ -52,6 +52,7 @@ func NewAgentLoop( registry: registry, state: stateManager, summarizing: sync.Map{}, + summaryJobs: make(chan summaryJob, 100), fallback: fallbackChain, cmdRegistry: commands.NewRegistry(commands.BuiltinDefinitions()), } @@ -195,6 +196,25 @@ func (al *AgentLoop) Run(ctx context.Context) error { return err } + // Start background summarization worker + al.wg.Add(1) + go func() { + defer al.wg.Done() + for { + select { + case <-ctx.Done(): + return + case job, ok := <-al.summaryJobs: + if !ok { + return + } + logger.Debug("Memory threshold reached. Optimizing conversation history...") + al.summarizeSession(job.agent, job.sessionKey) + al.summarizing.Delete(job.agent.ID + ":" + job.sessionKey) + } + } + }() + for al.running.Load() { select { case <-ctx.Done(): diff --git a/pkg/agent/loop_summary.go b/pkg/agent/loop_summary.go index 42260e45e..3ca6bac5f 100644 --- a/pkg/agent/loop_summary.go +++ b/pkg/agent/loop_summary.go @@ -8,7 +8,10 @@ package agent import ( "context" + "encoding/json" "fmt" + "os" + "path/filepath" "strings" "time" "unicode/utf8" @@ -26,11 +29,15 @@ func (al *AgentLoop) maybeSummarize(agent *AgentInstance, sessionKey, channel, c if len(newHistory) > agent.SummarizeMessageThreshold || tokenEstimate > threshold { summarizeKey := agent.ID + ":" + sessionKey if _, loading := al.summarizing.LoadOrStore(summarizeKey, true); !loading { - go func() { - defer al.summarizing.Delete(summarizeKey) - logger.Debug("Memory threshold reached. Optimizing conversation history...") - al.summarizeSession(agent, sessionKey) - }() + // Send to background worker queue + select { + case al.summaryJobs <- summaryJob{agent: agent, sessionKey: sessionKey}: + // job accepted + default: + // Worker queue is full, delete from map so it can be retried later + logger.WarnCF("agent", "Summarization worker queue is full, skipping summarization", map[string]any{"session_key": sessionKey}) + al.summarizing.Delete(summarizeKey) + } } } } @@ -43,6 +50,23 @@ func (al *AgentLoop) forceCompression(agent *AgentInstance, sessionKey string) { return } + // Create logging directory: /logs/{session_id}/summarizer/ + logDir := filepath.Join("logs", sessionKey, "summarizer") + if err := os.MkdirAll(logDir, 0755); err != nil { + logger.WarnCF("agent", "Failed to create summarizer log directory", map[string]any{"error": err.Error()}) + } + + // Dump input context + if b, err := json.MarshalIndent(history, "", " "); err == nil { + _ = os.WriteFile(filepath.Join(logDir, "input_context.json"), b, 0644) + } + + traceFile, _ := os.OpenFile(filepath.Join(logDir, "trace.log"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + if traceFile != nil { + defer traceFile.Close() + fmt.Fprintf(traceFile, "[%s] Starting summarization for session: %s\n", time.Now().Format(time.RFC3339), sessionKey) + } + // Keep system prompt (usually [0]) and the very last message (user's trigger) // We want to drop the oldest half of the *conversation* // Assuming [0] is system, [1:] is conversation @@ -101,6 +125,23 @@ func (al *AgentLoop) summarizeSession(agent *AgentInstance, sessionKey string) { return } + // Create logging directory: /logs/{session_id}/summarizer/ + logDir := filepath.Join("logs", sessionKey, "summarizer") + if err := os.MkdirAll(logDir, 0755); err != nil { + logger.WarnCF("agent", "Failed to create summarizer log directory", map[string]any{"error": err.Error()}) + } + + // Dump input context + if b, err := json.MarshalIndent(history, "", " "); err == nil { + _ = os.WriteFile(filepath.Join(logDir, "input_context.json"), b, 0644) + } + + traceFile, _ := os.OpenFile(filepath.Join(logDir, "trace.log"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + if traceFile != nil { + defer traceFile.Close() + fmt.Fprintf(traceFile, "[%s] Starting summarization for session: %s\n", time.Now().Format(time.RFC3339), sessionKey) + } + toSummarize := history[:len(history)-4] // Oversized Message Guard @@ -115,12 +156,18 @@ func (al *AgentLoop) summarizeSession(agent *AgentInstance, sessionKey string) { msgTokens := len(m.Content) / 2 if msgTokens > maxMessageTokens { omitted = true + if traceFile != nil { + fmt.Fprintf(traceFile, "[%s] Omitting oversized message from %s (length: %d)\n", time.Now().Format(time.RFC3339), m.Role, len(m.Content)) + } continue } validMessages = append(validMessages, m) } if len(validMessages) == 0 { + if traceFile != nil { + fmt.Fprintf(traceFile, "[%s] No valid messages to summarize.\n", time.Now().Format(time.RFC3339)) + } return } @@ -150,13 +197,23 @@ func (al *AgentLoop) summarizeSession(agent *AgentInstance, sessionKey string) { s2, ) + if traceFile != nil { + fmt.Fprintf(traceFile, "[%s] Merging multi-part summaries...\n", time.Now().Format(time.RFC3339)) + } + resp, err := al.retryLLMCall(ctx, agent, mergePrompt, llmMaxRetries) if err == nil && resp.Content != "" { finalSummary = resp.Content } else { + if traceFile != nil { + fmt.Fprintf(traceFile, "[%s] LLM merge failed: %v. Falling back to concatenation.\n", time.Now().Format(time.RFC3339), err) + } finalSummary = s1 + " " + s2 } } else { + if traceFile != nil { + fmt.Fprintf(traceFile, "[%s] Summarizing single batch...\n", time.Now().Format(time.RFC3339)) + } finalSummary, _ = al.summarizeBatch(ctx, agent, validMessages, summary) } @@ -165,9 +222,18 @@ func (al *AgentLoop) summarizeSession(agent *AgentInstance, sessionKey string) { } if finalSummary != "" { + if traceFile != nil { + fmt.Fprintf(traceFile, "[%s] Summarization complete. Saving session.\n", time.Now().Format(time.RFC3339)) + } + _ = os.WriteFile(filepath.Join(logDir, "result_summary.md"), []byte(finalSummary), 0644) + agent.Sessions.SetSummary(sessionKey, finalSummary) agent.Sessions.TruncateHistory(sessionKey, 4) agent.Sessions.Save(sessionKey) + } else { + if traceFile != nil { + fmt.Fprintf(traceFile, "[%s] Final summary is empty, skipping save.\n", time.Now().Format(time.RFC3339)) + } } }