;
; daemon.scm -- Continuous Learning Daemon (Phase 3)
;
; Keeps AtomSpace in memory for continuous learning.
; Avoids cold-start by running as a long-lived process.
;
; Architecture:
; 1. Cold start: Load from RocksDB once (43 min)
; 2. Accept text via named pipe (/tmp/opencog-text.fifo)
; 3. Incrementally update pair counts
; 4. Track stability metrics (MI delta, pair count delta)
; 5. Trigger MI recompute when deltas exceed thresholds
; 6. Persist to RocksDB continuously
;
; Usage:
; # Start daemon (cold start ~43 min)
; guile -s daemon.scm &
;
; # Feed text (from another terminal)
; echo "This is a test sentence." > /tmp/opencog-text.fifo
;
; # Stop daemon
; echo "QUIT" > /tmp/opencog-text.fifo
;
(use-modules (ice-9 format))
(use-modules (ice-9 rdelim))
(use-modules (ice-9 threads))
(use-modules (srfi srfi-1))
(use-modules (opencog) (opencog exec) (opencog persist) (opencog persist-rocks))
(use-modules (opencog lg))
(use-modules (opencog matrix))
(use-modules (opencog learn))
; ═══════════════════════════════════════════════════════════════════
; CONFIGURATION
; ═══════════════════════════════════════════════════════════════════
(define *rocks-path* "rocks://data/pairs.rdb")
(define *fifo-path* "/tmp/opencog-text.fifo")
(define *window-size* 6)
; Stability thresholds (trigger recompute when exceeded)
(define *mi-delta-threshold* 0.1) ; 10% change in average MI
(define *pair-count-threshold* 10000) ; 10K new pairs since last MI compute
; Memory bounds (based on 32GB system, leave 8GB for OS/apps)
; Current: 1.18M pairs = 3.5GB → ~3KB per pair
(define *max-pairs* 5000000) ; 5M pairs ≈ 15GB max
(define *prune-threshold* 3) ; Drop pairs with count < 3 when pruning
; ═══════════════════════════════════════════════════════════════════
; STATE TRACKING
; ═══════════════════════════════════════════════════════════════════
(define *running* #t)
(define *pairs-since-mi* 0)
(define *last-mi-time* 0)
(define *count-key* (Predicate "*-TruthValueKey-*"))
; ═══════════════════════════════════════════════════════════════════
; INCREMENTAL PAIR COUNTING (from Phase 1)
; ═══════════════════════════════════════════════════════════════════
(define *count-mutex* (make-mutex))
(define (count-inc! atom cnt)
"Increment count on ATOM, persist to RocksDB."
(if (not (cog-value atom *count-key*))
(begin
(lock-mutex *count-mutex*)
(if (not (cog-value atom *count-key*))
(fetch-atom atom))
(cog-inc-value! atom *count-key* cnt 2)
(unlock-mutex *count-mutex*))
(cog-inc-value! atom *count-key* cnt 2))
(store-atom atom))
(define (tokenize text)
"Split text into lowercase words."
(filter
(lambda (w) (>= (string-length w) 1))
(map (lambda (w)
(string-downcase (string-filter char-alphabetic? w)))
(string-split text #\space))))
(define (count-pairs-in-text text)
"Count word pairs in text within sliding window."
(let ((words (tokenize text))
(new-pairs 0))
;; Count each word
(for-each
(lambda (w)
(count-inc! (Word w) 1))
words)
;; Count pairs within window
(let ((n (length words)))
(do ((i 0 (+ i 1)))
((>= i n))
(do ((j (+ i 1) (+ j 1)))
((or (>= j n) (> (- j i) *window-size*)))
(let* ((w1 (list-ref words i))
(w2 (list-ref words j))
(pair (Edge (Bond "ANY")
(List (Word w1) (Word w2)))))
(count-inc! pair 1)
(set! new-pairs (+ new-pairs 1))))))
(set! *pairs-since-mi* (+ *pairs-since-mi* new-pairs))
new-pairs))
; ═══════════════════════════════════════════════════════════════════
; MEMORY MANAGEMENT
; ═══════════════════════════════════════════════════════════════════
(define (get-pair-count asa)
"Get current number of pairs in AtomSpace."
(length (asa 'get-all-elts)))
(define (get-count atom)
"Get count stored on atom."
(let ((tv (cog-value atom *count-key*)))
(if tv (cog-value-ref tv 2) 0)))
(define (prune-low-count-pairs! asa)
"Remove pairs with count below threshold to free memory."
(format #t "[~a] Pruning pairs with count < ~a...\n"
(current-time) *prune-threshold*)
(let ((removed 0))
(for-each
(lambda (pair)
(when (< (get-count pair) *prune-threshold*)
(cog-extract! pair)
(set! removed (+ removed 1))))
(asa 'get-all-elts))
(format #t "[~a] Pruned ~a low-count pairs.\n" (current-time) removed)
removed))
(define (check-memory-bounds! asa)
"Check if we've exceeded memory bounds, prune if needed."
(let ((pair-count (get-pair-count asa)))
(when (> pair-count *max-pairs*)
(format #t "[~a] WARNING: ~a pairs exceeds limit of ~a\n"
(current-time) pair-count *max-pairs*)
(prune-low-count-pairs! asa))))
; ═══════════════════════════════════════════════════════════════════
; STABILITY METRICS
; ═══════════════════════════════════════════════════════════════════
(define (should-recompute-mi?)
"Check if MI should be recomputed based on stability metrics."
(> *pairs-since-mi* *pair-count-threshold*))
(define (trigger-mi-recompute! asa)
"Recompute MI on all pairs."
(format #t "[~a] Triggering MI recompute (~a new pairs)...\n"
(current-time) *pairs-since-mi*)
(batch-all-pair-mi asa)
(set! *pairs-since-mi* 0)
(set! *last-mi-time* (current-time))
(format #t "[~a] MI recompute complete.\n" (current-time)))
; ═══════════════════════════════════════════════════════════════════
; TEXT INPUT PIPE
; ═══════════════════════════════════════════════════════════════════
(define (ensure-fifo-exists path)
"Create named pipe if it doesn't exist."
(unless (file-exists? path)
(system (format #f "mkfifo ~a" path))
(format #t "Created FIFO: ~a\n" path)))
(define (process-text-line line asa)
"Process one line of text. Returns #t to continue, #f to quit."
(cond
((string=? (string-trim-both line) "QUIT")
(format #t "[~a] Received QUIT command.\n" (current-time))
#f)
((string=? (string-trim-both line) "STATUS")
(let ((pair-count (get-pair-count asa))
(word-count (length (cog-get-atoms 'Word))))
(format #t "[~a] Status:\n" (current-time))
(format #t " Words: ~a\n" word-count)
(format #t " Pairs: ~a / ~a max (~,1f%)\n"
pair-count *max-pairs*
(* 100.0 (/ pair-count *max-pairs*)))
(format #t " Pairs since MI recompute: ~a / ~a threshold\n"
*pairs-since-mi* *pair-count-threshold*)
(format #t " Est. memory: ~,1f GB (of 24GB budget)\n"
(/ (* pair-count 3.0) 1000000.0))) ; ~3KB per pair
#t)
((string=? (string-trim-both line) "RECOMPUTE")
(trigger-mi-recompute! asa)
#t)
((> (string-length (string-trim-both line)) 0)
(let ((new-pairs (count-pairs-in-text line)))
(format #t "[~a] Processed: ~a new pairs\n" (current-time) new-pairs)
;; Check memory bounds
(check-memory-bounds! asa)
;; Check if MI recompute needed
(when (should-recompute-mi?)
(trigger-mi-recompute! asa))
#t))
(else #t)))
(define (read-text-loop asa)
"Main loop: read from FIFO and process text."
(format #t "[~a] Waiting for text on ~a...\n" (current-time) *fifo-path*)
(format #t " Commands: QUIT, STATUS, RECOMPUTE, or any text\n")
(let loop ()
(let* ((port (open-input-file *fifo-path*))
(line (read-line port)))
(close-input-port port)
(when (and (string? line) (process-text-line line asa))
(loop)))))
; ═══════════════════════════════════════════════════════════════════
; MAIN
; ═══════════════════════════════════════════════════════════════════
(define (run-daemon)
(format #t "═══ OpenCog Continuous Learning Daemon ═══\n\n")
;; Setup FIFO
(ensure-fifo-exists *fifo-path*)
;; Open database
(format #t "[~a] Opening database...\n" (current-time))
(define rsn (RocksStorageNode *rocks-path*))
(cog-open rsn)
;; Setup matrix API
(define ala (make-any-link-api))
(define aca (add-count-api ala))
(define asa (add-pair-stars aca))
;; Load existing data (COLD START - takes ~43 min)
(format #t "[~a] Loading existing pairs (cold start)...\n" (current-time))
(format #t " This takes ~40 minutes for 1M+ pairs.\n")
(format #t " After this, the daemon stays in memory.\n\n")
(asa 'fetch-pairs)
(format #t "[~a] Loaded ~a pairs, ~a words.\n"
(current-time)
(length (asa 'get-all-elts))
(length (asa 'left-basis)))
;; Enter main loop
(read-text-loop asa)
;; Cleanup
(format #t "[~a] Shutting down...\n" (current-time))
(barrier rsn)
(cog-close rsn)
(format #t "[~a] Done.\n" (current-time)))
; Only run if executed directly (not loaded as library)
(when (batch-mode?)
(run-daemon))