converge.go raw
1 // Package converge tracks lattice convergence during training.
2 //
3 // The lattice is considered converged when new text stops requiring new
4 // vertices. This is measured as the vertex creation rate: new vertices
5 // per token ingested. The rate follows a power law decay — fast growth
6 // early as basic structures are captured, then a long tail as rare
7 // constructions trickle in. Convergence is declared when the rate drops
8 // below a configurable threshold.
9 package converge
10
11 import (
12 "encoding/json"
13 "sync"
14
15 "git.mleku.dev/mleku/dendrite/pkg/ratio"
16 )
17
18 // DefaultThreshold is the default convergence threshold: 1 new vertex
19 // per 1,000,000 tokens. Below this rate, the lattice has captured
20 // essentially all structure present in the input distribution.
21 var DefaultThreshold = ratio.New(1, 1_000_000)
22
23 // DefaultWindowSize is the number of tokens per measurement window.
24 const DefaultWindowSize int64 = 100_000
25
26 // WindowStats records counts for a single measurement window.
27 type WindowStats struct {
28 TokensProcessed int64 `json:"tokens_processed"`
29 NewVertices int64 `json:"new_vertices"`
30 VertexRate ratio.Ratio `json:"vertex_rate"` // new_vertices / tokens_processed
31 }
32
33 // Report summarizes the convergence state.
34 type Report struct {
35 TotalTokens int64 `json:"total_tokens"`
36 TotalNewVertices int64 `json:"total_new_vertices"`
37 CurrentRate ratio.Ratio `json:"current_rate"`
38 Converged bool `json:"converged"`
39 WindowCount int `json:"window_count"`
40 RecentWindows []WindowStats `json:"recent_windows"`
41 }
42
43 // Tracker monitors vertex creation rate over sliding windows.
44 type Tracker struct {
45 mu sync.Mutex
46
47 windowSize int64
48 threshold ratio.Ratio
49
50 // Current window accumulators.
51 windowTokens int64
52 windowVertices int64
53
54 // Completed windows.
55 windows []WindowStats
56
57 // Lifetime totals.
58 totalTokens int64
59 totalVertices int64
60 }
61
62 // NewTracker creates a convergence tracker with the given window size
63 // and convergence threshold.
64 func NewTracker(windowSize int64, threshold ratio.Ratio) *Tracker {
65 if windowSize <= 0 {
66 windowSize = DefaultWindowSize
67 }
68 if threshold.IsZero() {
69 threshold = DefaultThreshold
70 }
71 return &Tracker{
72 windowSize: windowSize,
73 threshold: threshold,
74 }
75 }
76
77 // RecordToken increments the token counter by one.
78 // If the current window is full, it is finalized and a new one starts.
79 func (t *Tracker) RecordToken() {
80 t.mu.Lock()
81 defer t.mu.Unlock()
82 t.totalTokens++
83 t.windowTokens++
84 if t.windowTokens >= t.windowSize {
85 t.finalizeWindow()
86 }
87 }
88
89 // RecordTokens increments the token counter by n.
90 func (t *Tracker) RecordTokens(n int64) {
91 t.mu.Lock()
92 defer t.mu.Unlock()
93 t.totalTokens += n
94 t.windowTokens += n
95 for t.windowTokens >= t.windowSize {
96 t.finalizeWindow()
97 }
98 }
99
100 // RecordNewVertex increments the new vertex counter.
101 func (t *Tracker) RecordNewVertex() {
102 t.mu.Lock()
103 t.totalVertices++
104 t.windowVertices++
105 t.mu.Unlock()
106 }
107
108 // finalizeWindow closes the current measurement window and starts a new one.
109 // Must be called with lock held.
110 func (t *Tracker) finalizeWindow() {
111 ws := WindowStats{
112 TokensProcessed: t.windowTokens,
113 NewVertices: t.windowVertices,
114 }
115 if t.windowTokens > 0 {
116 ws.VertexRate = ratio.New(t.windowVertices, t.windowTokens)
117 }
118 t.windows = append(t.windows, ws)
119 t.windowTokens = 0
120 t.windowVertices = 0
121 }
122
123 // IsConverged reports whether the vertex creation rate in recent windows
124 // is below the threshold. Requires at least 3 completed windows to avoid
125 // false convergence on small inputs.
126 func (t *Tracker) IsConverged() bool {
127 t.mu.Lock()
128 defer t.mu.Unlock()
129
130 if len(t.windows) < 3 {
131 return false
132 }
133
134 // Check the last 3 windows. All must be below threshold.
135 for i := len(t.windows) - 3; i < len(t.windows); i++ {
136 if t.windows[i].VertexRate.Greater(t.threshold) {
137 return false
138 }
139 }
140 return true
141 }
142
143 // Report returns a summary of the current convergence state.
144 func (t *Tracker) Report() Report {
145 t.mu.Lock()
146 defer t.mu.Unlock()
147
148 r := Report{
149 TotalTokens: t.totalTokens,
150 TotalNewVertices: t.totalVertices,
151 WindowCount: len(t.windows),
152 }
153
154 if t.totalTokens > 0 {
155 r.CurrentRate = ratio.New(t.totalVertices, t.totalTokens)
156 }
157
158 // Converged check (same logic as IsConverged but without re-locking).
159 if len(t.windows) >= 3 {
160 r.Converged = true
161 for i := len(t.windows) - 3; i < len(t.windows); i++ {
162 if t.windows[i].VertexRate.Greater(t.threshold) {
163 r.Converged = false
164 break
165 }
166 }
167 }
168
169 // Include the last 10 windows for display.
170 start := 0
171 if len(t.windows) > 10 {
172 start = len(t.windows) - 10
173 }
174 r.RecentWindows = make([]WindowStats, len(t.windows)-start)
175 copy(r.RecentWindows, t.windows[start:])
176
177 return r
178 }
179
180 // Marshal serializes the tracker state to JSON for persistence.
181 func (t *Tracker) Marshal() ([]byte, error) {
182 t.mu.Lock()
183 defer t.mu.Unlock()
184 return json.Marshal(struct {
185 WindowSize int64 `json:"window_size"`
186 Windows []WindowStats `json:"windows"`
187 TotalTokens int64 `json:"total_tokens"`
188 TotalVertices int64 `json:"total_vertices"`
189 }{
190 WindowSize: t.windowSize,
191 Windows: t.windows,
192 TotalTokens: t.totalTokens,
193 TotalVertices: t.totalVertices,
194 })
195 }
196