Move the voice daemon wiring and startup out of voice.go
This commit is contained in:
kami
@@ -43,24 +43,17 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/kami/maven/internal/audio"
|
||||
"github.com/kami/maven/internal/config"
|
||||
"github.com/kami/maven/internal/delivery"
|
||||
"github.com/kami/maven/internal/delivery/voicesink"
|
||||
"github.com/kami/maven/internal/dialogue"
|
||||
"github.com/kami/maven/internal/ipc"
|
||||
"github.com/kami/maven/internal/llm"
|
||||
"github.com/kami/maven/internal/memory"
|
||||
"github.com/kami/maven/internal/pattern"
|
||||
"github.com/kami/maven/internal/phraser"
|
||||
@@ -72,219 +65,8 @@ import (
|
||||
"github.com/kami/maven/internal/ttsnorm"
|
||||
"github.com/kami/maven/internal/voice"
|
||||
"github.com/kami/maven/internal/weather"
|
||||
"github.com/kami/maven/internal/worker"
|
||||
)
|
||||
|
||||
// voiceWiring — everything the daemon needs to run the audio path. Held by
|
||||
// cmd/mavend/main.go alongside the other wirings; closed on shutdown.
|
||||
type voiceWiring struct {
|
||||
server *voice.Server
|
||||
sessions *voice.Sessions
|
||||
voiceSink delivery.Sink
|
||||
embedder router.Embedder
|
||||
handler *reactiveHandler // the reactive handler for IPC Chat
|
||||
// worker clients (set when configured as Remote): closed on shutdown so
|
||||
// mavsttd / mavttsd don't keep a stale conn into a restarting daemon.
|
||||
sttClient *worker.Client
|
||||
ttsClient *worker.Client
|
||||
}
|
||||
|
||||
// close releases the listener + worker conns. Safe to call on nil (when
|
||||
// voice is not wired — wireVoice returns nil,nil).
|
||||
func (w *voiceWiring) close() {
|
||||
if w == nil {
|
||||
return
|
||||
}
|
||||
if w.embedder != nil {
|
||||
_ = w.embedder.Close()
|
||||
}
|
||||
if w.server != nil {
|
||||
_ = w.server.Close()
|
||||
}
|
||||
if w.sttClient != nil {
|
||||
_ = w.sttClient.Close()
|
||||
}
|
||||
if w.ttsClient != nil {
|
||||
_ = w.ttsClient.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// wireVoice builds the audio path from cfg + a CoreAPI + a router. Returns
|
||||
// nil wiring + nil error when voice isn't enabled (the caller's voice sink
|
||||
// stays nil; the dispatcher's ChannelVoice routing drops silently).
|
||||
//
|
||||
// When voice is enabled, MUST wire a voicesink into the dispatcher's Voice
|
||||
// slot using w.sessions (the caller does that — see main.go).
|
||||
func wireVoice(cfg *config.Config, coreAPI ipc.CoreAPI, phr phraser.Phraser, memStore memory.Store, dataStore *store.Store, eco *ecosystemWiring) (*voiceWiring, error) {
|
||||
if cfg.Voice == nil || !cfg.Voice.Enabled {
|
||||
return nil, nil
|
||||
}
|
||||
w := &voiceWiring{}
|
||||
|
||||
// ----- stt (Stub in-process OR Remote via worker socket) -----
|
||||
var transcriber stt.Transcriber
|
||||
if cfg.Voice.Stt != nil && cfg.Voice.Stt.Socket != "" {
|
||||
c := worker.Dial(cfg.Voice.Stt.Socket)
|
||||
w.sttClient = c
|
||||
lang := cfg.Voice.Stt.Lang
|
||||
if lang == "" {
|
||||
lang = cfg.Voice.Lang
|
||||
}
|
||||
transcriber = stt.NewRemote(c, lang)
|
||||
} else {
|
||||
transcriber = stt.NewStub()
|
||||
}
|
||||
|
||||
// ----- tts (Stub in-process OR Remote) -----
|
||||
var synthesizer tts.Synthesizer
|
||||
if cfg.Voice.Tts != nil && cfg.Voice.Tts.Socket != "" {
|
||||
c := worker.Dial(cfg.Voice.Tts.Socket)
|
||||
w.ttsClient = c
|
||||
lang := cfg.Voice.Tts.Lang
|
||||
if lang == "" {
|
||||
lang = cfg.Voice.Lang
|
||||
}
|
||||
synthesizer = tts.NewRemote(c, lang, cfg.Voice.Tts.Voice)
|
||||
} else {
|
||||
synthesizer = tts.NewStub()
|
||||
}
|
||||
|
||||
// ----- router: embedder (ONNX when configured, floor HashEmbedder otherwise) -----
|
||||
var emb router.Embedder
|
||||
if cfg.Voice.Embedder != nil {
|
||||
onnx, err := router.NewONNXEmbedder(
|
||||
cfg.Voice.Embedder.ModelPath,
|
||||
cfg.Voice.Embedder.TokenizerPath,
|
||||
cfg.Voice.Embedder.LibPath,
|
||||
)
|
||||
if err != nil {
|
||||
w.close()
|
||||
return nil, fmt.Errorf("embedder: %w", err)
|
||||
}
|
||||
log.Printf("voice: onnx embedder loaded (%d dim)", onnx.Dim())
|
||||
emb = onnx
|
||||
} else {
|
||||
log.Printf("voice: embedder not configured, using HashEmbedder floor")
|
||||
emb = router.NewHashEmbedder(1024)
|
||||
}
|
||||
w.embedder = emb
|
||||
checkStoredEmbedder(dataStore, emb)
|
||||
|
||||
// ----- tool executor (the enabled act allowlist, store-backed) -----
|
||||
// Config tools are the declarative bootstrap: seed them into the store as
|
||||
// enabled (editing mavend.json IS the human enable act). Ad-hoc tools are
|
||||
// enabled later through the authed mavweb surface. The executor + matcher
|
||||
// both read the store live, so a newly-enabled tool is runnable without a
|
||||
// daemon restart.
|
||||
seedTools(coreAPI, cfg.Voice.Tools)
|
||||
exec := tool.NewExecutor(coreAPI, time.Duration(cfg.Voice.ToolTimeout))
|
||||
matcher := tool.NewMatcher(coreAPI)
|
||||
|
||||
// ----- weather provider (Open-Meteo when configured, Stub otherwise) -----
|
||||
var weatherProvider weather.Provider
|
||||
var weatherLocation string
|
||||
if cfg.Voice.Weather != nil && cfg.Voice.Weather.Provider == "open-meteo" {
|
||||
weatherProvider = weather.NewOpenMeteoProvider()
|
||||
weatherLocation = cfg.Voice.Weather.DefaultLocation
|
||||
log.Printf("voice: weather provider: open-meteo (default location: %s)", cfg.Voice.Weather.DefaultLocation)
|
||||
} else {
|
||||
weatherProvider = weather.NewStubProvider()
|
||||
log.Printf("voice: weather provider: stub (not configured)")
|
||||
}
|
||||
|
||||
// The replier uses the same llama-server as the phraser.
|
||||
var llmClient *llm.Client
|
||||
if lp, ok := phr.(*phraser.LLMPhraser); ok {
|
||||
llmClient = llm.New(lp.BaseURL(), 60*time.Second)
|
||||
}
|
||||
// ----- router (the cascade; floor examples seed the classifier) -----
|
||||
// The act matcher's allowlist is exactly the enabled tool names — the
|
||||
// router only matches acts the executor can run (one source of truth).
|
||||
threshold := cfg.Voice.RouterThreshold
|
||||
if threshold <= 0 {
|
||||
threshold = config.DefaultRouterThreshold
|
||||
}
|
||||
// The resident model routes by default: 63.2% of held-out intents right
|
||||
// against the classifier's 50.0%, at about 1s a turn instead of 30ms (see
|
||||
// config.VoiceConfig.LLMRouter). The classifier always stays wired as the
|
||||
// fallback, so a model error never breaks a turn.
|
||||
rtr := buildRouter(emb, matcher, threshold, pickLLMRouter(cfg.Voice.UseLLMRouter(), llmClient))
|
||||
|
||||
// ----- sessions registry (shared with voicesink) -----
|
||||
sessions := voice.NewSessions()
|
||||
w.sessions = sessions
|
||||
|
||||
// ----- voice sink (proactive nudges: dispatcher → voicesink → tts → push to client) -----
|
||||
w.voiceSink = voicesink.New(synthesizer, sessions)
|
||||
|
||||
// ----- memory (long-term vector storage) -----
|
||||
// Persistent (store-backed, survives restarts) when the daemon passes one;
|
||||
// falls back to the in-memory floor otherwise (tests / no-store paths).
|
||||
if memStore == nil {
|
||||
memStore = memory.NewInMemoryStore()
|
||||
}
|
||||
|
||||
// ----- dialogue (multi-turn slot carry-over; 2-min follow-up window) -----
|
||||
// Store-backed when the daemon passes a store, so a restart mid-conversation
|
||||
// keeps the thread (Vikunja #363). Sessions past their TTL are dropped on
|
||||
// load, never revived. Clarify's parked question stays in memory only.
|
||||
var dialogueSessions *dialogue.SessionStore
|
||||
if dataStore != nil {
|
||||
dialogueSessions = dialogue.NewPersistentSessionStore(2*time.Minute, dataStore)
|
||||
if err := dialogueSessions.Load(context.Background(), time.Now()); err != nil {
|
||||
log.Printf("dialogue: load saved sessions: %v", err)
|
||||
}
|
||||
} else {
|
||||
dialogueSessions = dialogue.NewSessionStore(2 * time.Minute)
|
||||
}
|
||||
clarifyStore := dialogue.NewClarifyStore(clarifyTTL)
|
||||
timeParser := router.NewPythonDateParser()
|
||||
|
||||
// ----- replier (LLM-backed when the engine is on, Stub floor otherwise) -----
|
||||
replier := voice.Replier(voice.NewStubReplier())
|
||||
if llmClient != nil {
|
||||
replier = newLLMReplier(llmClient, contextBlockFn(cfg, time.Now))
|
||||
}
|
||||
|
||||
// ----- the handler (the reactive path; closes over stt / tts / router / coreAPI / memory) -----
|
||||
h := &reactiveHandler{
|
||||
stt: transcriber,
|
||||
tts: synthesizer,
|
||||
router: rtr,
|
||||
embedder: emb,
|
||||
api: coreAPI,
|
||||
tools: exec,
|
||||
matcher: matcher,
|
||||
replier: replier,
|
||||
phraser: phr,
|
||||
now: time.Now,
|
||||
weatherProvider: weatherProvider,
|
||||
weatherLocation: weatherLocation,
|
||||
memStore: memStore,
|
||||
dataStore: dataStore,
|
||||
dialogueSessions: dialogueSessions,
|
||||
clarifyStore: clarifyStore,
|
||||
// 0 here (unset config) ⇒ the dialogue default.
|
||||
clarifyMaxAttempts: cfg.Voice.ClarifyMaxAttempts,
|
||||
extractor: router.Extractor{Time: timeParser, Acts: matcher, Facts: router.DefaultFactParser{}},
|
||||
queryMinScore: cfg.Voice.QueryMinScore,
|
||||
queryMinMargin: cfg.Voice.QueryMinMargin,
|
||||
timeParser: timeParser,
|
||||
ecosystem: eco,
|
||||
}
|
||||
|
||||
// ----- the server (TCP listener) -----
|
||||
srv := voice.NewServer(cfg.Voice.Bind, h, sessions)
|
||||
if err := srv.Listen(); err != nil {
|
||||
w.close()
|
||||
return nil, fmt.Errorf("voice listen: %w", err)
|
||||
}
|
||||
w.server = srv
|
||||
w.handler = h
|
||||
|
||||
return w, nil
|
||||
}
|
||||
|
||||
// reactiveHandler — voice.Handler implementation. One method: turn a
|
||||
// PushToTalkReq into a reply (audio + text). The handler is concurrency-
|
||||
// safe (the wired stt/tts/router/api all are); called from per-conn
|
||||
@@ -744,111 +526,6 @@ func (h *reactiveHandler) reply(ctx context.Context, text string, _ []string) (v
|
||||
return voice.PushToTalkResp{ReplyText: text, ReplyAudio: audioOut}, nil
|
||||
}
|
||||
|
||||
// pickLLMRouter returns the LLM router when the operator asked for it and there
|
||||
// is a llama-server to talk to, and nil otherwise. nil is safe: the cascade then
|
||||
// routes with the classifier, so an unusable setting costs accuracy, not turns.
|
||||
func pickLLMRouter(enabled bool, c *llm.Client) *router.LLMRouter {
|
||||
if !enabled {
|
||||
return nil
|
||||
}
|
||||
if c == nil {
|
||||
log.Printf("voice: voice.llm_router is on but there is no llama-server to route with (the phraser is not an LLM phraser) — using the classifier instead")
|
||||
return nil
|
||||
}
|
||||
log.Printf("voice: LLM router enabled")
|
||||
return router.NewLLMRouter(c)
|
||||
}
|
||||
|
||||
// buildRouter constructs the reactive-path router with the given embedder
|
||||
// and confidence threshold.
|
||||
// - stage-0 grammars from DefaultActMatcher whose fn allowlist is exactly
|
||||
// the enabled tool names (actFns) — the router only matches acts the
|
||||
// executor can run. Empty ⇒ every act refuses at the matcher.
|
||||
// - The embedder is provided by wireVoice: HashEmbedder (floor) when no
|
||||
// embedder config is present, or the ONNX multilingual model when
|
||||
// configured — same interface, one constructor change.
|
||||
// - 6 bootstrap examples covering the 5 intents + one compound-capture
|
||||
// placeholder. Spec calls for ~10 per intent at production; this is the
|
||||
// bootstrapping floor swapped by tuning the seed set later.
|
||||
// - Threshold is from voice.router_threshold config (default 0.55).
|
||||
func buildRouter(emb router.Embedder, acts router.ActMatcher, threshold float64, llmR *router.LLMRouter) *router.Router {
|
||||
cls := router.NewClassifier(emb)
|
||||
seedClassifier(cls)
|
||||
grammars := router.DefaultGrammars(acts)
|
||||
grammars = append(grammars, router.SystemTimeDateGrammars()...)
|
||||
grammars = append(grammars, router.ReminderGrammar())
|
||||
return router.New(router.Config{
|
||||
Grammars: grammars,
|
||||
Classifier: cls,
|
||||
Extractor: router.Extractor{
|
||||
Time: router.NewPythonDateParser(),
|
||||
Acts: acts,
|
||||
Facts: router.DefaultFactParser{},
|
||||
},
|
||||
Threshold: threshold,
|
||||
LLM: llmR,
|
||||
})
|
||||
}
|
||||
|
||||
// seedDir is the directory containing intent seed files. Each file is named
|
||||
// <intent>.txt and contains one training example per line (blank lines and
|
||||
// lines starting with # are ignored). Relative to the working directory.
|
||||
const seedDir = "models/seeds"
|
||||
|
||||
// seedClassifier floors the embedded examples so the cold-boot path
|
||||
// doesn't return ErrNoIntents. Loads examples from seedDir — one file per
|
||||
// intent (act.txt, reminder.txt, fact.txt, note.txt, query.txt). When the
|
||||
// classifier can't decide it falls through to Clarify — the last-resort
|
||||
// path asks the user to rephrase rather than guessing wrong.
|
||||
func seedClassifier(c *router.Classifier) {
|
||||
intents := []router.Intent{
|
||||
router.IntentAct,
|
||||
router.IntentReminder,
|
||||
router.IntentFact,
|
||||
router.IntentNote,
|
||||
router.IntentQuery,
|
||||
router.IntentChat,
|
||||
router.IntentSystem,
|
||||
}
|
||||
total := 0
|
||||
for _, intent := range intents {
|
||||
n, err := loadSeedFile(c, intent)
|
||||
if err != nil {
|
||||
log.Printf("voice: seed %s: %v", intent, err)
|
||||
continue
|
||||
}
|
||||
total += n
|
||||
}
|
||||
log.Printf("voice: loaded %d seed examples from %s", total, seedDir)
|
||||
}
|
||||
|
||||
func loadSeedFile(c *router.Classifier, intent router.Intent) (int, error) {
|
||||
path := filepath.Join(seedDir, string(intent)+".txt")
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("open %s: %w", path, err)
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
var count int
|
||||
sc := bufio.NewScanner(f)
|
||||
for sc.Scan() {
|
||||
line := strings.TrimSpace(sc.Text())
|
||||
if line == "" || strings.HasPrefix(line, "#") {
|
||||
continue
|
||||
}
|
||||
if err := c.AddExample(context.Background(), intent, line); err != nil {
|
||||
log.Printf("voice: seed %s: skipping %q: %v", intent, line, err)
|
||||
continue
|
||||
}
|
||||
count++
|
||||
}
|
||||
if err := sc.Err(); err != nil {
|
||||
return count, fmt.Errorf("scan %s: %w", path, err)
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
|
||||
// park stores a destructive act awaiting confirmation. Overwrites any prior
|
||||
// pending (last-asked wins — single-user box).
|
||||
func (h *reactiveHandler) park(fn string, args []string, phrase string) {
|
||||
@@ -1007,93 +684,3 @@ func actPhrase(fn string, args []string) string {
|
||||
}
|
||||
return fn + " " + strings.Join(args, " ")
|
||||
}
|
||||
|
||||
// seedTools upserts the config-declared tools into the store as enabled. Editing
|
||||
// mavend.json is a human act, so a config tool is enabled by definition; this
|
||||
// makes the declarative config the reproducible bootstrap while the store stays
|
||||
// the single runtime source of truth (mavweb enables ad-hoc ones on top).
|
||||
func seedTools(api ipc.CoreAPI, tools []config.ToolConfig) {
|
||||
ctx := context.Background()
|
||||
now := time.Now()
|
||||
n := 0
|
||||
for _, tc := range tools {
|
||||
if tc.Name == "" || len(tc.Cmd) == 0 {
|
||||
log.Printf("voice: skipping malformed tool config %+v", tc)
|
||||
continue
|
||||
}
|
||||
if err := api.EnableTool(ctx, tc.Name, tc.Cmd, tc.Destructive, tc.Scope, now); err != nil {
|
||||
log.Printf("voice: seed tool %q: %v", tc.Name, err)
|
||||
continue
|
||||
}
|
||||
n++
|
||||
}
|
||||
log.Printf("voice: seeded %d act tools from config", n)
|
||||
}
|
||||
|
||||
// reembedOnStart is the -reembed flag (set in run()). Opt-in on purpose: see
|
||||
// runReembed.
|
||||
var reembedOnStart bool
|
||||
|
||||
// checkStoredEmbedder compares the embedder we just loaded with the one that
|
||||
// wrote the vectors already in the DB (Vikunja #378).
|
||||
//
|
||||
// The two models we have both make 384-dim vectors, so a size check catches
|
||||
// nothing: after a swap, recall silently compares vectors from different
|
||||
// spaces and the scores are noise. So we say it out loud. Recall itself is not
|
||||
// changed here — the fix is `mavend -reembed`.
|
||||
func checkStoredEmbedder(dataStore *store.Store, emb router.Embedder) {
|
||||
if dataStore == nil {
|
||||
return
|
||||
}
|
||||
current := router.EmbedderID(emb)
|
||||
if reembedOnStart {
|
||||
runReembed(dataStore, emb, current)
|
||||
return
|
||||
}
|
||||
stored, mismatch, err := dataStore.CheckEmbedder(context.Background(), current)
|
||||
if err != nil {
|
||||
log.Printf("voice: embedder marker check failed: %v", err)
|
||||
return
|
||||
}
|
||||
if mismatch {
|
||||
log.Printf("voice: WARNING embedder MISMATCH — stored vectors were written by %q but the configured embedder is %q; recall scores are noise until the notes and facts are re-embedded — run `mavend -reembed` once (Vikunja #378)", stored, current)
|
||||
return
|
||||
}
|
||||
log.Printf("voice: embedder marker ok (%s)", current)
|
||||
}
|
||||
|
||||
// runReembed is the one-shot backfill behind -reembed.
|
||||
//
|
||||
// Why a flag and not automatic on mismatch: the embedder is ONNX on the
|
||||
// laptop's CPU, so a few thousand notes is minutes of work. Doing that silently
|
||||
// inside a normal start would look like the daemon hanging on boot. So the user
|
||||
// runs it once, deliberately, after an embedder swap; the mismatch warning
|
||||
// above tells them to. It re-embeds, logs what it did, and then the daemon
|
||||
// carries on serving as usual — no separate binary, no second start needed.
|
||||
func runReembed(dataStore *store.Store, emb router.Embedder, current string) {
|
||||
log.Printf("voice: re-embedding stored notes and facts with %s — this can take a few minutes, do not interrupt", current)
|
||||
res, err := dataStore.ReembedAll(context.Background(), current,
|
||||
// EmbedPassage, not EmbedQuery: these are stored texts being searched
|
||||
// FOR, which is the side they were written with.
|
||||
func(ctx context.Context, text string) ([]float32, error) {
|
||||
return router.EmbedPassage(ctx, emb, text)
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("voice: re-embed FAILED, nothing was changed and no marker was written — safe to run again: %v", err)
|
||||
return
|
||||
}
|
||||
if res.Skipped {
|
||||
log.Printf("voice: re-embed skipped — the stored vectors were already written by %s", current)
|
||||
return
|
||||
}
|
||||
log.Printf("voice: re-embed done — %d notes in the notes table, %d notes and %d facts in the memory index, took %s; stored vectors now belong to %s",
|
||||
res.Notes, res.MemNotes, res.Facts, res.Took.Round(time.Second), current)
|
||||
|
||||
// A row with no text cannot be re-embedded, so its vector is still the old
|
||||
// model's noise while the marker now says everything is current. Both write
|
||||
// paths always store the text, so this should be zero — say it loudly
|
||||
// rather than bury it in the line above if it ever isn't.
|
||||
if res.NoText > 0 {
|
||||
log.Printf("voice: WARNING %d stored rows had no text, so their vectors could not be re-embedded and are still noise; they will never match anything useful (Vikunja #378)", res.NoText)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,434 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/kami/maven/internal/config"
|
||||
"github.com/kami/maven/internal/delivery"
|
||||
"github.com/kami/maven/internal/delivery/voicesink"
|
||||
"github.com/kami/maven/internal/dialogue"
|
||||
"github.com/kami/maven/internal/ipc"
|
||||
"github.com/kami/maven/internal/llm"
|
||||
"github.com/kami/maven/internal/memory"
|
||||
"github.com/kami/maven/internal/phraser"
|
||||
"github.com/kami/maven/internal/router"
|
||||
"github.com/kami/maven/internal/store"
|
||||
"github.com/kami/maven/internal/stt"
|
||||
"github.com/kami/maven/internal/tool"
|
||||
"github.com/kami/maven/internal/tts"
|
||||
"github.com/kami/maven/internal/voice"
|
||||
"github.com/kami/maven/internal/weather"
|
||||
"github.com/kami/maven/internal/worker"
|
||||
)
|
||||
|
||||
// voiceWiring — everything the daemon needs to run the audio path. Held by
|
||||
// cmd/mavend/main.go alongside the other wirings; closed on shutdown.
|
||||
type voiceWiring struct {
|
||||
server *voice.Server
|
||||
sessions *voice.Sessions
|
||||
voiceSink delivery.Sink
|
||||
embedder router.Embedder
|
||||
handler *reactiveHandler // the reactive handler for IPC Chat
|
||||
// worker clients (set when configured as Remote): closed on shutdown so
|
||||
// mavsttd / mavttsd don't keep a stale conn into a restarting daemon.
|
||||
sttClient *worker.Client
|
||||
ttsClient *worker.Client
|
||||
}
|
||||
|
||||
// close releases the listener + worker conns. Safe to call on nil (when
|
||||
// voice is not wired — wireVoice returns nil,nil).
|
||||
func (w *voiceWiring) close() {
|
||||
if w == nil {
|
||||
return
|
||||
}
|
||||
if w.embedder != nil {
|
||||
_ = w.embedder.Close()
|
||||
}
|
||||
if w.server != nil {
|
||||
_ = w.server.Close()
|
||||
}
|
||||
if w.sttClient != nil {
|
||||
_ = w.sttClient.Close()
|
||||
}
|
||||
if w.ttsClient != nil {
|
||||
_ = w.ttsClient.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// wireVoice builds the audio path from cfg + a CoreAPI + a router. Returns
|
||||
// nil wiring + nil error when voice isn't enabled (the caller's voice sink
|
||||
// stays nil; the dispatcher's ChannelVoice routing drops silently).
|
||||
//
|
||||
// When voice is enabled, MUST wire a voicesink into the dispatcher's Voice
|
||||
// slot using w.sessions (the caller does that — see main.go).
|
||||
func wireVoice(cfg *config.Config, coreAPI ipc.CoreAPI, phr phraser.Phraser, memStore memory.Store, dataStore *store.Store, eco *ecosystemWiring) (*voiceWiring, error) {
|
||||
if cfg.Voice == nil || !cfg.Voice.Enabled {
|
||||
return nil, nil
|
||||
}
|
||||
w := &voiceWiring{}
|
||||
|
||||
// ----- stt (Stub in-process OR Remote via worker socket) -----
|
||||
var transcriber stt.Transcriber
|
||||
if cfg.Voice.Stt != nil && cfg.Voice.Stt.Socket != "" {
|
||||
c := worker.Dial(cfg.Voice.Stt.Socket)
|
||||
w.sttClient = c
|
||||
lang := cfg.Voice.Stt.Lang
|
||||
if lang == "" {
|
||||
lang = cfg.Voice.Lang
|
||||
}
|
||||
transcriber = stt.NewRemote(c, lang)
|
||||
} else {
|
||||
transcriber = stt.NewStub()
|
||||
}
|
||||
|
||||
// ----- tts (Stub in-process OR Remote) -----
|
||||
var synthesizer tts.Synthesizer
|
||||
if cfg.Voice.Tts != nil && cfg.Voice.Tts.Socket != "" {
|
||||
c := worker.Dial(cfg.Voice.Tts.Socket)
|
||||
w.ttsClient = c
|
||||
lang := cfg.Voice.Tts.Lang
|
||||
if lang == "" {
|
||||
lang = cfg.Voice.Lang
|
||||
}
|
||||
synthesizer = tts.NewRemote(c, lang, cfg.Voice.Tts.Voice)
|
||||
} else {
|
||||
synthesizer = tts.NewStub()
|
||||
}
|
||||
|
||||
// ----- router: embedder (ONNX when configured, floor HashEmbedder otherwise) -----
|
||||
var emb router.Embedder
|
||||
if cfg.Voice.Embedder != nil {
|
||||
onnx, err := router.NewONNXEmbedder(
|
||||
cfg.Voice.Embedder.ModelPath,
|
||||
cfg.Voice.Embedder.TokenizerPath,
|
||||
cfg.Voice.Embedder.LibPath,
|
||||
)
|
||||
if err != nil {
|
||||
w.close()
|
||||
return nil, fmt.Errorf("embedder: %w", err)
|
||||
}
|
||||
log.Printf("voice: onnx embedder loaded (%d dim)", onnx.Dim())
|
||||
emb = onnx
|
||||
} else {
|
||||
log.Printf("voice: embedder not configured, using HashEmbedder floor")
|
||||
emb = router.NewHashEmbedder(1024)
|
||||
}
|
||||
w.embedder = emb
|
||||
checkStoredEmbedder(dataStore, emb)
|
||||
|
||||
// ----- tool executor (the enabled act allowlist, store-backed) -----
|
||||
// Config tools are the declarative bootstrap: seed them into the store as
|
||||
// enabled (editing mavend.json IS the human enable act). Ad-hoc tools are
|
||||
// enabled later through the authed mavweb surface. The executor + matcher
|
||||
// both read the store live, so a newly-enabled tool is runnable without a
|
||||
// daemon restart.
|
||||
seedTools(coreAPI, cfg.Voice.Tools)
|
||||
exec := tool.NewExecutor(coreAPI, time.Duration(cfg.Voice.ToolTimeout))
|
||||
matcher := tool.NewMatcher(coreAPI)
|
||||
|
||||
// ----- weather provider (Open-Meteo when configured, Stub otherwise) -----
|
||||
var weatherProvider weather.Provider
|
||||
var weatherLocation string
|
||||
if cfg.Voice.Weather != nil && cfg.Voice.Weather.Provider == "open-meteo" {
|
||||
weatherProvider = weather.NewOpenMeteoProvider()
|
||||
weatherLocation = cfg.Voice.Weather.DefaultLocation
|
||||
log.Printf("voice: weather provider: open-meteo (default location: %s)", cfg.Voice.Weather.DefaultLocation)
|
||||
} else {
|
||||
weatherProvider = weather.NewStubProvider()
|
||||
log.Printf("voice: weather provider: stub (not configured)")
|
||||
}
|
||||
|
||||
// The replier uses the same llama-server as the phraser.
|
||||
var llmClient *llm.Client
|
||||
if lp, ok := phr.(*phraser.LLMPhraser); ok {
|
||||
llmClient = llm.New(lp.BaseURL(), 60*time.Second)
|
||||
}
|
||||
// ----- router (the cascade; floor examples seed the classifier) -----
|
||||
// The act matcher's allowlist is exactly the enabled tool names — the
|
||||
// router only matches acts the executor can run (one source of truth).
|
||||
threshold := cfg.Voice.RouterThreshold
|
||||
if threshold <= 0 {
|
||||
threshold = config.DefaultRouterThreshold
|
||||
}
|
||||
// The resident model routes by default: 63.2% of held-out intents right
|
||||
// against the classifier's 50.0%, at about 1s a turn instead of 30ms (see
|
||||
// config.VoiceConfig.LLMRouter). The classifier always stays wired as the
|
||||
// fallback, so a model error never breaks a turn.
|
||||
rtr := buildRouter(emb, matcher, threshold, pickLLMRouter(cfg.Voice.UseLLMRouter(), llmClient))
|
||||
|
||||
// ----- sessions registry (shared with voicesink) -----
|
||||
sessions := voice.NewSessions()
|
||||
w.sessions = sessions
|
||||
|
||||
// ----- voice sink (proactive nudges: dispatcher → voicesink → tts → push to client) -----
|
||||
w.voiceSink = voicesink.New(synthesizer, sessions)
|
||||
|
||||
// ----- memory (long-term vector storage) -----
|
||||
// Persistent (store-backed, survives restarts) when the daemon passes one;
|
||||
// falls back to the in-memory floor otherwise (tests / no-store paths).
|
||||
if memStore == nil {
|
||||
memStore = memory.NewInMemoryStore()
|
||||
}
|
||||
|
||||
// ----- dialogue (multi-turn slot carry-over; 2-min follow-up window) -----
|
||||
// Store-backed when the daemon passes a store, so a restart mid-conversation
|
||||
// keeps the thread (Vikunja #363). Sessions past their TTL are dropped on
|
||||
// load, never revived. Clarify's parked question stays in memory only.
|
||||
var dialogueSessions *dialogue.SessionStore
|
||||
if dataStore != nil {
|
||||
dialogueSessions = dialogue.NewPersistentSessionStore(2*time.Minute, dataStore)
|
||||
if err := dialogueSessions.Load(context.Background(), time.Now()); err != nil {
|
||||
log.Printf("dialogue: load saved sessions: %v", err)
|
||||
}
|
||||
} else {
|
||||
dialogueSessions = dialogue.NewSessionStore(2 * time.Minute)
|
||||
}
|
||||
clarifyStore := dialogue.NewClarifyStore(clarifyTTL)
|
||||
timeParser := router.NewPythonDateParser()
|
||||
|
||||
// ----- replier (LLM-backed when the engine is on, Stub floor otherwise) -----
|
||||
replier := voice.Replier(voice.NewStubReplier())
|
||||
if llmClient != nil {
|
||||
replier = newLLMReplier(llmClient, contextBlockFn(cfg, time.Now))
|
||||
}
|
||||
|
||||
// ----- the handler (the reactive path; closes over stt / tts / router / coreAPI / memory) -----
|
||||
h := &reactiveHandler{
|
||||
stt: transcriber,
|
||||
tts: synthesizer,
|
||||
router: rtr,
|
||||
embedder: emb,
|
||||
api: coreAPI,
|
||||
tools: exec,
|
||||
matcher: matcher,
|
||||
replier: replier,
|
||||
phraser: phr,
|
||||
now: time.Now,
|
||||
weatherProvider: weatherProvider,
|
||||
weatherLocation: weatherLocation,
|
||||
memStore: memStore,
|
||||
dataStore: dataStore,
|
||||
dialogueSessions: dialogueSessions,
|
||||
clarifyStore: clarifyStore,
|
||||
// 0 here (unset config) ⇒ the dialogue default.
|
||||
clarifyMaxAttempts: cfg.Voice.ClarifyMaxAttempts,
|
||||
extractor: router.Extractor{Time: timeParser, Acts: matcher, Facts: router.DefaultFactParser{}},
|
||||
queryMinScore: cfg.Voice.QueryMinScore,
|
||||
queryMinMargin: cfg.Voice.QueryMinMargin,
|
||||
timeParser: timeParser,
|
||||
ecosystem: eco,
|
||||
}
|
||||
|
||||
// ----- the server (TCP listener) -----
|
||||
srv := voice.NewServer(cfg.Voice.Bind, h, sessions)
|
||||
if err := srv.Listen(); err != nil {
|
||||
w.close()
|
||||
return nil, fmt.Errorf("voice listen: %w", err)
|
||||
}
|
||||
w.server = srv
|
||||
w.handler = h
|
||||
|
||||
return w, nil
|
||||
}
|
||||
|
||||
// pickLLMRouter returns the LLM router when the operator asked for it and there
|
||||
// is a llama-server to talk to, and nil otherwise. nil is safe: the cascade then
|
||||
// routes with the classifier, so an unusable setting costs accuracy, not turns.
|
||||
func pickLLMRouter(enabled bool, c *llm.Client) *router.LLMRouter {
|
||||
if !enabled {
|
||||
return nil
|
||||
}
|
||||
if c == nil {
|
||||
log.Printf("voice: voice.llm_router is on but there is no llama-server to route with (the phraser is not an LLM phraser) — using the classifier instead")
|
||||
return nil
|
||||
}
|
||||
log.Printf("voice: LLM router enabled")
|
||||
return router.NewLLMRouter(c)
|
||||
}
|
||||
|
||||
// buildRouter constructs the reactive-path router with the given embedder
|
||||
// and confidence threshold.
|
||||
// - stage-0 grammars from DefaultActMatcher whose fn allowlist is exactly
|
||||
// the enabled tool names (actFns) — the router only matches acts the
|
||||
// executor can run. Empty ⇒ every act refuses at the matcher.
|
||||
// - The embedder is provided by wireVoice: HashEmbedder (floor) when no
|
||||
// embedder config is present, or the ONNX multilingual model when
|
||||
// configured — same interface, one constructor change.
|
||||
// - 6 bootstrap examples covering the 5 intents + one compound-capture
|
||||
// placeholder. Spec calls for ~10 per intent at production; this is the
|
||||
// bootstrapping floor swapped by tuning the seed set later.
|
||||
// - Threshold is from voice.router_threshold config (default 0.55).
|
||||
func buildRouter(emb router.Embedder, acts router.ActMatcher, threshold float64, llmR *router.LLMRouter) *router.Router {
|
||||
cls := router.NewClassifier(emb)
|
||||
seedClassifier(cls)
|
||||
grammars := router.DefaultGrammars(acts)
|
||||
grammars = append(grammars, router.SystemTimeDateGrammars()...)
|
||||
grammars = append(grammars, router.ReminderGrammar())
|
||||
return router.New(router.Config{
|
||||
Grammars: grammars,
|
||||
Classifier: cls,
|
||||
Extractor: router.Extractor{
|
||||
Time: router.NewPythonDateParser(),
|
||||
Acts: acts,
|
||||
Facts: router.DefaultFactParser{},
|
||||
},
|
||||
Threshold: threshold,
|
||||
LLM: llmR,
|
||||
})
|
||||
}
|
||||
|
||||
// seedDir is the directory containing intent seed files. Each file is named
|
||||
// <intent>.txt and contains one training example per line (blank lines and
|
||||
// lines starting with # are ignored). Relative to the working directory.
|
||||
const seedDir = "models/seeds"
|
||||
|
||||
// seedClassifier floors the embedded examples so the cold-boot path
|
||||
// doesn't return ErrNoIntents. Loads examples from seedDir — one file per
|
||||
// intent (act.txt, reminder.txt, fact.txt, note.txt, query.txt). When the
|
||||
// classifier can't decide it falls through to Clarify — the last-resort
|
||||
// path asks the user to rephrase rather than guessing wrong.
|
||||
func seedClassifier(c *router.Classifier) {
|
||||
intents := []router.Intent{
|
||||
router.IntentAct,
|
||||
router.IntentReminder,
|
||||
router.IntentFact,
|
||||
router.IntentNote,
|
||||
router.IntentQuery,
|
||||
router.IntentChat,
|
||||
router.IntentSystem,
|
||||
}
|
||||
total := 0
|
||||
for _, intent := range intents {
|
||||
n, err := loadSeedFile(c, intent)
|
||||
if err != nil {
|
||||
log.Printf("voice: seed %s: %v", intent, err)
|
||||
continue
|
||||
}
|
||||
total += n
|
||||
}
|
||||
log.Printf("voice: loaded %d seed examples from %s", total, seedDir)
|
||||
}
|
||||
|
||||
func loadSeedFile(c *router.Classifier, intent router.Intent) (int, error) {
|
||||
path := filepath.Join(seedDir, string(intent)+".txt")
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("open %s: %w", path, err)
|
||||
}
|
||||
defer f.Close()
|
||||
|
||||
var count int
|
||||
sc := bufio.NewScanner(f)
|
||||
for sc.Scan() {
|
||||
line := strings.TrimSpace(sc.Text())
|
||||
if line == "" || strings.HasPrefix(line, "#") {
|
||||
continue
|
||||
}
|
||||
if err := c.AddExample(context.Background(), intent, line); err != nil {
|
||||
log.Printf("voice: seed %s: skipping %q: %v", intent, line, err)
|
||||
continue
|
||||
}
|
||||
count++
|
||||
}
|
||||
if err := sc.Err(); err != nil {
|
||||
return count, fmt.Errorf("scan %s: %w", path, err)
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
|
||||
// seedTools upserts the config-declared tools into the store as enabled. Editing
|
||||
// mavend.json is a human act, so a config tool is enabled by definition; this
|
||||
// makes the declarative config the reproducible bootstrap while the store stays
|
||||
// the single runtime source of truth (mavweb enables ad-hoc ones on top).
|
||||
func seedTools(api ipc.CoreAPI, tools []config.ToolConfig) {
|
||||
ctx := context.Background()
|
||||
now := time.Now()
|
||||
n := 0
|
||||
for _, tc := range tools {
|
||||
if tc.Name == "" || len(tc.Cmd) == 0 {
|
||||
log.Printf("voice: skipping malformed tool config %+v", tc)
|
||||
continue
|
||||
}
|
||||
if err := api.EnableTool(ctx, tc.Name, tc.Cmd, tc.Destructive, tc.Scope, now); err != nil {
|
||||
log.Printf("voice: seed tool %q: %v", tc.Name, err)
|
||||
continue
|
||||
}
|
||||
n++
|
||||
}
|
||||
log.Printf("voice: seeded %d act tools from config", n)
|
||||
}
|
||||
|
||||
// reembedOnStart is the -reembed flag (set in run()). Opt-in on purpose: see
|
||||
// runReembed.
|
||||
var reembedOnStart bool
|
||||
|
||||
// checkStoredEmbedder compares the embedder we just loaded with the one that
|
||||
// wrote the vectors already in the DB (Vikunja #378).
|
||||
//
|
||||
// The two models we have both make 384-dim vectors, so a size check catches
|
||||
// nothing: after a swap, recall silently compares vectors from different
|
||||
// spaces and the scores are noise. So we say it out loud. Recall itself is not
|
||||
// changed here — the fix is `mavend -reembed`.
|
||||
func checkStoredEmbedder(dataStore *store.Store, emb router.Embedder) {
|
||||
if dataStore == nil {
|
||||
return
|
||||
}
|
||||
current := router.EmbedderID(emb)
|
||||
if reembedOnStart {
|
||||
runReembed(dataStore, emb, current)
|
||||
return
|
||||
}
|
||||
stored, mismatch, err := dataStore.CheckEmbedder(context.Background(), current)
|
||||
if err != nil {
|
||||
log.Printf("voice: embedder marker check failed: %v", err)
|
||||
return
|
||||
}
|
||||
if mismatch {
|
||||
log.Printf("voice: WARNING embedder MISMATCH — stored vectors were written by %q but the configured embedder is %q; recall scores are noise until the notes and facts are re-embedded — run `mavend -reembed` once (Vikunja #378)", stored, current)
|
||||
return
|
||||
}
|
||||
log.Printf("voice: embedder marker ok (%s)", current)
|
||||
}
|
||||
|
||||
// runReembed is the one-shot backfill behind -reembed.
|
||||
//
|
||||
// Why a flag and not automatic on mismatch: the embedder is ONNX on the
|
||||
// laptop's CPU, so a few thousand notes is minutes of work. Doing that silently
|
||||
// inside a normal start would look like the daemon hanging on boot. So the user
|
||||
// runs it once, deliberately, after an embedder swap; the mismatch warning
|
||||
// above tells them to. It re-embeds, logs what it did, and then the daemon
|
||||
// carries on serving as usual — no separate binary, no second start needed.
|
||||
func runReembed(dataStore *store.Store, emb router.Embedder, current string) {
|
||||
log.Printf("voice: re-embedding stored notes and facts with %s — this can take a few minutes, do not interrupt", current)
|
||||
res, err := dataStore.ReembedAll(context.Background(), current,
|
||||
// EmbedPassage, not EmbedQuery: these are stored texts being searched
|
||||
// FOR, which is the side they were written with.
|
||||
func(ctx context.Context, text string) ([]float32, error) {
|
||||
return router.EmbedPassage(ctx, emb, text)
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("voice: re-embed FAILED, nothing was changed and no marker was written — safe to run again: %v", err)
|
||||
return
|
||||
}
|
||||
if res.Skipped {
|
||||
log.Printf("voice: re-embed skipped — the stored vectors were already written by %s", current)
|
||||
return
|
||||
}
|
||||
log.Printf("voice: re-embed done — %d notes in the notes table, %d notes and %d facts in the memory index, took %s; stored vectors now belong to %s",
|
||||
res.Notes, res.MemNotes, res.Facts, res.Took.Round(time.Second), current)
|
||||
|
||||
// A row with no text cannot be re-embedded, so its vector is still the old
|
||||
// model's noise while the marker now says everything is current. Both write
|
||||
// paths always store the text, so this should be zero — say it loudly
|
||||
// rather than bury it in the line above if it ever isn't.
|
||||
if res.NoText > 0 {
|
||||
log.Printf("voice: WARNING %d stored rows had no text, so their vectors could not be re-embedded and are still noise; they will never match anything useful (Vikunja #378)", res.NoText)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user