feat(agent): add fewshot, summary, task_completion; update run, context, loop
- fewshot: few-shot prompting support - summary: summary generation - task_completion: task completion logic - agent_run, context, loop, offloading_tool_runtime, summarizer updates
This commit is contained in:
ZanzyTHEbar
parent
7ee300fcb2
commit
45b525b832
9 changed files with 1142 additions and 127 deletions
|
|
@ -222,7 +222,18 @@ func (al *AgentLoop) loadSessionState(ctx context.Context, opts processOptions)
|
|||
summary = al.sessions.GetSummary(opts.SessionKey)
|
||||
}
|
||||
|
||||
return al.applyDAGCompression(ctx, opts.SessionKey, history), summary
|
||||
// Zero-cost continuity (LCM ADR-002): skip DAG compression when context
|
||||
// usage is below the soft compaction threshold. This eliminates overhead
|
||||
// for ~80% of interactions where context is not under pressure.
|
||||
softPct, _ := al.compactionThresholds()
|
||||
softThreshold := al.contextWindow * softPct / 100
|
||||
tokenEstimate := al.estimateTokens(history)
|
||||
if tokenEstimate <= softThreshold {
|
||||
al.contextBuilder.SetContextTreeBlock("")
|
||||
return history, summary
|
||||
}
|
||||
|
||||
return al.applyContextTreeSelection(ctx, opts.SessionKey, opts.UserMessage, history), summary
|
||||
}
|
||||
|
||||
func (al *AgentLoop) buildPromptMessages(opts processOptions, history []messages.Message, summary string) []messages.Message {
|
||||
|
|
|
|||
|
|
@ -29,7 +29,7 @@ type ContextBuilder struct {
|
|||
observationBlock string // Pre-rendered observation block for prompt injection
|
||||
focusBlock string // Pre-rendered active focus block
|
||||
knowledgeBlock string // Pre-rendered knowledge block from Focus completions
|
||||
dagBlock string // Pre-rendered DAG compressed history
|
||||
contextTreeBlock string // Pre-rendered Context-Tree selected history
|
||||
contextWindow int // Max tokens for context window (0 = no limit)
|
||||
|
||||
cacheMu sync.Mutex
|
||||
|
|
@ -99,9 +99,9 @@ func (cb *ContextBuilder) SetKnowledgeBlock(block string) {
|
|||
cb.knowledgeBlock = block
|
||||
}
|
||||
|
||||
// SetDAGBlock sets the pre-rendered DAG compressed history for prompt injection.
|
||||
func (cb *ContextBuilder) SetDAGBlock(block string) {
|
||||
cb.dagBlock = block
|
||||
// SetContextTreeBlock sets the pre-rendered query-selected context history block for prompt injection.
|
||||
func (cb *ContextBuilder) SetContextTreeBlock(block string) {
|
||||
cb.contextTreeBlock = block
|
||||
}
|
||||
|
||||
// SetContextWindow configures the token budget for the system prompt.
|
||||
|
|
@ -226,9 +226,9 @@ Do NOT assume skill content — always load before applying.
|
|||
sections = append(sections, contextSection{"knowledge", cb.knowledgeBlock, 6})
|
||||
}
|
||||
|
||||
// P7: DAG compressed history (lowest priority — can be reconstructed)
|
||||
if cb.dagBlock != "" {
|
||||
sections = append(sections, contextSection{"dag", "# Conversation History (Compressed)\n\n" + cb.dagBlock, 7})
|
||||
// P7: Context-Tree selected history (query-adaptive)
|
||||
if cb.contextTreeBlock != "" {
|
||||
sections = append(sections, contextSection{"context_tree", "# Conversation Context (Query-Selected)\n\n" + cb.contextTreeBlock, 7})
|
||||
}
|
||||
|
||||
// Proportional budget enforcement: each section gets a share of the
|
||||
|
|
@ -287,7 +287,7 @@ var priorityWeight = [8]float64{
|
|||
4: 0.10, // observations
|
||||
5: 0.08, // focus
|
||||
6: 0.06, // knowledge
|
||||
7: 0.06, // DAG
|
||||
7: 0.06, // Context Tree
|
||||
}
|
||||
|
||||
// applyProportionalBudget distributes tokens among sections using priority
|
||||
|
|
|
|||
235
pkg/agent/fewshot.go
Normal file
235
pkg/agent/fewshot.go
Normal file
|
|
@ -0,0 +1,235 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/logger"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/memory/dag"
|
||||
)
|
||||
|
||||
// FewShotProvider provides few-shot examples for prompt injection.
|
||||
type FewShotProvider struct {
|
||||
store dag.AuditStore
|
||||
formatter *dag.FewShotFormatter
|
||||
maxExamples int
|
||||
}
|
||||
|
||||
// NewFewShotProvider creates a new few-shot provider.
|
||||
func NewFewShotProvider(store dag.AuditStore, maxExamples int) *FewShotProvider {
|
||||
return &FewShotProvider{
|
||||
store: store,
|
||||
formatter: dag.NewFewShotFormatter(maxExamples, 5),
|
||||
maxExamples: maxExamples,
|
||||
}
|
||||
}
|
||||
|
||||
// GetExamples retrieves and formats few-shot examples for a given intent.
|
||||
func (fsp *FewShotProvider) GetExamples(ctx context.Context, agentID, intent string) string {
|
||||
chains, err := fsp.store.GetTopChains(ctx, agentID, intent, fsp.maxExamples)
|
||||
if err != nil {
|
||||
logger.DebugCF("agent", "Failed to get few-shot chains", map[string]interface{}{"error": err})
|
||||
return ""
|
||||
}
|
||||
|
||||
if len(chains) == 0 {
|
||||
return ""
|
||||
}
|
||||
|
||||
return fsp.formatter.FormatForPrompt(chains, intent)
|
||||
}
|
||||
|
||||
// GetExamplesForQuery retrieves examples matching a natural language query.
|
||||
func (fsp *FewShotProvider) GetExamplesForQuery(ctx context.Context, agentID, query string) string {
|
||||
// Simple intent detection from query
|
||||
intent := detectIntent(query)
|
||||
return fsp.GetExamples(ctx, agentID, intent)
|
||||
}
|
||||
|
||||
// detectIntent performs simple keyword-based intent detection.
|
||||
func detectIntent(query string) string {
|
||||
queryLower := strings.ToLower(query)
|
||||
|
||||
// Search-related
|
||||
if strings.Contains(queryLower, "find") ||
|
||||
strings.Contains(queryLower, "search") ||
|
||||
strings.Contains(queryLower, "lookup") {
|
||||
return "research"
|
||||
}
|
||||
|
||||
// File operations
|
||||
if strings.Contains(queryLower, "file") ||
|
||||
strings.Contains(queryLower, "read") ||
|
||||
strings.Contains(queryLower, "write") ||
|
||||
strings.Contains(queryLower, "edit") {
|
||||
return "file_research"
|
||||
}
|
||||
|
||||
// Code/development
|
||||
if strings.Contains(queryLower, "code") ||
|
||||
strings.Contains(queryLower, "function") ||
|
||||
strings.Contains(queryLower, "test") ||
|
||||
strings.Contains(queryLower, "bug") ||
|
||||
strings.Contains(queryLower, "fix") {
|
||||
return "development"
|
||||
}
|
||||
|
||||
// Git operations
|
||||
if strings.Contains(queryLower, "git") ||
|
||||
strings.Contains(queryLower, "commit") ||
|
||||
strings.Contains(queryLower, "branch") ||
|
||||
strings.Contains(queryLower, "push") ||
|
||||
strings.Contains(queryLower, "pull") {
|
||||
return "git_workflow"
|
||||
}
|
||||
|
||||
// Execution
|
||||
if strings.Contains(queryLower, "run") ||
|
||||
strings.Contains(queryLower, "execute") ||
|
||||
strings.Contains(queryLower, "command") ||
|
||||
strings.Contains(queryLower, "shell") {
|
||||
return "execution"
|
||||
}
|
||||
|
||||
return ""
|
||||
}
|
||||
|
||||
// BuildFewShotBlock creates a formatted few-shot block for prompt injection.
|
||||
func BuildFewShotBlock(examples string) string {
|
||||
if examples == "" {
|
||||
return ""
|
||||
}
|
||||
|
||||
var b strings.Builder
|
||||
b.WriteString("## Example Task Completion Patterns\n\n")
|
||||
b.WriteString("When approaching similar tasks, consider these successful patterns:\n\n")
|
||||
b.WriteString(examples)
|
||||
b.WriteString("\nUse these patterns as guidance when completing your current task.\n")
|
||||
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// FewShotInjector handles injection of few-shot examples into prompts.
|
||||
type FewShotInjector struct {
|
||||
provider *FewShotProvider
|
||||
enabled bool
|
||||
}
|
||||
|
||||
// NewFewShotInjector creates a few-shot injector.
|
||||
func NewFewShotInjector(store dag.AuditStore, enabled bool, maxExamples int) *FewShotInjector {
|
||||
return &FewShotInjector{
|
||||
provider: NewFewShotProvider(store, maxExamples),
|
||||
enabled: enabled,
|
||||
}
|
||||
}
|
||||
|
||||
// Inject adds few-shot examples to a system prompt based on the user's query.
|
||||
func (fsi *FewShotInjector) Inject(ctx context.Context, agentID, query, systemPrompt string) string {
|
||||
if !fsi.enabled {
|
||||
return systemPrompt
|
||||
}
|
||||
examples := fsi.provider.GetExamplesForQuery(ctx, agentID, query)
|
||||
return fsi.injectBlock(systemPrompt, examples)
|
||||
}
|
||||
|
||||
func (fsi *FewShotInjector) injectBlock(systemPrompt, examples string) string {
|
||||
if examples == "" {
|
||||
return systemPrompt
|
||||
}
|
||||
block := BuildFewShotBlock(examples)
|
||||
if idx := strings.LastIndex(systemPrompt, "## Current Task"); idx > 0 {
|
||||
return systemPrompt[:idx] + block + "\n\n" + systemPrompt[idx:]
|
||||
}
|
||||
return systemPrompt + "\n\n" + block
|
||||
}
|
||||
|
||||
// InjectByIntent adds few-shot examples for a specific intent category.
|
||||
func (fsi *FewShotInjector) InjectByIntent(ctx context.Context, agentID, intent, systemPrompt string) string {
|
||||
if !fsi.enabled {
|
||||
return systemPrompt
|
||||
}
|
||||
examples := fsi.provider.GetExamples(ctx, agentID, intent)
|
||||
return fsi.injectBlock(systemPrompt, examples)
|
||||
}
|
||||
|
||||
// InjectForTools adds examples showing successful usage of specific tools.
|
||||
func (fsi *FewShotInjector) InjectForTools(ctx context.Context, agentID string, toolNames []string, systemPrompt string) string {
|
||||
if !fsi.enabled || len(toolNames) == 0 {
|
||||
return systemPrompt
|
||||
}
|
||||
examples := fsi.provider.GetExamples(ctx, agentID, "development")
|
||||
return fsi.injectToolBlock(systemPrompt, toolNames, examples)
|
||||
}
|
||||
|
||||
func (fsi *FewShotInjector) injectToolBlock(systemPrompt string, toolNames []string, examples string) string {
|
||||
if examples == "" {
|
||||
return systemPrompt
|
||||
}
|
||||
var b strings.Builder
|
||||
b.WriteString("## Tool Usage Patterns\n\n")
|
||||
b.WriteString(fmt.Sprintf("When using %s, follow these patterns:\n\n", strings.Join(toolNames, ", ")))
|
||||
b.WriteString(examples)
|
||||
return systemPrompt + "\n\n" + b.String()
|
||||
}
|
||||
|
||||
// ChainSummary provides a summary of available action chains.
|
||||
type ChainSummary struct {
|
||||
TotalChains int
|
||||
ByIntent map[string]int
|
||||
ByCategory map[string]int
|
||||
AverageScore float64
|
||||
TopIntents []string
|
||||
}
|
||||
|
||||
// GetSummary returns a summary of available chains for an agent.
|
||||
func (fsp *FewShotProvider) GetSummary(ctx context.Context, agentID string) (*ChainSummary, error) {
|
||||
// Get chains across all intents
|
||||
chains, err := fsp.store.GetTopChains(ctx, agentID, "", 100)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
summary := &ChainSummary{
|
||||
TotalChains: len(chains),
|
||||
ByIntent: make(map[string]int),
|
||||
ByCategory: make(map[string]int),
|
||||
}
|
||||
|
||||
var totalScore float64
|
||||
intentScores := make(map[string]float64)
|
||||
|
||||
for _, chain := range chains {
|
||||
summary.ByIntent[chain.Intent]++
|
||||
summary.ByCategory[chain.Category]++
|
||||
totalScore += chain.Score
|
||||
intentScores[chain.Intent] += chain.Score
|
||||
}
|
||||
|
||||
if len(chains) > 0 {
|
||||
summary.AverageScore = totalScore / float64(len(chains))
|
||||
}
|
||||
|
||||
// Find top intents by count
|
||||
type intentCount struct {
|
||||
intent string
|
||||
count int
|
||||
score float64
|
||||
}
|
||||
|
||||
var intents []intentCount
|
||||
for intent, count := range summary.ByIntent {
|
||||
intents = append(intents, intentCount{intent, count, intentScores[intent]})
|
||||
}
|
||||
|
||||
sort.Slice(intents, func(i, j int) bool {
|
||||
return intents[i].count > intents[j].count
|
||||
})
|
||||
|
||||
for _, ic := range intents {
|
||||
summary.TopIntents = append(summary.TopIntents, ic.intent)
|
||||
}
|
||||
|
||||
return summary, nil
|
||||
}
|
||||
|
|
@ -402,31 +402,25 @@ func TestIntegration_Streaming_TextDeltas(t *testing.T) {
|
|||
ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
// Collect stream deltas in background
|
||||
var deltas []string
|
||||
var deltaDone = make(chan struct{})
|
||||
go func() {
|
||||
defer close(deltaDone)
|
||||
for {
|
||||
msg, ok := msgBus.SubscribeOutbound(ctx)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if msg.StreamDelta {
|
||||
deltas = append(deltas, msg.Content)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// Process with streaming
|
||||
response, err := al.ProcessDirectStreaming(ctx, "Stream me", "stream-session", "test", "chat-1")
|
||||
if err != nil {
|
||||
t.Fatalf("ProcessDirectStreaming failed: %v", err)
|
||||
}
|
||||
|
||||
// Cancel to stop delta collector
|
||||
cancel()
|
||||
<-deltaDone
|
||||
// Drain outbound stream deltas after processing completes.
|
||||
var deltas []string
|
||||
for {
|
||||
readCtx, readCancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
|
||||
msg, ok := msgBus.SubscribeOutbound(readCtx)
|
||||
readCancel()
|
||||
if !ok {
|
||||
break
|
||||
}
|
||||
if msg.StreamDelta {
|
||||
deltas = append(deltas, msg.Content)
|
||||
}
|
||||
}
|
||||
|
||||
// Verify complete response
|
||||
if response != "Hello from streaming agent response" {
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ import (
|
|||
"github.com/ZanzyTHEbar/dragonscale/pkg/channels"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/config"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/constants"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/cortex"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/ids"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/logger"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/memory"
|
||||
|
|
@ -63,7 +64,7 @@ type AgentLoop struct {
|
|||
running atomic.Bool // Owner: loop.go — lifecycle gate controlled by Run/Stop only
|
||||
summarizing sync.Map // Owner: summarizer.go — intended for async summarization lockout, currently gated by TODO path
|
||||
summarizeFailures sync.Map // Owner: summarizer.go — write/read in forceCompression + summarizeSession error paths
|
||||
dagCache sync.Map // Owner: summarizer.go — sessionKey → dagCacheEntry, skips recompression when compressible count unchanged
|
||||
contextTreeCache sync.Map // Owner: summarizer.go — sessionKey → contextTreeCacheEntry keyed by query and history size
|
||||
auditChan chan *memory.AuditEntry // Buffered channel for async audit logging; drained by background worker
|
||||
auditDone chan struct{} // Closed when audit worker exits
|
||||
focusDirty sync.Map // sessionKey → struct{}: set by focus tool callbacks, cleared after context reload
|
||||
|
|
@ -73,6 +74,7 @@ type AgentLoop struct {
|
|||
commandRegistry []SlashCommand
|
||||
outputOverride atomic.Value // Owner: command_handler.go — CLI output redirection target for internal messages
|
||||
toolResultSearch fantasy.AgentTool
|
||||
cortex *cortex.Cortex
|
||||
}
|
||||
|
||||
type outputTarget struct {
|
||||
|
|
@ -389,11 +391,59 @@ func NewAgentLoop(ctx context.Context, cfg *config.Config, msgBus *bus.MessageBu
|
|||
al.SetupSecureBus(secretStore, securebus.DefaultBusConfig())
|
||||
subagentManager.SetRunLoop(MakeUnifiedRunLoopFunc(al))
|
||||
|
||||
// Initialize Cortex autonomous scheduler with foundation tasks.
|
||||
// DecayStore and BackfillStore are satisfied by LibSQLDelegate via duck typing.
|
||||
var decayStore cortex.DecayStore
|
||||
if ds, ok := al.memDelegate.(cortex.DecayStore); ok {
|
||||
decayStore = ds
|
||||
}
|
||||
var backfillStore cortex.BackfillStore
|
||||
if bs, ok := al.memDelegate.(cortex.BackfillStore); ok {
|
||||
backfillStore = bs
|
||||
}
|
||||
var embedFn func(ctx context.Context, text string) ([]float32, error)
|
||||
if al.memoryStore != nil && al.memoryStore.Embedder() != nil {
|
||||
embedder := al.memoryStore.Embedder()
|
||||
embedFn = func(ctx context.Context, text string) ([]float32, error) {
|
||||
return embedder.Embed(ctx, text)
|
||||
}
|
||||
}
|
||||
// ConsolidationStore for memory graph maintenance
|
||||
var consolidationStore cortex.ConsolidationStore
|
||||
if cs, ok := al.memDelegate.(cortex.ConsolidationStore); ok {
|
||||
consolidationStore = cs
|
||||
}
|
||||
// PruneStore for permanent deletion of quarantined items
|
||||
var pruneStore cortex.PruneStore
|
||||
if ps, ok := al.memDelegate.(cortex.PruneStore); ok {
|
||||
pruneStore = ps
|
||||
}
|
||||
// RLStore for reinforcement learning weight updates
|
||||
var rlStore cortex.RLStore
|
||||
if rs, ok := al.memDelegate.(cortex.RLStore); ok {
|
||||
rlStore = rs
|
||||
}
|
||||
// AuditAnalysisStore for audit log pattern detection
|
||||
var auditStore cortex.AuditAnalysisStore
|
||||
if aus, ok := al.memDelegate.(cortex.AuditAnalysisStore); ok {
|
||||
auditStore = aus
|
||||
}
|
||||
cortexTasks := []cortex.Task{
|
||||
cortex.NewDecayTask(cortex.DefaultDecayConfig(), decayStore),
|
||||
cortex.NewBackfillTask(cortex.DefaultBackfillConfig(), backfillStore, embedFn),
|
||||
cortex.NewConsolidationTask(cortex.DefaultConsolidationConfig(), consolidationStore),
|
||||
cortex.NewPruneTask(cortex.DefaultPruneConfig(), pruneStore),
|
||||
cortex.NewRLTask(rlStore, pkg.NAME),
|
||||
cortex.NewAuditAnalysisTask(auditStore),
|
||||
}
|
||||
al.cortex = cortex.New(cortexTasks, 60*time.Second)
|
||||
|
||||
return al, nil
|
||||
}
|
||||
|
||||
func (al *AgentLoop) Run(ctx context.Context) error {
|
||||
al.running.Store(true)
|
||||
go al.cortex.Start(ctx)
|
||||
|
||||
for al.running.Load() {
|
||||
select {
|
||||
|
|
|
|||
|
|
@ -116,9 +116,11 @@ func (r OffloadingToolRuntime) Execute(ctx context.Context, tools []fantasy.Agen
|
|||
}
|
||||
}
|
||||
|
||||
preview = truncateRunes(payloadText, threshold)
|
||||
preview = strings.TrimSpace(preview) + "\n\n" +
|
||||
// Type-aware summarization instead of blind truncation
|
||||
summary := SummarizeContent(payloadText, threshold)
|
||||
preview = summary.Summary + "\n\n" +
|
||||
"[TRUNCATED]\n" +
|
||||
"[Type: " + string(summary.ContentType) + "]\n" +
|
||||
"- run_id: " + r.RunID.String() + "\n" +
|
||||
"- tool_call_id: " + tc.ToolCallID + "\n" +
|
||||
"- chunk_count: " + strconv.FormatInt(chunkCount, 10) + "\n" +
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import (
|
|||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
|
|
@ -12,6 +13,7 @@ import (
|
|||
"github.com/ZanzyTHEbar/dragonscale/pkg/bus"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/config"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/constants"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/contexttree"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/ids"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/logger"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/memory"
|
||||
|
|
@ -19,18 +21,52 @@ import (
|
|||
"github.com/ZanzyTHEbar/dragonscale/pkg/memory/observation"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/messages"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/tools"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/utils"
|
||||
)
|
||||
|
||||
// maybeSummarize only triggers emergency compression when hard limits are exceeded.
|
||||
// Normal background compaction is intentionally disabled for the unified kernel.
|
||||
// maybeSummarize implements LCM ADR-002 dual-threshold compaction.
|
||||
// Below soft threshold: no-op (zero cost).
|
||||
// Between soft and hard: async background compaction.
|
||||
// Above hard: synchronous blocking compaction with 3-level escalation.
|
||||
func (al *AgentLoop) maybeSummarize(ctx context.Context, sessionKey, channel, chatID string) {
|
||||
newHistory := al.sessions.GetHistory(sessionKey)
|
||||
tokenEstimate := al.estimateTokens(newHistory)
|
||||
criticalThreshold := al.contextWindow * 95 / 100
|
||||
|
||||
if tokenEstimate > criticalThreshold {
|
||||
al.forceCompression(ctx, sessionKey, channel, chatID)
|
||||
softPct, hardPct := al.compactionThresholds()
|
||||
softThreshold := al.contextWindow * softPct / 100
|
||||
hardThreshold := al.contextWindow * hardPct / 100
|
||||
|
||||
if tokenEstimate <= softThreshold {
|
||||
return
|
||||
}
|
||||
|
||||
if tokenEstimate > hardThreshold {
|
||||
al.forceCompression(ctx, sessionKey, channel, chatID)
|
||||
return
|
||||
}
|
||||
|
||||
go func() {
|
||||
al.summarizeSession(context.WithoutCancel(ctx), sessionKey)
|
||||
}()
|
||||
}
|
||||
|
||||
// compactionThresholds returns the soft and hard compaction percentages from config.
|
||||
func (al *AgentLoop) compactionThresholds() (softPct, hardPct int) {
|
||||
softPct = 70
|
||||
hardPct = 90
|
||||
if al.cfg != nil {
|
||||
c := al.cfg.Agents.Defaults.Compaction
|
||||
if c.SoftThresholdPct > 0 && c.SoftThresholdPct < 100 {
|
||||
softPct = c.SoftThresholdPct
|
||||
}
|
||||
if c.HardThresholdPct > 0 && c.HardThresholdPct <= 100 {
|
||||
hardPct = c.HardThresholdPct
|
||||
}
|
||||
}
|
||||
if hardPct <= softPct {
|
||||
hardPct = softPct + 10
|
||||
}
|
||||
return softPct, hardPct
|
||||
}
|
||||
|
||||
// EmergencyProvenance captures provenance metadata for postmortem when
|
||||
|
|
@ -118,7 +154,7 @@ func (al *AgentLoop) forceCompression(ctx context.Context, sessionKey, channel,
|
|||
HistoryMsgCount: len(history),
|
||||
})
|
||||
|
||||
al.summarizeSession(ctx, sessionKey)
|
||||
al.summarizeSession(ctx, sessionKey, cycle)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -250,7 +286,12 @@ func (al *AgentLoop) persistOversizedRecoveryRefs(ctx context.Context, sessionKe
|
|||
}
|
||||
|
||||
// summarizeSession summarizes the conversation history for a session.
|
||||
func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey string) {
|
||||
// An optional escalationLevel controls summarization aggressiveness (1=normal, 2=aggressive, 3=deterministic).
|
||||
func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey string, escalationLevel ...int) {
|
||||
level := 1
|
||||
if len(escalationLevel) > 0 && escalationLevel[0] > 0 {
|
||||
level = escalationLevel[0]
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(parentCtx, 120*time.Second)
|
||||
defer cancel()
|
||||
|
||||
|
|
@ -304,8 +345,8 @@ func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey stri
|
|||
part1 := validMessages[:mid]
|
||||
part2 := validMessages[mid:]
|
||||
|
||||
s1, _ := al.summarizeBatch(ctx, part1, "")
|
||||
s2, _ := al.summarizeBatch(ctx, part2, "")
|
||||
s1, _ := al.summarizeBatchEscalated(ctx, part1, "", level)
|
||||
s2, _ := al.summarizeBatchEscalated(ctx, part2, "", level)
|
||||
|
||||
// Merge them
|
||||
mergePrompt := fmt.Sprintf("Merge these two conversation summaries into one cohesive summary:\n\n1: %s\n\n2: %s", s1, s2)
|
||||
|
|
@ -316,7 +357,7 @@ func (al *AgentLoop) summarizeSession(parentCtx context.Context, sessionKey stri
|
|||
finalSummary = s1 + " " + s2
|
||||
}
|
||||
} else {
|
||||
finalSummary, _ = al.summarizeBatch(ctx, validMessages, summary)
|
||||
finalSummary, _ = al.summarizeBatchEscalated(ctx, validMessages, summary, level)
|
||||
}
|
||||
|
||||
if omitted && finalSummary != "" {
|
||||
|
|
@ -393,6 +434,38 @@ func (al *AgentLoop) summarizeBatch(ctx context.Context, batch []messages.Messag
|
|||
return al.callModel(ctx, prompt.String())
|
||||
}
|
||||
|
||||
// summarizeBatchEscalated applies escalation levels to summarization (LCM ADR-002).
|
||||
// Level 1: normal summary (existing behavior)
|
||||
// Level 2: aggressive bullet-point compression
|
||||
// Level 3: deterministic truncation (no LLM call)
|
||||
func (al *AgentLoop) summarizeBatchEscalated(ctx context.Context, batch []messages.Message, existingSummary string, level int) (string, error) {
|
||||
switch level {
|
||||
case 1:
|
||||
return al.summarizeBatch(ctx, batch, existingSummary)
|
||||
case 2:
|
||||
var prompt strings.Builder
|
||||
prompt.WriteString("Compress the following conversation into a VERY brief bullet-point list (max 5 bullets). Preserve only critical facts, decisions, and action items.\n")
|
||||
if existingSummary != "" {
|
||||
fmt.Fprintf(&prompt, "Prior context: %s\n", existingSummary)
|
||||
}
|
||||
prompt.WriteString("\nCONVERSATION:\n")
|
||||
for _, m := range batch {
|
||||
fmt.Fprintf(&prompt, "%s: %s\n", m.Role, m.Content)
|
||||
}
|
||||
return al.callModel(ctx, prompt.String())
|
||||
default:
|
||||
if len(batch) == 0 {
|
||||
return existingSummary, nil
|
||||
}
|
||||
first := batch[0]
|
||||
last := batch[len(batch)-1]
|
||||
return fmt.Sprintf("[Deterministic truncation of %d messages] First: %s: %s | Last: %s: %s",
|
||||
len(batch),
|
||||
first.Role, utils.Truncate(first.Content, 200),
|
||||
last.Role, utils.Truncate(last.Content, 200)), nil
|
||||
}
|
||||
}
|
||||
|
||||
// callModel makes a direct call to the Fantasy LanguageModel (no tools, no agent loop).
|
||||
// Used for summarization and other simple generation tasks.
|
||||
func (al *AgentLoop) callModel(ctx context.Context, prompt string) (string, error) {
|
||||
|
|
@ -423,136 +496,139 @@ func (al *AgentLoop) sessionsToMessagePairs(sessionKey string) []observation.Mes
|
|||
return pairs
|
||||
}
|
||||
|
||||
// dagCacheEntry holds the cached DAG compression output for a session,
|
||||
// enabling skip of recompression and repersistence when the compressible
|
||||
// portion hasn't grown since the last call.
|
||||
type dagCacheEntry struct {
|
||||
msgCount int
|
||||
rendered string
|
||||
dag *dag.DAG
|
||||
persistFailed bool
|
||||
// contextTreeCacheEntry caches rendered query-selected history blocks per session.
|
||||
type contextTreeCacheEntry struct {
|
||||
msgCount int
|
||||
query string
|
||||
rendered string
|
||||
}
|
||||
|
||||
// applyDAGCompression compresses old history into a DAG summary block and
|
||||
// returns only the tail messages that should be passed as raw conversation.
|
||||
// The compressed portion is injected into the system prompt via contextBuilder.
|
||||
// When memDelegate implements dag.DAGPersister, the DAG is persisted for dag_expand/describe/grep.
|
||||
//
|
||||
// Incremental optimization: if the compressible message count matches the
|
||||
// cached count, the previous DAG and rendered block are reused without
|
||||
// recompression or repersistence.
|
||||
func (al *AgentLoop) applyDAGCompression(ctx context.Context, sessionKey string, history []messages.Message) []messages.Message {
|
||||
const minHistoryForDAG = 16
|
||||
// applyContextTreeSelection selects relevant historical context via query-adaptive
|
||||
// Context Tree scoring and keeps only the raw tail as messages.
|
||||
func (al *AgentLoop) applyContextTreeSelection(ctx context.Context, sessionKey, query string, history []messages.Message) []messages.Message {
|
||||
const minHistoryForSelection = 16
|
||||
|
||||
if len(history) < minHistoryForDAG {
|
||||
al.contextBuilder.SetDAGBlock("")
|
||||
if len(history) < minHistoryForSelection {
|
||||
al.contextBuilder.SetContextTreeBlock("")
|
||||
return history
|
||||
}
|
||||
|
||||
budget := dag.ComputeBudget(al.contextWindow, dag.DefaultBudgetConfig())
|
||||
tailCount := dag.TailMessageCount(budget.RawTail)
|
||||
if tailCount >= len(history) {
|
||||
al.contextBuilder.SetDAGBlock("")
|
||||
al.contextBuilder.SetContextTreeBlock("")
|
||||
return history
|
||||
}
|
||||
|
||||
// Split: compress old, keep tail raw
|
||||
compressible := history[:len(history)-tailCount]
|
||||
tail := history[len(history)-tailCount:]
|
||||
|
||||
// Tool-call-aware: don't split on a "tool" message
|
||||
for len(tail) > 0 && tail[0].Role == "tool" && len(compressible) > 0 {
|
||||
tail = append([]messages.Message{compressible[len(compressible)-1]}, tail...)
|
||||
compressible = compressible[:len(compressible)-1]
|
||||
}
|
||||
|
||||
if len(compressible) == 0 {
|
||||
al.contextBuilder.SetDAGBlock("")
|
||||
al.contextBuilder.SetContextTreeBlock("")
|
||||
return history
|
||||
}
|
||||
|
||||
// Check DAG cache: skip recompression if compressible count hasn't changed
|
||||
if cached, ok := al.dagCache.Load(sessionKey); ok {
|
||||
entry := cached.(dagCacheEntry)
|
||||
if entry.msgCount == len(compressible) {
|
||||
al.contextBuilder.SetDAGBlock(entry.rendered)
|
||||
// Retry failed persistence from previous turn
|
||||
if entry.persistFailed {
|
||||
al.retryDAGPersist(ctx, sessionKey, entry)
|
||||
}
|
||||
cacheHit := false
|
||||
if cached, ok := al.contextTreeCache.Load(sessionKey); ok {
|
||||
entry := cached.(contextTreeCacheEntry)
|
||||
if entry.msgCount == len(compressible) && entry.query == query {
|
||||
al.contextBuilder.SetContextTreeBlock(entry.rendered)
|
||||
return tail
|
||||
}
|
||||
}
|
||||
|
||||
dagMsgs := make([]dag.Message, len(compressible))
|
||||
for i, m := range compressible {
|
||||
dagMsgs[i] = dag.Message{Role: m.Role, Content: m.Content}
|
||||
tree := contexttree.NewContextTree(contexttree.DefaultScoringConfig())
|
||||
rootID := tree.Root.ID
|
||||
for _, m := range compressible {
|
||||
tree.AddNode(rootID, contextNodeTypeForRole(m.Role), m.Content, nil, contexttree.ExtractTerms(m.Content))
|
||||
}
|
||||
|
||||
compressor := dag.NewCompressor(dag.DefaultCompressorConfig())
|
||||
d := compressor.Compress(dagMsgs)
|
||||
queryTerms := contexttree.ExtractTerms(query)
|
||||
if len(queryTerms) == 0 && len(tail) > 0 {
|
||||
queryTerms = contexttree.ExtractTerms(tail[len(tail)-1].Content)
|
||||
}
|
||||
|
||||
rendered := dag.RenderDAGForBudget(d, budget.DAGSummaries)
|
||||
al.contextBuilder.SetDAGBlock(rendered)
|
||||
scores := tree.ScoreAll(nil, queryTerms)
|
||||
nodes := make([]*contexttree.ContextNode, 0, len(tree.NodeIndex)-1)
|
||||
for id, node := range tree.NodeIndex {
|
||||
if node.Type == contexttree.NodeTypeRoot {
|
||||
continue
|
||||
}
|
||||
node.TotalScore = scores[id]
|
||||
nodes = append(nodes, node)
|
||||
}
|
||||
|
||||
// Cache the result
|
||||
al.dagCache.Store(sessionKey, dagCacheEntry{
|
||||
msgCount: len(compressible),
|
||||
rendered: rendered,
|
||||
dag: d,
|
||||
sort.Slice(nodes, func(i, j int) bool {
|
||||
if nodes[i].TotalScore == nodes[j].TotalScore {
|
||||
return nodes[i].CreatedAt.After(nodes[j].CreatedAt)
|
||||
}
|
||||
return nodes[i].TotalScore > nodes[j].TotalScore
|
||||
})
|
||||
|
||||
// Persist DAG for dag_expand, dag_describe, dag_grep (additive; in-memory behavior unchanged)
|
||||
persistOK := al.tryDAGPersist(ctx, sessionKey, d, len(compressible))
|
||||
if !persistOK {
|
||||
// Mark for retry on next cache hit
|
||||
al.dagCache.Store(sessionKey, dagCacheEntry{
|
||||
msgCount: len(compressible),
|
||||
rendered: rendered,
|
||||
dag: d,
|
||||
persistFailed: true,
|
||||
})
|
||||
selectionBudget := budget.DAGSummaries
|
||||
if selectionBudget <= 0 {
|
||||
selectionBudget = 512
|
||||
}
|
||||
selected := make([]*contexttree.ContextNode, 0, len(nodes))
|
||||
usedTokens := 0
|
||||
for _, node := range nodes {
|
||||
nodeTokens := observation.EstimateTokens(node.Content)
|
||||
if nodeTokens == 0 {
|
||||
continue
|
||||
}
|
||||
if usedTokens+nodeTokens > selectionBudget {
|
||||
continue
|
||||
}
|
||||
selected = append(selected, node)
|
||||
usedTokens += nodeTokens
|
||||
}
|
||||
|
||||
logger.DebugCF("agent", "DAG compression applied",
|
||||
map[string]interface{}{
|
||||
"total_msgs": len(history),
|
||||
"compressed_msgs": len(compressible),
|
||||
"tail_msgs": len(tail),
|
||||
"dag_nodes": len(d.Nodes),
|
||||
"cache_hit": false,
|
||||
})
|
||||
rendered := renderContextTreeSelection(selected)
|
||||
al.contextBuilder.SetContextTreeBlock(rendered)
|
||||
al.contextTreeCache.Store(sessionKey, contextTreeCacheEntry{msgCount: len(compressible), query: query, rendered: rendered})
|
||||
|
||||
logger.DebugCF("agent", "Context-Tree selection applied", map[string]interface{}{
|
||||
"total_msgs": len(history),
|
||||
"compressed_msgs": len(compressible),
|
||||
"tail_msgs": len(tail),
|
||||
"selected_nodes": len(selected),
|
||||
"selected_tokens": usedTokens,
|
||||
"cache_hit": cacheHit,
|
||||
})
|
||||
|
||||
return tail
|
||||
}
|
||||
|
||||
func (al *AgentLoop) tryDAGPersist(ctx context.Context, sessionKey string, d *dag.DAG, msgCount int) bool {
|
||||
dp, ok := al.memDelegate.(dag.DAGPersister)
|
||||
if !ok {
|
||||
return true
|
||||
func contextNodeTypeForRole(role string) contexttree.NodeType {
|
||||
switch role {
|
||||
case "tool":
|
||||
return contexttree.NodeTypeToolCall
|
||||
case "assistant":
|
||||
return contexttree.NodeTypeSummary
|
||||
default:
|
||||
return contexttree.NodeTypeMessage
|
||||
}
|
||||
if err := dp.PersistDAG(ctx, pkg.NAME, sessionKey, &dag.PersistSnapshot{
|
||||
FromMsgIdx: 0,
|
||||
ToMsgIdx: msgCount,
|
||||
MsgCount: msgCount,
|
||||
DAG: d,
|
||||
}); err != nil {
|
||||
logger.WarnCF("agent", "DAG persist failed (will retry next turn)",
|
||||
map[string]interface{}{"error": err.Error(), "session_key": sessionKey})
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (al *AgentLoop) retryDAGPersist(ctx context.Context, sessionKey string, entry dagCacheEntry) {
|
||||
if al.tryDAGPersist(ctx, sessionKey, entry.dag, entry.msgCount) {
|
||||
al.dagCache.Store(sessionKey, dagCacheEntry{
|
||||
msgCount: entry.msgCount,
|
||||
rendered: entry.rendered,
|
||||
dag: entry.dag,
|
||||
persistFailed: false,
|
||||
})
|
||||
func renderContextTreeSelection(selected []*contexttree.ContextNode) string {
|
||||
if len(selected) == 0 {
|
||||
return ""
|
||||
}
|
||||
var b strings.Builder
|
||||
for _, node := range selected {
|
||||
line := strings.ReplaceAll(node.Content, "\n", " ")
|
||||
line = strings.TrimSpace(line)
|
||||
if len(line) > 320 {
|
||||
line = line[:317] + "..."
|
||||
}
|
||||
fmt.Fprintf(&b, "[%s score=%.3f] %s\n", node.Type, node.TotalScore, line)
|
||||
}
|
||||
return strings.TrimSpace(b.String())
|
||||
}
|
||||
|
||||
func (al *AgentLoop) estimateTokens(msgs []messages.Message) int {
|
||||
|
|
|
|||
577
pkg/agent/summary.go
Normal file
577
pkg/agent/summary.go
Normal file
|
|
@ -0,0 +1,577 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"encoding/csv"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// ContentType represents the detected type of content for summarization.
|
||||
type ContentType string
|
||||
|
||||
const (
|
||||
ContentTypeJSON ContentType = "json"
|
||||
ContentTypeCode ContentType = "code"
|
||||
ContentTypeCSV ContentType = "csv"
|
||||
ContentTypeText ContentType = "text"
|
||||
ContentTypeError ContentType = "error"
|
||||
)
|
||||
|
||||
// TypeSummary contains a type-aware summary of content.
|
||||
type TypeSummary struct {
|
||||
ContentType ContentType `json:"content_type"`
|
||||
Summary string `json:"summary"`
|
||||
Details any `json:"details,omitempty"`
|
||||
}
|
||||
|
||||
// SummarizeContent detects content type and generates an appropriate summary.
|
||||
// This replaces blind truncation with meaningful previews based on content structure.
|
||||
func SummarizeContent(content string, maxLen int) TypeSummary {
|
||||
content = strings.TrimSpace(content)
|
||||
if content == "" {
|
||||
return TypeSummary{ContentType: ContentTypeText, Summary: "(empty)"}
|
||||
}
|
||||
|
||||
// Check for error patterns first (they can appear in any content type)
|
||||
if isError(content) {
|
||||
return summarizeError(content, maxLen)
|
||||
}
|
||||
|
||||
// Try JSON first (most specific)
|
||||
if isJSON(content) {
|
||||
return summarizeJSON(content, maxLen)
|
||||
}
|
||||
|
||||
// Try CSV
|
||||
if isCSV(content) {
|
||||
return summarizeCSV(content, maxLen)
|
||||
}
|
||||
|
||||
// Try code (heuristic-based)
|
||||
if isCode(content) {
|
||||
return summarizeCode(content, maxLen)
|
||||
}
|
||||
|
||||
// Default to text summarization
|
||||
return summarizeText(content, maxLen)
|
||||
}
|
||||
|
||||
// isJSON checks if content appears to be valid JSON.
|
||||
func isJSON(content string) bool {
|
||||
trimmed := strings.TrimSpace(content)
|
||||
if !strings.HasPrefix(trimmed, "{") && !strings.HasPrefix(trimmed, "[") {
|
||||
return false
|
||||
}
|
||||
var js any
|
||||
if err := json.Unmarshal([]byte(trimmed), &js); err != nil {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// summarizeJSON creates a JSON schema summary (keys for objects, length for arrays).
|
||||
func summarizeJSON(content string, maxLen int) TypeSummary {
|
||||
trimmed := strings.TrimSpace(content)
|
||||
|
||||
var data any
|
||||
if err := json.Unmarshal([]byte(trimmed), &data); err != nil {
|
||||
return TypeSummary{ContentType: ContentTypeJSON, Summary: truncateRunes(content, maxLen)}
|
||||
}
|
||||
|
||||
switch v := data.(type) {
|
||||
case map[string]any:
|
||||
keys := make([]string, 0, len(v))
|
||||
for k := range v {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
schema := fmt.Sprintf("JSON object with %d fields: %s", len(keys), strings.Join(keys, ", "))
|
||||
if len(schema) > maxLen {
|
||||
schema = schema[:maxLen-3] + "..."
|
||||
}
|
||||
return TypeSummary{
|
||||
ContentType: ContentTypeJSON,
|
||||
Summary: schema,
|
||||
Details: map[string]any{"keys": keys, "type": "object", "field_count": len(v)},
|
||||
}
|
||||
|
||||
case []any:
|
||||
// Show first few items as sample
|
||||
sampleCount := min(len(v), 3)
|
||||
var sampleItems []string
|
||||
for i := 0; i < sampleCount; i++ {
|
||||
switch item := v[i].(type) {
|
||||
case map[string]any:
|
||||
itemKeys := make([]string, 0, len(item))
|
||||
for k := range item {
|
||||
itemKeys = append(itemKeys, k)
|
||||
}
|
||||
sampleItems = append(sampleItems, fmt.Sprintf("{%d fields}", len(itemKeys)))
|
||||
case string:
|
||||
sampleItems = append(sampleItems, fmt.Sprintf("%q", truncateRunes(item, 30)))
|
||||
default:
|
||||
sampleItems = append(sampleItems, fmt.Sprintf("%v", item))
|
||||
}
|
||||
}
|
||||
|
||||
summary := fmt.Sprintf("JSON array with %d items", len(v))
|
||||
if len(sampleItems) > 0 {
|
||||
summary += fmt.Sprintf(" [sample: %s]", strings.Join(sampleItems, ", "))
|
||||
}
|
||||
summary = truncateSummary(summary, maxLen)
|
||||
return TypeSummary{
|
||||
ContentType: ContentTypeJSON,
|
||||
Summary: summary,
|
||||
Details: map[string]any{"length": len(v), "type": "array", "sample_count": sampleCount},
|
||||
}
|
||||
|
||||
default:
|
||||
return TypeSummary{ContentType: ContentTypeJSON, Summary: fmt.Sprintf("JSON scalar: %v", v)}
|
||||
}
|
||||
}
|
||||
|
||||
// isError checks if content contains error patterns like stack traces,
|
||||
// exception messages, or failure indicators at the start of lines.
|
||||
func isError(content string) bool {
|
||||
errorPatterns := []string{
|
||||
"error:", "exception:", "stack trace", "panic:", "fail:",
|
||||
"runtime error", "fatal:", "warning:", "syntax error",
|
||||
"compilation error", "build failed", "test failed",
|
||||
}
|
||||
|
||||
lines := strings.Split(content, "\n")
|
||||
errorCount := 0
|
||||
|
||||
for _, line := range lines {
|
||||
trimmed := strings.TrimSpace(line)
|
||||
lower := strings.ToLower(trimmed)
|
||||
|
||||
// Check for error patterns at the start of lines
|
||||
for _, pattern := range errorPatterns {
|
||||
if strings.HasPrefix(lower, pattern) {
|
||||
errorCount++
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Additional check for common error indicators anywhere in line
|
||||
if strings.Contains(lower, "--- stack trace ---") ||
|
||||
strings.Contains(lower, "caused by:") ||
|
||||
strings.Contains(lower, "at line") ||
|
||||
strings.Contains(lower, "error code:") {
|
||||
errorCount++
|
||||
}
|
||||
}
|
||||
|
||||
// Require at least 2 error indicators or 1 strong indicator
|
||||
return errorCount >= 2 ||
|
||||
strings.Contains(strings.ToLower(content), "panic:") ||
|
||||
strings.Contains(strings.ToLower(content), "stack trace")
|
||||
}
|
||||
|
||||
// summarizeError creates a summary of error content, extracting key error messages.
|
||||
func summarizeError(content string, maxLen int) TypeSummary {
|
||||
lines := strings.Split(content, "\n")
|
||||
|
||||
var errorMessages []string
|
||||
var contextLines []string
|
||||
|
||||
for _, line := range lines {
|
||||
trimmed := strings.TrimSpace(line)
|
||||
if trimmed == "" {
|
||||
continue
|
||||
}
|
||||
lower := strings.ToLower(trimmed)
|
||||
|
||||
// Capture error message lines
|
||||
if strings.HasPrefix(lower, "error:") ||
|
||||
strings.HasPrefix(lower, "exception:") ||
|
||||
strings.HasPrefix(lower, "panic:") ||
|
||||
strings.HasPrefix(lower, "fail:") ||
|
||||
strings.HasPrefix(lower, "fatal:") ||
|
||||
strings.HasPrefix(lower, "warning:") {
|
||||
errorMessages = append(errorMessages, trimmed)
|
||||
} else if len(contextLines) < 3 && !strings.Contains(lower, "at ") {
|
||||
// Capture first few non-stack-trace lines as context
|
||||
contextLines = append(contextLines, trimmed)
|
||||
}
|
||||
}
|
||||
|
||||
var summary string
|
||||
if len(errorMessages) > 0 {
|
||||
summary = "Error: " + errorMessages[0]
|
||||
if len(errorMessages) > 1 {
|
||||
summary += fmt.Sprintf(" (%d errors)", len(errorMessages))
|
||||
}
|
||||
} else if len(contextLines) > 0 {
|
||||
summary = "Error context: " + contextLines[0]
|
||||
} else {
|
||||
summary = "Error detected in content"
|
||||
}
|
||||
|
||||
// Include line count info
|
||||
lineCount := len(lines)
|
||||
summary = fmt.Sprintf("%s [%d lines]", summary, lineCount)
|
||||
|
||||
summary = truncateSummary(summary, maxLen)
|
||||
|
||||
return TypeSummary{
|
||||
ContentType: ContentTypeError,
|
||||
Summary: summary,
|
||||
Details: map[string]any{
|
||||
"error_count": len(errorMessages),
|
||||
"line_count": lineCount,
|
||||
"errors": errorMessages[:min(len(errorMessages), 5)],
|
||||
"has_panic": strings.Contains(strings.ToLower(content), "panic:"),
|
||||
"has_trace": strings.Contains(strings.ToLower(content), "stack trace"),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// isCSV checks if content appears to be CSV format.
|
||||
// Uses multiple heuristics to reduce false positives on code with commas:
|
||||
// - Header detection: first line should look like headers, not code
|
||||
// - Numeric data check: subsequent lines should contain numeric values
|
||||
// - Minimum 2 columns consistently across lines
|
||||
func isCSV(content string) bool {
|
||||
lines := strings.Split(strings.TrimSpace(content), "\n")
|
||||
if len(lines) < 2 {
|
||||
return false
|
||||
}
|
||||
|
||||
// Filter out empty lines for processing
|
||||
var nonEmptyLines []string
|
||||
for _, line := range lines {
|
||||
if strings.TrimSpace(line) != "" {
|
||||
nonEmptyLines = append(nonEmptyLines, line)
|
||||
}
|
||||
}
|
||||
if len(nonEmptyLines) < 2 {
|
||||
return false
|
||||
}
|
||||
|
||||
// Use at most 5 non-empty lines for analysis
|
||||
checkLines := nonEmptyLines
|
||||
if len(checkLines) > 5 {
|
||||
checkLines = checkLines[:5]
|
||||
}
|
||||
|
||||
// Check for consistent comma count (minimum 2 columns = at least 1 comma)
|
||||
commaCounts := make([]int, len(checkLines))
|
||||
for i, line := range checkLines {
|
||||
commaCounts[i] = strings.Count(line, ",")
|
||||
}
|
||||
|
||||
// Require at least 2 columns (1 comma) in all lines
|
||||
for _, count := range commaCounts {
|
||||
if count < 1 {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// Check for consistent column count (allowing for trailing commas)
|
||||
firstCount := commaCounts[0]
|
||||
consistent := 0
|
||||
for _, count := range commaCounts[1:] {
|
||||
if count == firstCount || count == firstCount-1 {
|
||||
consistent++
|
||||
}
|
||||
}
|
||||
if consistent < len(commaCounts)-1 {
|
||||
return false
|
||||
}
|
||||
|
||||
// Header detection: first line should look like headers
|
||||
// Headers typically don't contain common code patterns
|
||||
header := checkLines[0]
|
||||
headerLower := strings.ToLower(header)
|
||||
|
||||
// Reject headers that look like code statements
|
||||
codePatterns := []string{
|
||||
"func ", "def ", "class ", "if ", "for ", "while ", "return",
|
||||
"import ", "package ", "#include", "const ", "var ", "let ",
|
||||
"public ", "private ", "protected ", "static ", "void ",
|
||||
}
|
||||
for _, pattern := range codePatterns {
|
||||
if strings.HasPrefix(headerLower, pattern) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// Headers typically have balanced alphanumeric content
|
||||
// Count alphanumeric vs special chars in header fields
|
||||
fields := strings.Split(header, ",")
|
||||
if len(fields) < 2 {
|
||||
return false
|
||||
}
|
||||
|
||||
headerLooksValid := true
|
||||
for _, field := range fields {
|
||||
trimmed := strings.TrimSpace(field)
|
||||
if trimmed == "" {
|
||||
continue
|
||||
}
|
||||
// Headers should contain mostly letters, not symbols
|
||||
letterCount := 0
|
||||
symbolCount := 0
|
||||
for _, r := range trimmed {
|
||||
if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') {
|
||||
letterCount++
|
||||
} else if r == '(' || r == ')' || r == '{' || r == '}' || r == ';' || r == '=' {
|
||||
symbolCount++
|
||||
}
|
||||
}
|
||||
// If field has more symbols than letters, likely code
|
||||
if symbolCount > letterCount/2 {
|
||||
headerLooksValid = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if !headerLooksValid {
|
||||
return false
|
||||
}
|
||||
|
||||
// Check data lines for CSV-like content (numeric values, quoted strings)
|
||||
if len(checkLines) >= 2 {
|
||||
dataLines := checkLines[1:]
|
||||
numericLineCount := 0
|
||||
|
||||
for _, line := range dataLines {
|
||||
dataFields := strings.Split(line, ",")
|
||||
hasNumericField := false
|
||||
for _, field := range dataFields {
|
||||
trimmed := strings.TrimSpace(field)
|
||||
if trimmed == "" {
|
||||
continue
|
||||
}
|
||||
// Check if field looks like a number (integer, float, or quoted)
|
||||
if (trimmed[0] >= '0' && trimmed[0] <= '9') ||
|
||||
strings.HasPrefix(trimmed, "\"") ||
|
||||
strings.HasPrefix(trimmed, "'") ||
|
||||
trimmed == "true" || trimmed == "false" ||
|
||||
trimmed == "null" || trimmed == "NA" {
|
||||
hasNumericField = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if hasNumericField {
|
||||
numericLineCount++
|
||||
}
|
||||
}
|
||||
|
||||
// At least half of data lines should have CSV-like values
|
||||
if numericLineCount < len(dataLines)/2 {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// summarizeCSV creates a CSV header and stats summary.
|
||||
func summarizeCSV(content string, maxLen int) TypeSummary {
|
||||
reader := csv.NewReader(strings.NewReader(content))
|
||||
reader.FieldsPerRecord = -1 // Allow variable
|
||||
|
||||
// Read header
|
||||
header, err := reader.Read()
|
||||
if err != nil {
|
||||
return TypeSummary{ContentType: ContentTypeCSV, Summary: truncateRunes(content, maxLen)}
|
||||
}
|
||||
|
||||
// Count rows
|
||||
rowCount := 0
|
||||
for {
|
||||
_, err := reader.Read()
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
rowCount++
|
||||
}
|
||||
|
||||
summary := fmt.Sprintf("CSV with %d columns: %s (%d data rows)",
|
||||
len(header), strings.Join(header, ", "), rowCount)
|
||||
|
||||
summary = truncateSummary(summary, maxLen)
|
||||
|
||||
return TypeSummary{
|
||||
ContentType: ContentTypeCSV,
|
||||
Summary: summary,
|
||||
Details: map[string]any{"columns": header, "column_count": len(header), "row_count": rowCount},
|
||||
}
|
||||
}
|
||||
|
||||
// isCode uses heuristics to detect if content is source code.
|
||||
func isCode(content string) bool {
|
||||
codeIndicators := []string{
|
||||
"func ", "def ", "class ", "import ", "package ",
|
||||
"#include", "public class", "private ", "function ",
|
||||
"const ", "let ", "var ", "=> ", ":= ",
|
||||
}
|
||||
|
||||
lines := strings.Split(content, "\n")
|
||||
indicatorCount := 0
|
||||
|
||||
for _, line := range lines {
|
||||
trimmed := strings.TrimSpace(line)
|
||||
for _, indicator := range codeIndicators {
|
||||
if strings.HasPrefix(trimmed, indicator) {
|
||||
indicatorCount++
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// If more than 3 code-like lines, consider it code
|
||||
return indicatorCount > 3
|
||||
}
|
||||
|
||||
// summarizeCode creates a code structure summary (functions, classes, imports).
|
||||
func summarizeCode(content string, maxLen int) TypeSummary {
|
||||
lines := strings.Split(content, "\n")
|
||||
|
||||
var functions, classes, imports []string
|
||||
braceDepth := 0
|
||||
|
||||
for _, line := range lines {
|
||||
trimmed := strings.TrimSpace(line)
|
||||
|
||||
// Track brace depth for scope
|
||||
for _, r := range trimmed {
|
||||
switch r {
|
||||
case '{', '(', '[':
|
||||
braceDepth++
|
||||
case '}', ')', ']':
|
||||
braceDepth--
|
||||
}
|
||||
}
|
||||
|
||||
// Top-level declarations
|
||||
if braceDepth == 0 || braceDepth == 1 {
|
||||
if strings.HasPrefix(trimmed, "func ") || strings.HasPrefix(trimmed, "def ") ||
|
||||
strings.HasPrefix(trimmed, "function ") {
|
||||
parts := strings.Fields(trimmed)
|
||||
if len(parts) >= 2 {
|
||||
name := parts[1]
|
||||
if idx := strings.Index(name, "("); idx > 0 {
|
||||
name = name[:idx]
|
||||
}
|
||||
functions = append(functions, name)
|
||||
}
|
||||
}
|
||||
if strings.HasPrefix(trimmed, "class ") || strings.HasPrefix(trimmed, "public class ") ||
|
||||
strings.HasPrefix(trimmed, "type ") && strings.Contains(trimmed, "struct") {
|
||||
parts := strings.Fields(trimmed)
|
||||
if len(parts) >= 2 {
|
||||
name := parts[1]
|
||||
if idx := strings.Index(name, "{"); idx > 0 {
|
||||
name = name[:idx]
|
||||
}
|
||||
classes = append(classes, name)
|
||||
}
|
||||
}
|
||||
if strings.HasPrefix(trimmed, "import ") || strings.HasPrefix(trimmed, "#include") ||
|
||||
strings.HasPrefix(trimmed, "package ") || strings.HasPrefix(trimmed, "using ") {
|
||||
imports = append(imports, trimmed)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
summaryParts := []string{}
|
||||
if len(functions) > 0 {
|
||||
summaryParts = append(summaryParts, fmt.Sprintf("%d functions", len(functions)))
|
||||
}
|
||||
if len(classes) > 0 {
|
||||
summaryParts = append(summaryParts, fmt.Sprintf("%d types/classes", len(classes)))
|
||||
}
|
||||
if len(imports) > 0 {
|
||||
summaryParts = append(summaryParts, fmt.Sprintf("%d imports", len(imports)))
|
||||
}
|
||||
|
||||
summary := "Code: " + strings.Join(summaryParts, ", ")
|
||||
if len(functions) > 0 {
|
||||
funcList := strings.Join(functions[:min(len(functions), 5)], ", ")
|
||||
if len(functions) > 5 {
|
||||
funcList += "..."
|
||||
}
|
||||
summary += fmt.Sprintf(" [funcs: %s]", funcList)
|
||||
}
|
||||
|
||||
summary = truncateSummary(summary, maxLen)
|
||||
|
||||
return TypeSummary{
|
||||
ContentType: ContentTypeCode,
|
||||
Summary: summary,
|
||||
Details: map[string]any{
|
||||
"functions": functions,
|
||||
"classes": classes,
|
||||
"imports": len(imports),
|
||||
"function_count": len(functions),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// summarizeText creates a meaningful text excerpt (first + last parts).
|
||||
func summarizeText(content string, maxLen int) TypeSummary {
|
||||
lines := strings.Split(content, "\n")
|
||||
|
||||
// Get first few non-empty lines
|
||||
var firstLines []string
|
||||
for _, line := range lines {
|
||||
if strings.TrimSpace(line) != "" {
|
||||
firstLines = append(firstLines, line)
|
||||
if len(firstLines) >= 3 {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Get last few non-empty lines
|
||||
var lastLines []string
|
||||
for i := len(lines) - 1; i >= 0; i-- {
|
||||
if strings.TrimSpace(lines[i]) != "" {
|
||||
lastLines = append([]string{lines[i]}, lastLines...)
|
||||
if len(lastLines) >= 2 {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var summary string
|
||||
if len(firstLines) > 0 {
|
||||
summary = strings.Join(firstLines, "\n")
|
||||
if len(lastLines) > 0 && len(lines) > len(firstLines)+len(lastLines) {
|
||||
summary += "\n...\n" + strings.Join(lastLines, "\n")
|
||||
}
|
||||
}
|
||||
|
||||
// Word count
|
||||
wordCount := len(strings.Fields(content))
|
||||
summary = fmt.Sprintf("Text (%d words): %s", wordCount, summary)
|
||||
|
||||
summary = truncateSummary(summary, maxLen)
|
||||
|
||||
return TypeSummary{
|
||||
ContentType: ContentTypeText,
|
||||
Summary: summary,
|
||||
Details: map[string]any{"word_count": wordCount, "line_count": len(lines)},
|
||||
}
|
||||
}
|
||||
|
||||
func min(a, b int) int {
|
||||
if a < b {
|
||||
return a
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
// truncateSummary safely truncates a string to maxLen, adding "..." when there's room.
|
||||
func truncateSummary(s string, maxLen int) string {
|
||||
if len(s) <= maxLen {
|
||||
return s
|
||||
}
|
||||
if maxLen > 3 {
|
||||
return s[:maxLen-3] + "..."
|
||||
}
|
||||
return s[:maxLen]
|
||||
}
|
||||
70
pkg/agent/task_completion.go
Normal file
70
pkg/agent/task_completion.go
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/ids"
|
||||
"github.com/ZanzyTHEbar/dragonscale/pkg/logger"
|
||||
)
|
||||
|
||||
// TaskCompletion tracks the outcome of an agent run for RL analysis.
|
||||
type TaskCompletion struct {
|
||||
TaskID string
|
||||
Description string
|
||||
TokensUsed int
|
||||
ToolCalls int
|
||||
Errors int
|
||||
UserCorrections int
|
||||
Completed bool
|
||||
SelfReports []MemoryRating // MemoryID + Score
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
// MemoryRating represents a self-reported usefulness score for a memory.
|
||||
type MemoryRating struct {
|
||||
MemoryID ids.UUID
|
||||
Score int // 0-3 scale
|
||||
}
|
||||
|
||||
// TaskCompletionStore is the interface for storing task completion records.
|
||||
// Implemented by the memory delegate.
|
||||
type TaskCompletionStore interface {
|
||||
StoreTaskCompletion(ctx context.Context, agentID string, completion TaskCompletion, conversationID, runID ids.UUID) error
|
||||
GetCompletedTasks(ctx context.Context, agentID string, since time.Time) ([]TaskCompletion, error)
|
||||
UpdateMemorySelfReport(ctx context.Context, memoryID ids.UUID, score int) error
|
||||
}
|
||||
|
||||
// endTask stores task completion data and self-reports.
|
||||
// This is called at the end of an agent run to record the outcome for RL analysis.
|
||||
func (al *AgentLoop) endTask(ctx context.Context, conversationID, runID ids.UUID, completion TaskCompletion) error {
|
||||
if al.memDelegate == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Store self-report scores if the delegate implements RLStore
|
||||
if rlStore, ok := al.memDelegate.(interface {
|
||||
UpdateMemorySelfReport(ctx context.Context, memoryID ids.UUID, score int) error
|
||||
}); ok {
|
||||
for _, rating := range completion.SelfReports {
|
||||
if err := rlStore.UpdateMemorySelfReport(ctx, rating.MemoryID, rating.Score); err != nil {
|
||||
// Log but don't fail - self-reports are best-effort
|
||||
logger.WarnCF("agent", "Failed to store self-report",
|
||||
map[string]interface{}{
|
||||
"memory_id": rating.MemoryID.String(),
|
||||
"error": err.Error(),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Store task completion record if the delegate implements TaskCompletionStore
|
||||
if store, ok := al.memDelegate.(TaskCompletionStore); ok {
|
||||
if err := store.StoreTaskCompletion(ctx, pkg.NAME, completion, conversationID, runID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue