211 lines
8.6 KiB
Go
211 lines
8.6 KiB
Go
package llm
|
|
|
|
import (
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/structuredoutput"
|
|
pk "gitea.maximumdirect.net/eric/promptkit"
|
|
)
|
|
|
|
// TokenCount retains the distinction between zero and unavailable usage.
|
|
type TokenCount struct {
|
|
Value int `json:"value"`
|
|
Known bool `json:"known"`
|
|
Complete bool `json:"complete"`
|
|
}
|
|
|
|
type Usage struct {
|
|
Prompt TokenCount `json:"prompt"`
|
|
Completion TokenCount `json:"completion"`
|
|
Total TokenCount `json:"total"`
|
|
Cached TokenCount `json:"cached"`
|
|
CacheWrite TokenCount `json:"cache_write"`
|
|
}
|
|
|
|
// ExecutionObservation contains execution metadata only. It never contains
|
|
// endpoint URLs, credentials, messages, provider errors, or generated text.
|
|
type ExecutionObservation struct {
|
|
CallID uint64 `json:"call_id"`
|
|
Context structuredoutput.DiagnosticContext `json:"context"`
|
|
Kind string `json:"kind"`
|
|
RunID string `json:"run_id,omitempty"`
|
|
Time time.Time `json:"time"`
|
|
BackendID string `json:"backend_id,omitempty"`
|
|
ProfileID string `json:"profile_id,omitempty"`
|
|
Model string `json:"model,omitempty"`
|
|
Generation uint64 `json:"generation,omitempty"`
|
|
Attempt uint64 `json:"attempt,omitempty"`
|
|
Dispatched bool `json:"dispatched"`
|
|
CapacityWait time.Duration `json:"capacity_wait_ns"`
|
|
GenerationDuration time.Duration `json:"generation_duration_ns"`
|
|
RetryDelay time.Duration `json:"retry_delay_ns"`
|
|
Duration time.Duration `json:"duration_ns"`
|
|
Outcome string `json:"outcome,omitempty"`
|
|
Failure string `json:"failure,omitempty"`
|
|
StatusCode int `json:"status_code,omitempty"`
|
|
RetriesExhausted bool `json:"retries_exhausted,omitempty"`
|
|
Attempts uint64 `json:"attempts"`
|
|
TransientRetries uint64 `json:"transient_retries"`
|
|
StructuralRepairs uint64 `json:"structural_repairs"`
|
|
Usage Usage `json:"usage"`
|
|
}
|
|
|
|
type LogicalCall struct {
|
|
CallID uint64 `json:"call_id"`
|
|
Context structuredoutput.DiagnosticContext `json:"context"`
|
|
Prompt structuredoutput.PromptMetadata `json:"prompt"`
|
|
Started time.Time `json:"started"`
|
|
Finished time.Time `json:"finished,omitempty"`
|
|
WallDuration time.Duration `json:"wall_duration_ns"`
|
|
Failure structuredoutput.FailureKind `json:"failure,omitempty"`
|
|
}
|
|
|
|
type ObservationSnapshot struct {
|
|
Calls []LogicalCall `json:"calls"`
|
|
Events []ExecutionObservation `json:"events"`
|
|
}
|
|
|
|
type ExecutionSummary struct {
|
|
LogicalCalls uint64 `json:"logical_calls"`
|
|
ProviderAttempts uint64 `json:"provider_attempts"`
|
|
TransientRetries uint64 `json:"transient_retries"`
|
|
StructuralRepairs uint64 `json:"structural_repairs"`
|
|
Usage Usage `json:"usage"`
|
|
}
|
|
|
|
// ObservationCollector accepts short synchronous callbacks from parallel calls.
|
|
// Snapshot should be taken after the runner has joined active section work.
|
|
type ObservationCollector struct {
|
|
mu sync.Mutex
|
|
calls map[uint64]LogicalCall
|
|
events []ExecutionObservation
|
|
}
|
|
|
|
func NewObservationCollector() *ObservationCollector {
|
|
return &ObservationCollector{calls: make(map[uint64]LogicalCall)}
|
|
}
|
|
|
|
func (c *ObservationCollector) Hooks() CallHooks {
|
|
return CallHooks{OnCall: c.recordCall, OnEventWithCall: c.recordEvent}
|
|
}
|
|
|
|
func copyContext(value structuredoutput.DiagnosticContext) structuredoutput.DiagnosticContext {
|
|
if value.SectionIndex != nil {
|
|
index := *value.SectionIndex
|
|
value.SectionIndex = &index
|
|
}
|
|
if value.BatchIndex != nil {
|
|
index := *value.BatchIndex
|
|
value.BatchIndex = &index
|
|
}
|
|
return value
|
|
}
|
|
|
|
func (c *ObservationCollector) recordCall(record structuredoutput.CallRecord) {
|
|
if c == nil {
|
|
return
|
|
}
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
call := LogicalCall{CallID: record.CallID, Context: copyContext(record.Context), Prompt: record.Prompt, Started: record.Started, Finished: record.Finished, Failure: record.Failure}
|
|
if !record.Finished.IsZero() {
|
|
call.WallDuration = record.Finished.Sub(record.Started)
|
|
}
|
|
c.calls[record.CallID] = call
|
|
}
|
|
|
|
func (c *ObservationCollector) recordEvent(callID uint64, context structuredoutput.DiagnosticContext, event pk.ExecutionEvent) {
|
|
if c == nil {
|
|
return
|
|
}
|
|
count := func(value pk.ObservedTokenCount) TokenCount {
|
|
return TokenCount{Value: value.Value, Known: value.Known, Complete: value.Complete}
|
|
}
|
|
observation := ExecutionObservation{
|
|
CallID: callID, Context: copyContext(context), Kind: string(event.Kind), RunID: event.RunID,
|
|
Time: event.Time, BackendID: event.BackendID, ProfileID: event.ProfileID, Model: event.Model,
|
|
Generation: event.Generation, Attempt: event.Attempt, Dispatched: event.Dispatched,
|
|
CapacityWait: event.CapacityWait, GenerationDuration: event.GenerationDuration, RetryDelay: event.RetryDelay, Duration: event.Duration,
|
|
Outcome: string(event.Outcome), Failure: string(event.Failure), StatusCode: event.StatusCode, RetriesExhausted: event.RetriesExhausted,
|
|
Attempts: event.Attempts, TransientRetries: event.TransientRetries, StructuralRepairs: event.RepairCalls,
|
|
Usage: Usage{Prompt: count(event.Usage.PromptTokens), Completion: count(event.Usage.CompletionTokens), Total: count(event.Usage.TotalTokens), Cached: count(event.Usage.CachedTokens), CacheWrite: count(event.Usage.CacheWriteTokens)},
|
|
}
|
|
c.mu.Lock()
|
|
c.events = append(c.events, observation)
|
|
c.mu.Unlock()
|
|
}
|
|
|
|
func (c *ObservationCollector) Snapshot() ObservationSnapshot {
|
|
if c == nil {
|
|
return ObservationSnapshot{}
|
|
}
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
snapshot := ObservationSnapshot{Calls: make([]LogicalCall, 0, len(c.calls)), Events: make([]ExecutionObservation, len(c.events))}
|
|
for _, call := range c.calls {
|
|
call.Context = copyContext(call.Context)
|
|
snapshot.Calls = append(snapshot.Calls, call)
|
|
}
|
|
copy(snapshot.Events, c.events)
|
|
for i := range snapshot.Events {
|
|
snapshot.Events[i].Context = copyContext(snapshot.Events[i].Context)
|
|
}
|
|
sort.Slice(snapshot.Calls, func(i, j int) bool { return snapshot.Calls[i].CallID < snapshot.Calls[j].CallID })
|
|
sort.SliceStable(snapshot.Events, func(i, j int) bool {
|
|
if snapshot.Events[i].CallID != snapshot.Events[j].CallID {
|
|
return snapshot.Events[i].CallID < snapshot.Events[j].CallID
|
|
}
|
|
return snapshot.Events[i].Time.Before(snapshot.Events[j].Time)
|
|
})
|
|
return snapshot
|
|
}
|
|
|
|
// Summarize counts terminal usage once per logical call. PromptKit's terminal
|
|
// usage already aggregates its attempts, retries, and repairs.
|
|
func Summarize(snapshot ObservationSnapshot) ExecutionSummary {
|
|
result := ExecutionSummary{LogicalCalls: uint64(len(snapshot.Calls))}
|
|
terminal := make(map[uint64]ExecutionObservation)
|
|
for _, event := range snapshot.Events {
|
|
if event.Kind == string(pk.ExecutionFinished) {
|
|
terminal[event.CallID] = event
|
|
}
|
|
}
|
|
dispatched := false
|
|
aggregate := func(dst *TokenCount, source TokenCount) {
|
|
if source.Known {
|
|
dst.Value += source.Value
|
|
dst.Known = true
|
|
}
|
|
dst.Complete = dst.Complete && source.Complete
|
|
}
|
|
result.Usage = Usage{Prompt: TokenCount{Complete: true}, Completion: TokenCount{Complete: true}, Total: TokenCount{Complete: true}, Cached: TokenCount{Complete: true}, CacheWrite: TokenCount{Complete: true}}
|
|
for _, call := range snapshot.Calls {
|
|
event, ok := terminal[call.CallID]
|
|
if !ok {
|
|
continue
|
|
}
|
|
result.ProviderAttempts += event.Attempts
|
|
result.TransientRetries += event.TransientRetries
|
|
result.StructuralRepairs += event.StructuralRepairs
|
|
if event.Attempts == 0 {
|
|
continue
|
|
}
|
|
dispatched = true
|
|
aggregate(&result.Usage.Prompt, event.Usage.Prompt)
|
|
aggregate(&result.Usage.Completion, event.Usage.Completion)
|
|
aggregate(&result.Usage.Total, event.Usage.Total)
|
|
aggregate(&result.Usage.Cached, event.Usage.Cached)
|
|
aggregate(&result.Usage.CacheWrite, event.Usage.CacheWrite)
|
|
}
|
|
if !dispatched {
|
|
result.Usage.Prompt.Known = true
|
|
result.Usage.Completion.Known = true
|
|
result.Usage.Total.Known = true
|
|
result.Usage.Cached.Known = true
|
|
result.Usage.CacheWrite.Known = true
|
|
}
|
|
return result
|
|
}
|