micro--go-micro/flow/analyze.go

package flow

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
	"time"

	"go-micro.dev/v6/ai"
)

// AnalyzeOptions configures Analyze.
type AnalyzeOptions struct {
	// MaxFeedbackSamples bounds the number of representative grader feedback
	// strings retained per candidate. Values <= 0 use a small default.
	MaxFeedbackSamples int
}

// AnalyzeOption configures Analyze.
type AnalyzeOption func(*AnalyzeOptions)

// AnalyzeMaxFeedbackSamples sets how many grader feedback examples are kept for
// each candidate in the report.
func AnalyzeMaxFeedbackSamples(n int) AnalyzeOption {
	return func(o *AnalyzeOptions) { o.MaxFeedbackSamples = n }
}

// Report is the machine-readable output of Analyze. Candidates are ordered from
// worst to best so an agent, CLI, or human can pick the first improvement to try.
type Report struct {
	Candidates []Candidate `json:"candidates"`
}

// Candidate identifies one underperforming flow step and the trace evidence that
// made it worth improving.
type Candidate struct {
	Step           string        `json:"step"`
	Metric         string        `json:"metric"`
	Score          float64       `json:"score"`
	Runs           int           `json:"runs"`
	Failures       int           `json:"failures"`
	PassRate       float64       `json:"pass_rate"`
	ErrorRate      float64       `json:"error_rate"`
	AverageRetries float64       `json:"average_retries"`
	P50Latency     time.Duration `json:"p50_latency"`
	P95Latency     time.Duration `json:"p95_latency"`
	SampleFeedback []string      `json:"sample_feedback,omitempty"`
	RunIDs         []string      `json:"run_ids,omitempty"`
}

// Analyze aggregates a bounded window of persisted flow runs and returns ranked
// hill-climbing candidates. It uses the same Run records read by Checkpoint.List:
// failed verification fields in step results drive pass-rate and feedback, step
// status drives error rate, and retry attempts contribute retry pressure. An
// empty window returns an empty report.
func Analyze(runs []Run, opts ...AnalyzeOption) Report {
	o := AnalyzeOptions{MaxFeedbackSamples: 3}
	for _, opt := range opts {
		opt(&o)
	}
	if o.MaxFeedbackSamples <= 0 {
		o.MaxFeedbackSamples = 3
	}

	stats := map[string]*stepStats{}
	for _, run := range runs {
		for _, step := range run.Steps {
			if step.Name == "" {
				continue
			}
			s := stats[step.Name]
			if s == nil {
				s = &stepStats{}
				stats[step.Name] = s
			}
			s.runs++
			s.runIDs = appendUnique(s.runIDs, run.ID)
			if step.Attempts > 1 {
				s.retries += step.Attempts - 1
			}
			if step.Status == "failed" || step.Error != "" {
				s.errors++
			}
			if len(run.Steps) > 0 && !run.Started.IsZero() && !run.Updated.IsZero() {
				s.latencies = append(s.latencies, run.Updated.Sub(run.Started)/time.Duration(len(run.Steps)))
			}
			passed, feedback, ok := verificationFields(step.Result)
			if ok {
				s.graded++
				if !passed {
					s.gradeFailures++
					if feedback != "" && len(s.feedback) < o.MaxFeedbackSamples {
						s.feedback = append(s.feedback, feedback)
					}
				}
			}
		}
	}

	report := Report{}
	for step, s := range stats {
		if s.runs == 0 {
			continue
		}
		failures := s.errors + s.gradeFailures
		passRate := 1.0
		if s.graded > 0 {
			passRate = float64(s.graded-s.gradeFailures) / float64(s.graded)
		} else if s.errors > 0 {
			passRate = float64(s.runs-s.errors) / float64(s.runs)
		}
		errorRate := float64(s.errors) / float64(s.runs)
		avgRetries := float64(s.retries) / float64(s.runs)
		score := float64(s.gradeFailures)*3 + float64(s.errors)*2 + avgRetries
		metric := "pass_rate"
		if s.gradeFailures == 0 && s.errors > 0 {
			metric = "error_rate"
		} else if s.gradeFailures == 0 && s.errors == 0 && s.retries > 0 {
			metric = "retry_count"
		}
		report.Candidates = append(report.Candidates, Candidate{
			Step: step, Metric: metric, Score: score, Runs: s.runs, Failures: failures,
			PassRate: passRate, ErrorRate: errorRate, AverageRetries: avgRetries,
			P50Latency: percentile(s.latencies, 0.50), P95Latency: percentile(s.latencies, 0.95),
			SampleFeedback: append([]string(nil), s.feedback...), RunIDs: append([]string(nil), s.runIDs...),
		})
	}
	sort.SliceStable(report.Candidates, func(i, j int) bool {
		a, b := report.Candidates[i], report.Candidates[j]
		if a.Score == b.Score {
			return a.Step < b.Step
		}
		return a.Score > b.Score
	})
	return report
}

type stepStats struct {
	runs, graded, gradeFailures, errors, retries int
	feedback, runIDs                             []string
	latencies                                    []time.Duration
}

// PromptOptimizer proposes prompt improvements for a candidate without mutating
// the source flow. Applying the returned prompt stays explicitly gated by the caller.
type PromptOptimizer struct{ model ai.Model }

// LLMOptimizer returns an optimizer that asks model to revise prompts for
// Analyze candidates. The model is injected so tests and callers can use mocks.
func LLMOptimizer(model ai.Model) *PromptOptimizer { return &PromptOptimizer{model: model} }

// OptimizePrompt asks the model for a revised prompt for candidate using the
// current prompt and trace feedback. It returns only the proposal; it never
// modifies a Flow, Step, or Checkpoint.
func (o *PromptOptimizer) OptimizePrompt(ctx context.Context, candidate Candidate, currentPrompt string) (string, error) {
	if o == nil || o.model == nil {
		return "", fmt.Errorf("flow: LLMOptimizer requires a model")
	}
	prompt := fmt.Sprintf("Revise this workflow step prompt to improve the failing step.\nStep: %s\nMetric: %s\nScore: %.2f\nFeedback:\n- %s\n\nCurrent prompt:\n%s\n\nReturn only the revised prompt.", candidate.Step, candidate.Metric, candidate.Score, strings.Join(candidate.SampleFeedback, "\n- "), currentPrompt)
	resp, err := o.model.Generate(ctx, &ai.Request{Prompt: prompt})
	if err != nil {
		return "", err
	}
	proposal := strings.TrimSpace(resp.Answer)
	if proposal == "" {
		proposal = strings.TrimSpace(resp.Reply)
	}
	if proposal == "" {
		return "", fmt.Errorf("flow: LLMOptimizer returned an empty prompt")
	}
	return proposal, nil
}

func verificationFields(result string) (bool, string, bool) {
	if result == "" {
		return false, "", false
	}
	var obj map[string]any
	if err := json.Unmarshal([]byte(result), &obj); err != nil {
		return false, "", false
	}
	v, ok := obj["verification_passed"].(bool)
	if !ok {
		return false, "", false
	}
	fb, _ := obj["verification_feedback"].(string)
	return v, fb, true
}

func appendUnique(values []string, value string) []string {
	if value == "" {
		return values
	}
	for _, v := range values {
		if v == value {
			return values
		}
	}
	return append(values, value)
}

func percentile(values []time.Duration, p float64) time.Duration {
	if len(values) == 0 {
		return 0
	}
	sorted := append([]time.Duration(nil), values...)
	sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })
	idx := int(float64(len(sorted)-1) * p)
	return sorted[idx]
}