mirror of
https://github.com/mudler/LocalAI.git
synced 2026-07-30 18:09:05 -04:00
* feat(llama-cpp): route Score through the slot loop Score previously bypassed the slot loop with a direct llama_decode: a conflict guard aborted the whole process if scoring raced generation, the config validator had to reject score alongside chat/completion/embeddings, and every candidate re-decoded the full shared prompt. Add SERVER_TASK_TYPE_SCORE to the (patched) upstream server so score tasks are scheduled like any other slot work: generation and scoring serialize naturally, the shared prompt is decoded once per call, and the slot's prompt cache carries the conversation prefix across calls. Context checkpoints at the score boundary and at the cache-divergence point keep SWA/hybrid/recurrent models (e.g. LFM2.5) from re-prefilling the whole prompt per candidate: warm-turn scoring on a 6-option set drops from ~8s to ~0.5s on a desktop CPU. The conflict guard and the validation split are removed; declaring score with generation usecases on one config is now supported and shares the slot cache. Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * feat(realtime): classifier wire types and pipeline config Wire types and YAML config for realtime classifier mode: sessions carry a localai_classifier extension (options with canned replies/tool calls, softmax threshold, normalization, history trimming, fallback modes, and a deterministic wake-word address gate), mirrored by pipeline.classifier in the model YAML and surfaced in the config-meta registry. The localai.classifier.result server event reports the full score distribution per turn. Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * feat(realtime): classifier response flow Classifier-mode responses: instead of autoregressive generation, each user turn is prefill-scored against the option list (router.ScoreClassifier prompt/candidate shapes over the Score primitive) and the winning option's canned reply and tool call are emitted through the existing response machinery. Below-threshold turns take the configured fallback (none / canned reply / generate); empty transcripts and unaddressed turns (wake word not mentioned) skip scoring entirely. The scoring probe defaults to the latest user message only — small scorers echo canned replies from prior turns back as the top option otherwise. Built for hardware that can afford prompt processing but not decode: with slot-based Score the option list stays KV-cached across turns, so a turn costs roughly one forward pass over the new words. session_update_error events now carry the validation cause instead of a generic message. Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * fix(realtime): bound the VAD tick's scan window and buffer retention The VAD tick loop re-scanned the entire input buffer every 300ms and only trimmed it on zero-segment ticks or commits. Audio that keeps producing segments without a committing pause (steady noise a mic pipeline lets through, music, continuous speech) grew the buffer toward the 100MB cap with each tick rescanning all of it — O(n^2), measured at ~3.3ms of silero per buffered second: past ~90s retained, ticks run back to back and pin ~4 cores until the stream stops. Silero's recurrent state only carries a few hundred ms of context, so rescanning old audio buys nothing. Clip the slice handed to the VAD to the largest silence the commit test can need to measure (server_vad silence window or the semantic eagerness fallback) plus a warm-up margin, and rebase the returned segment times so every downstream consumer keeps whole-buffer coordinates. An open turn whose clipped window is all silence now commits (the silence outran the window) instead of being discarded as no-speech. Independently, retain at most 90s of raw buffer, rebasing the live-feed and EOU cursors on trim — this also bounds the previously unbounded VAD-error path. Turn boundaries are otherwise unchanged: no forced commits, no new coordinator states. pipeline.turn_detection.vad_window_sec can widen the scan window; values below the automatic floor are ignored. The tick body is extracted into vadTick so specs can drive turn detection synchronously (same shape as classifySoundWindow); the babble reproduction that pinned 4 cores now plateaus under 10% of one core. Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * fix(backend): let per-model threads override the global default ModelOptions overrode a set per-model threads value with the app-level --threads whenever the latter was non-zero — and WithThreads defaults it to the physical core count, so it always was. The YAML threads: knob has been dead config: a tiny VAD model could never opt down from the global pool size. SetDefaults already fills an unset per-model value from the app config, which is the intended precedence; resolve threads through a helper that honors it (explicit threads: 0 still means unset). Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * chore(gallery): single-thread the silero VAD Silero is a ~2MB recurrent model with no exploitable graph parallelism: measured per-call latency is identical at 1 and 10 ORT threads, while every extra pool thread just spin-waits between the realtime loop's frequent tiny inferences. Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * docs(realtime): classifier mode, VAD scan window, threads precedence Document the realtime classifier mode (options, threshold guidance, wake-word address gate, empty-transcript handling), the VAD scan window and 90s buffer retention (pipeline.turn_detection.vad_window_sec), the per-model threads precedence, and the M3 classifier note in the realtime state-machine design doc. Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * perf(llama-cpp): score all candidates in one batched decode One scoring call is now a single SERVER_TASK_TYPE_SCORE task: the slot decodes the shared prefix (prompt + longest common candidate token prefix) once, then forks one sequence per candidate off it (metadata-only for the unified KV cache, copy-on-write for recurrent state) and decodes every candidate's unique tail in one llama_decode. Previously each candidate was its own task that restored the boundary checkpoint and re-decoded its full tail sequentially, paying per-candidate task and decode overhead. The context reserves SERVER_SCORE_FORK_SEQS extra sequence ids (and recurrent-state cells) beyond the parallel slots via the new common_params::n_seq_score_forks. Forking requires the unified KV cache (already this backend's default) since per-sequence streams would shrink n_ctx_seq; an explicit kv_unified:false disables forking and Score calls that need it fail cleanly. Candidates beyond the fork/output budget decode in successive chunks. Wire contract and scores are unchanged: per-token logprobs are stitched from the shared region and the forked tails. Verified bitwise deterministic call-to-call and independent of candidate order (no cross-fork leakage via equal-length candidate swap); ranking matches the per-candidate implementation on the drone battery (winner softmax 0.99996 vs 0.99997), and >16-candidate chunking, prefix-of-another and empty candidates all pass. Measured on a desktop CPU: warm /api/score calls 0.52s -> 0.23s; warm realtime classifier turns 196-303ms. The 9-candidate drone turn decodes ~17 unique tail tokens in one batch instead of nine sequential ~220ms checkpoint-restore tasks. Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * fix(realtime): gate scoring capacity by model usecase Reserve llama.cpp scoring slots only for models that explicitly declare the score usecase, while allowing score to coexist with chat and completion. Reject incompatible unified-KV settings and classifier activation on models without scoring capacity. Propagate application defaults when resolving realtime and preload pipeline stages so unset thread counts are resolved consistently without overriding explicit model settings. Assisted-by: Codex:gpt-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * fix(ci): honor APT mirrors in the prebuilt llama-cpp compile step The builder-prebuilt path installs gcc-14 with apt directly and ignored the APT_MIRROR/APT_PORTS_MIRROR build args the from-source path already honors, so an ubuntu mirror outage broke every arm64 backend build. Pass the args into the stage and run apt-mirror.sh (already in the build context via COPY . /LocalAI) before the apt step. Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * feat(realtime): classifier argument slots via constrained completion Hybrid classify-then-complete: a classifier option's canned tool call can declare typed argument slots (number | enum | string, with defaults and prompt hints) referenced as "{{name}}" in the arguments template. When the option wins, the slots are filled by a short grammar-constrained completion that continues the exact scoring prompt — rendered by the same cached ScoreClassifier, so the llama.cpp prompt cache is already warm — with the chosen route JSON re-opened at the first slot field. A GBNF grammar pins the field skeleton and frees only the values; temperature 0, a couple dozen tokens at most (~300ms on a desktop CPU for two slots). Slot declarations and hints ride the option descriptions in the shared system prompt, informing scoring and the fill alike at no per-turn token cost. The localai.classifier.result event carries the final arguments and a fill_latency_ms. On inference failure the slots' defaults apply; a slot without a default fails the response (or falls through with fallback.mode: generate). Slot filling requires completion alongside score in the scoring model's known_usecases. Verified end-to-end on the Pi drone demo: "fly forward three meters" in distance mode classifies forward and infers {"distance": 3, "units": "meters"} in ~310ms, and the drone flies exactly 3 units. Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * feat(realtime): splice filled slot values into classifier replies A classifier option's spoken reply can now reference its tool's argument slots ("Going forward {{distance}} {{units}}."): the values inferred by the slot-fill completion — or the recovery defaults — are spliced into the reply as plain text before it is emitted, so what the assistant says confirms what it actually inferred. Placeholders without a value stay literal, and options without slots are untouched. FillToolArguments now returns the raw slot values alongside the spliced arguments JSON to make the reply templating possible. Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * fix(realtime): harden classifier slot completion Reserve context for constrained slot filling, size completions from their encoded output, and encode enum grammar literals as valid JSON. Reject empty enum values and cover the failure modes with regression tests. Assisted-by: Codex:gpt-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * feat(realtime): prewarm the classifier scoring prompt on registration Swapping a session's classifier option list (a voice-switched command mode, for instance) made the next turns pay a full re-prefill of the new option-list prompt — measured 2.4s vs 0.3s warm on a desktop CPU, and worse: on hybrid-memory models like LFM2.5, whose state cannot be partially rewound (llama.cpp can only restore checkpoints), *every* probe change re-prefilled from scratch whenever the last checkpoint missed the probe boundary, so even same-list turns intermittently cost full prefills. Registering an option list (pipeline seed or session.update) now fires a best-effort background prewarm: two throwaway scores with distinct probes. The first prefills the new option-list prompt; the second, diverging exactly where per-turn probe text starts, plants the backend's rewind point (KV checkpoint) at the stable-prefix boundary that every real turn reuses. The prewarm hides behind the canned mode-switch reply — by the time it finishes speaking, the cache is warm. Idempotent per option set, detached from the registering request's lifetime. Measured on the drone demo (LFM2.5-1.2B, desktop CPU): first turn after a mode switch 2374ms -> 340ms; intermittent same-list full prefills (1.3-2.1s) all -> under 0.5s. For clients that swap lists frequently, options: [parallel:2] on the scoring model additionally keeps one slot per list via prefix-similarity routing (+26MB RSS, unified KV). Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * perf(llama-cpp): checkpoint scoring at the caller-declared stable prefix Hybrid-memory models (LFM2.5 shortconv, Qwen3.5 deltanet — where new small models are headed) cannot rewind their state, so any prompt-cache reuse that needs a rewind falls back to a full re-prefill. For classifier scoring that meant every probe change re-processed the whole option-list prompt: the server's checkpoints were placed reactively (at wherever the previous task happened to diverge), so a checkpoint past the next divergence was erased rather than restored — measured as intermittent 2-10s turns on prompts with a 95%+ common prefix. The classifier now computes the probe-invariant prompt prefix once (the byte-wise common prefix of two synthetic probe renders) and declares its length with every Score request; the server maps it to a token boundary and forces a KV checkpoint exactly there on each score prefill. That checkpoint sits at or before every future divergence under the same option list, so it always survives and always restores — repeat scoring costs probe+candidates regardless of how the probe changes. Also: - prewarm reruns on every option-list registration instead of memoizing per list: with boundary checkpoints a redundant rewarm costs two probe-sized decodes, while skipping one after a slot eviction (three lists sharing fewer slots evict in LRU cascades) silently moves a full re-prefill onto the user's next turn - new llama.cpp backend option rs_seq:N exposes bounded recurrent-state rollback outside speculative decoding; measured impractical for deltanet-scale states (65GB for 64 snapshots on Qwen3.5-4B) but cheap insurance for small-state models - docs: the multi-list recipe (parallel:N + sps:0.5 — the default slot similarity threshold funnels distinct lists onto one slot) Measured on the drone demo (LFM2.5-1.2B scorer, desktop CPU), steady state: every turn 285-421ms including mode switches, vs 2.4s post-switch and intermittent 1.3-2.9s re-prefills before. Assisted-by: Claude:claude-fable-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * fix(realtime): align classifier cache guidance Document the single-score prewarm behavior and clean the vendored score patch formatting. Assisted-by: Codex:gpt-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> * fix(llama-cpp): guard score task for fork backends TurboQuant and Bonsai reuse the primary gRPC server against llama.cpp forks that do not carry LocalAI's slot-based Score patches. Compile the Score integration only for the patched primary backend and return UNIMPLEMENTED from fork builds instead of referencing absent task types and common_params fields. Assisted-by: Codex:gpt-5 [gh] Signed-off-by: Richard Palethorpe <io@richiejp.com> * fix(dev): generate gRPC code before commit lint The coverage phase regenerates ignored protobuf bindings, but lint runs first and can fail against missing or stale output. Generate the pinned bindings before lint so the gate always type-checks the current schema. Assisted-by: Codex:gpt-5 Signed-off-by: Richard Palethorpe <io@richiejp.com> --------- Signed-off-by: Richard Palethorpe <io@richiejp.com>
711 lines
26 KiB
Go
711 lines
26 KiB
Go
package e2e_test
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/labstack/echo/v4"
|
|
localaiapp "github.com/mudler/LocalAI/core/application"
|
|
"github.com/mudler/LocalAI/core/config"
|
|
httpapi "github.com/mudler/LocalAI/core/http"
|
|
"github.com/mudler/LocalAI/pkg/system"
|
|
. "github.com/onsi/ginkgo/v2"
|
|
. "github.com/onsi/gomega"
|
|
"github.com/phayes/freeport"
|
|
"gopkg.in/yaml.v3"
|
|
|
|
"github.com/mudler/xlog"
|
|
"github.com/openai/openai-go/v3"
|
|
"github.com/openai/openai-go/v3/option"
|
|
)
|
|
|
|
var (
|
|
anthropicBaseURL string
|
|
ollamaBaseURL string
|
|
tmpDir string
|
|
backendPath string
|
|
modelsPath string
|
|
configPath string
|
|
app *echo.Echo
|
|
appCtx context.Context
|
|
appCancel context.CancelFunc
|
|
client openai.Client
|
|
apiPort int
|
|
apiURL string
|
|
mockBackendPath string
|
|
cloudProxyPath string
|
|
mcpServerURL string
|
|
mcpServerShutdown func()
|
|
localAIApp *localaiapp.Application
|
|
|
|
// Cloud-proxy fake upstreams. Live for the whole suite so the four
|
|
// cloud-proxy model YAMLs can point at their URLs at startup time.
|
|
cpOpenAIUpstream *fakeOpenAIUpstreamServer
|
|
cpAnthropicUpstream *fakeAnthropicUpstreamServer
|
|
|
|
// Live PII NER tier. Set only when PII_NER_MODEL_GGUF points at a
|
|
// privacy-filter GGUF and the privacy-filter backend is discoverable
|
|
// (REALTIME_BACKENDS_PATH). Empty => the NER specs Skip, exactly like the
|
|
// cloud-proxy specs Skip without their binary. This is what the hermetic
|
|
// suite cannot do (e2e_suite_test.go comment at the cp-translate detector):
|
|
// run the real GGUF NER tier instead of only the in-process pattern tier.
|
|
piiNERModel string
|
|
piiNERBlockModel string
|
|
)
|
|
|
|
var _ = BeforeSuite(func() {
|
|
var err error
|
|
|
|
// Create temporary directory
|
|
tmpDir, err = os.MkdirTemp("", "mock-backend-e2e-*")
|
|
Expect(err).ToNot(HaveOccurred())
|
|
|
|
backendPath = filepath.Join(tmpDir, "backends")
|
|
modelsPath = filepath.Join(tmpDir, "models")
|
|
Expect(os.MkdirAll(backendPath, 0755)).To(Succeed())
|
|
Expect(os.MkdirAll(modelsPath, 0755)).To(Succeed())
|
|
|
|
// Build mock backend
|
|
mockBackendDir := filepath.Join("..", "e2e", "mock-backend")
|
|
mockBackendPath = filepath.Join(backendPath, "mock-backend")
|
|
|
|
// Check if mock-backend binary exists in the mock-backend directory
|
|
possiblePaths := []string{
|
|
filepath.Join(mockBackendDir, "mock-backend"),
|
|
filepath.Join("tests", "e2e", "mock-backend", "mock-backend"),
|
|
filepath.Join("..", "..", "tests", "e2e", "mock-backend", "mock-backend"),
|
|
}
|
|
|
|
found := false
|
|
for _, p := range possiblePaths {
|
|
if _, err := os.Stat(p); err == nil {
|
|
mockBackendPath = p
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
|
|
if !found {
|
|
// Try to find it relative to current working directory
|
|
wd, _ := os.Getwd()
|
|
relPath := filepath.Join(wd, "..", "..", "tests", "e2e", "mock-backend", "mock-backend")
|
|
if _, err := os.Stat(relPath); err == nil {
|
|
mockBackendPath = relPath
|
|
found = true
|
|
}
|
|
}
|
|
|
|
Expect(found).To(BeTrue(), "mock-backend binary not found. Run 'make build-mock-backend' first")
|
|
|
|
// Make sure it's executable
|
|
Expect(os.Chmod(mockBackendPath, 0755)).To(Succeed())
|
|
|
|
// Create model config YAML
|
|
modelConfig := map[string]any{
|
|
"name": "mock-model",
|
|
"backend": "mock-backend",
|
|
"parameters": map[string]any{
|
|
"model": "mock-model.bin",
|
|
},
|
|
}
|
|
configPath = filepath.Join(modelsPath, "mock-model.yaml")
|
|
configYAML, err := yaml.Marshal(modelConfig)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(configPath, configYAML, 0644)).To(Succeed())
|
|
|
|
// Create model config for autoparser tests (NoGrammar so tool calls
|
|
// are driven entirely by the backend's ChatDeltas, not grammar enforcement)
|
|
autoparserConfig := map[string]any{
|
|
"name": "mock-model-autoparser",
|
|
"backend": "mock-backend",
|
|
"parameters": map[string]any{
|
|
"model": "mock-model.bin",
|
|
},
|
|
"function": map[string]any{
|
|
"grammar": map[string]any{
|
|
"disable": true,
|
|
},
|
|
},
|
|
}
|
|
autoparserPath := filepath.Join(modelsPath, "mock-model-autoparser.yaml")
|
|
autoparserYAML, err := yaml.Marshal(autoparserConfig)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(autoparserPath, autoparserYAML, 0644)).To(Succeed())
|
|
|
|
// Create model config for thinking model + autoparser tests.
|
|
// The chat template ends with <|channel>thought to simulate Gemma 4 thinking models.
|
|
// This triggers DetectThinkingStartToken and PrependThinkingTokenIfNeeded in the
|
|
// reasoning extraction path, reproducing a bug where clean content from the C++
|
|
// autoparser gets misclassified as unclosed reasoning.
|
|
thinkingAutoparserConfig := map[string]any{
|
|
"name": "mock-model-thinking-autoparser",
|
|
"backend": "mock-backend",
|
|
"parameters": map[string]any{
|
|
"model": "mock-model.bin",
|
|
},
|
|
"template": map[string]any{
|
|
"chat": "{{.Input}}\n<|turn>model\n<|channel>thought\n<channel|>",
|
|
},
|
|
"function": map[string]any{
|
|
"grammar": map[string]any{
|
|
"disable": true,
|
|
},
|
|
},
|
|
}
|
|
thinkingAutoparserPath := filepath.Join(modelsPath, "mock-model-thinking-autoparser.yaml")
|
|
thinkingAutoparserYAML, err := yaml.Marshal(thinkingAutoparserConfig)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(thinkingAutoparserPath, thinkingAutoparserYAML, 0644)).To(Succeed())
|
|
|
|
// Start mock MCP server and create MCP-enabled model config
|
|
mcpServerURL, mcpServerShutdown = startMockMCPServer()
|
|
mcpConfig := mcpModelConfig(mcpServerURL)
|
|
mcpConfigPath := filepath.Join(modelsPath, "mock-model-mcp.yaml")
|
|
mcpConfigYAML, err := yaml.Marshal(mcpConfig)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(mcpConfigPath, mcpConfigYAML, 0644)).To(Succeed())
|
|
|
|
// Create pipeline model configs for realtime API tests.
|
|
// Each component model uses the same mock-backend binary.
|
|
for _, name := range []string{"mock-vad", "mock-stt", "mock-llm", "mock-tts"} {
|
|
cfg := map[string]any{
|
|
"name": name,
|
|
"backend": "mock-backend",
|
|
"parameters": map[string]any{
|
|
"model": name + ".bin",
|
|
},
|
|
}
|
|
if name == "mock-llm" {
|
|
// Realtime classifier tests exercise concurrent generation and
|
|
// scoring on the same model, matching the production slot setup.
|
|
cfg["known_usecases"] = []string{"chat", "score"}
|
|
}
|
|
data, err := yaml.Marshal(cfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, name+".yaml"), data, 0644)).To(Succeed())
|
|
}
|
|
|
|
// Path-resolution model — declares relative draft_model / mmproj paths
|
|
// so the e2e test can confirm they arrive at the backend resolved
|
|
// against the models directory (regression guard for issue #9675).
|
|
pathResolutionCfg := map[string]any{
|
|
"name": "mock-model-path-resolution",
|
|
"backend": "mock-backend",
|
|
"parameters": map[string]any{
|
|
"model": "subdir/mock-main.bin",
|
|
},
|
|
"draft_model": "subdir/mock-draft.bin",
|
|
"mmproj": "subdir/mock-mmproj.bin",
|
|
}
|
|
pathResolutionData, err := yaml.Marshal(pathResolutionCfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, "mock-model-path-resolution.yaml"), pathResolutionData, 0644)).To(Succeed())
|
|
|
|
// Same but with an absolute draft_model — must not let a user-supplied
|
|
// config reach files outside the models directory. filepath.Join
|
|
// strips the leading slash, so /etc/passwd becomes <modelsPath>/etc/passwd.
|
|
pathEscapeCfg := map[string]any{
|
|
"name": "mock-model-path-escape",
|
|
"backend": "mock-backend",
|
|
"parameters": map[string]any{
|
|
"model": "subdir/mock-main.bin",
|
|
},
|
|
"draft_model": "/etc/passwd",
|
|
}
|
|
pathEscapeData, err := yaml.Marshal(pathEscapeCfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, "mock-model-path-escape.yaml"), pathEscapeData, 0644)).To(Succeed())
|
|
|
|
// Diarization model — known_usecases bypasses the FLAG_DIARIZATION
|
|
// backend-name guard so the /v1/audio/diarization route can dispatch
|
|
// to the mock backend.
|
|
diarizeCfg := map[string]any{
|
|
"name": "mock-diarize",
|
|
"backend": "mock-backend",
|
|
"known_usecases": []string{"FLAG_DIARIZATION"},
|
|
"parameters": map[string]any{
|
|
"model": "mock-diarize.bin",
|
|
},
|
|
}
|
|
diarizeData, err := yaml.Marshal(diarizeCfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, "mock-diarize.yaml"), diarizeData, 0644)).To(Succeed())
|
|
|
|
// Pipeline model that wires the component models together.
|
|
pipelineCfg := map[string]any{
|
|
"name": "realtime-pipeline",
|
|
"pipeline": map[string]any{
|
|
"vad": "mock-vad",
|
|
"transcription": "mock-stt",
|
|
"llm": "mock-llm",
|
|
"tts": "mock-tts",
|
|
},
|
|
}
|
|
pipelineData, err := yaml.Marshal(pipelineCfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, "realtime-pipeline.yaml"), pipelineData, 0644)).To(Succeed())
|
|
|
|
// Classifier-mode pipeline (LocalAI extension): responses are
|
|
// prefill-scored against the option list via the mock backend's
|
|
// ROUTE_HINT-driven Score instead of being generated. Threshold 0.6:
|
|
// a hinted option scores ≈0.99, no hint leaves a uniform distribution
|
|
// (0.5 for two options) and triggers the fallback reply.
|
|
classifierPipelineCfg := map[string]any{
|
|
"name": "realtime-pipeline-classifier",
|
|
"pipeline": map[string]any{
|
|
"vad": "mock-vad",
|
|
"transcription": "mock-stt",
|
|
"llm": "mock-llm",
|
|
"tts": "mock-tts",
|
|
"classifier": map[string]any{
|
|
"enabled": true,
|
|
"threshold": 0.6,
|
|
"fallback": map[string]any{"mode": "reply", "reply": "Say again?"},
|
|
"options": []map[string]any{
|
|
{
|
|
"id": "up",
|
|
"description": "the user asks the drone to fly up",
|
|
"reply": "Going up.",
|
|
"tool": map[string]any{"name": "move", "arguments": map[string]any{"direction": "up"}},
|
|
},
|
|
{
|
|
"id": "greeting",
|
|
"description": "the user greets the assistant",
|
|
"reply": "Hello.",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
classifierPipelineData, err := yaml.Marshal(classifierPipelineCfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, "realtime-pipeline-classifier.yaml"), classifierPipelineData, 0644)).To(Succeed())
|
|
|
|
// Speaker-recognition model (mock-backend) + a voice-recognition-gated
|
|
// pipeline for the realtime gate e2e. The reference WAV carries a positive
|
|
// DC bias so the mock embeds it to one orthogonal "speaker"; the test then
|
|
// drives matching (authorized) and opposite-bias (unauthorized) audio.
|
|
speakerCfg := map[string]any{
|
|
"name": "mock-speaker",
|
|
"backend": "mock-backend",
|
|
"parameters": map[string]any{"model": "mock-speaker.bin"},
|
|
}
|
|
speakerData, err := yaml.Marshal(speakerCfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, "mock-speaker.yaml"), speakerData, 0644)).To(Succeed())
|
|
|
|
voiceRefPath := filepath.Join(modelsPath, "e2e-voice-ref.wav")
|
|
Expect(os.WriteFile(voiceRefPath, wavFromPCM(pcmWithDC(300, 16000, 1000, 8000), 16000), 0644)).To(Succeed())
|
|
|
|
gatedCfg := map[string]any{
|
|
"name": "realtime-pipeline-gated",
|
|
"pipeline": map[string]any{
|
|
"vad": "mock-vad",
|
|
"transcription": "mock-stt",
|
|
"llm": "mock-llm",
|
|
"tts": "mock-tts",
|
|
"voice_recognition": map[string]any{
|
|
"model": "mock-speaker",
|
|
"mode": "verify",
|
|
"threshold": 0.25,
|
|
"when": "every",
|
|
"on_reject": "drop_event",
|
|
"references": []map[string]any{
|
|
{"name": "e2e-speaker", "audio": voiceRefPath},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
gatedData, err := yaml.Marshal(gatedCfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, "realtime-pipeline-gated.yaml"), gatedData, 0644)).To(Succeed())
|
|
|
|
// Identity-surfacing pipeline: the same speaker backend, but enforce:false
|
|
// (never drop a turn) plus an identity block so the server emits the
|
|
// conversation.item.speaker event and personalizes the LLM turn. Used by the
|
|
// speaker-identity e2e specs.
|
|
identityCfg := map[string]any{
|
|
"name": "realtime-pipeline-identity",
|
|
"pipeline": map[string]any{
|
|
"vad": "mock-vad",
|
|
"transcription": "mock-stt",
|
|
"llm": "mock-llm",
|
|
"tts": "mock-tts",
|
|
"voice_recognition": map[string]any{
|
|
"model": "mock-speaker",
|
|
"mode": "verify",
|
|
"threshold": 0.25,
|
|
"when": "every",
|
|
"enforce": false,
|
|
"references": []map[string]any{
|
|
{"name": "e2e-speaker", "audio": voiceRefPath},
|
|
},
|
|
"identity": map[string]any{
|
|
"announce": true,
|
|
"announce_unknown": true,
|
|
"personalize": true,
|
|
"inject_name": true,
|
|
"inject_system_note": true,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
identityData, err := yaml.Marshal(identityCfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, "realtime-pipeline-identity.yaml"), identityData, 0644)).To(Succeed())
|
|
|
|
// Router model setup: a score classifier (mock-backend Score) selects
|
|
// between two candidate chat models based on keyword matches against the
|
|
// candidate label fragments. Exercises the full RouteModel middleware path
|
|
// — probe extraction, ScoreClassifier.fitMessages (with the classifier's
|
|
// real TokenizeString and ContextSize wired), Score RPC, and fanout to
|
|
// the chosen candidate. The classifier MUST carry a chat template, since
|
|
// buildClassifier now rejects routers whose classifier model has none.
|
|
chatMLTpl := map[string]any{
|
|
"chat": "{{.Input -}}\n<|im_start|>assistant\n",
|
|
"chat_message": "<|im_start|>{{ .RoleName }}\n{{ if .Content }}{{ .Content }}{{ end }}<|im_end|>",
|
|
}
|
|
classifierCfg := map[string]any{
|
|
"name": "mock-classifier",
|
|
"backend": "mock-backend",
|
|
"known_usecases": []string{"score"},
|
|
"context_size": 4096,
|
|
"stopwords": []string{"<|im_end|>"},
|
|
"parameters": map[string]any{"model": "mock-classifier.bin"},
|
|
"template": chatMLTpl,
|
|
}
|
|
classifierData, err := yaml.Marshal(classifierCfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, "mock-classifier.yaml"), classifierData, 0644)).To(Succeed())
|
|
|
|
for _, name := range []string{"mock-cand-casual", "mock-cand-code"} {
|
|
candCfg := map[string]any{
|
|
"name": name,
|
|
"backend": "mock-backend",
|
|
"known_usecases": []string{"chat"},
|
|
"parameters": map[string]any{"model": name + ".bin"},
|
|
}
|
|
candData, err := yaml.Marshal(candCfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, name+".yaml"), candData, 0644)).To(Succeed())
|
|
}
|
|
|
|
routerCfg := map[string]any{
|
|
"name": "smart-router",
|
|
"known_usecases": []string{"chat"},
|
|
"router": map[string]any{
|
|
"classifier": "score",
|
|
"classifier_model": "mock-classifier",
|
|
"activation_threshold": 0.40,
|
|
"fallback": "mock-cand-casual",
|
|
"policies": []map[string]any{
|
|
{"label": "casual-chat", "description": "small talk and general conversation"},
|
|
{"label": "code-generation", "description": "writing or debugging code"},
|
|
{"label": "math-reasoning", "description": "arithmetic and word problems"},
|
|
},
|
|
"candidates": []map[string]any{
|
|
{"model": "mock-cand-casual", "labels": []string{"casual-chat"}},
|
|
{"model": "mock-cand-code", "labels": []string{"code-generation", "math-reasoning"}},
|
|
},
|
|
},
|
|
}
|
|
routerData, err := yaml.Marshal(routerCfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, "smart-router.yaml"), routerData, 0644)).To(Succeed())
|
|
|
|
// If REALTIME_TEST_MODEL=realtime-test-pipeline, auto-create a pipeline
|
|
// config from the REALTIME_VAD/STT/LLM/TTS env vars so real-model tests
|
|
// can run without the user having to write a YAML file manually.
|
|
if os.Getenv("REALTIME_TEST_MODEL") == "realtime-test-pipeline" {
|
|
rtVAD := os.Getenv("REALTIME_VAD")
|
|
rtSTT := os.Getenv("REALTIME_STT")
|
|
rtLLM := os.Getenv("REALTIME_LLM")
|
|
rtTTS := os.Getenv("REALTIME_TTS")
|
|
|
|
if rtVAD != "" && rtSTT != "" && rtLLM != "" && rtTTS != "" {
|
|
testPipeline := map[string]any{
|
|
"name": "realtime-test-pipeline",
|
|
"pipeline": map[string]any{
|
|
"vad": rtVAD,
|
|
"transcription": rtSTT,
|
|
"llm": rtLLM,
|
|
"tts": rtTTS,
|
|
},
|
|
}
|
|
data, writeErr := yaml.Marshal(testPipeline)
|
|
Expect(writeErr).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, "realtime-test-pipeline.yaml"), data, 0644)).To(Succeed())
|
|
xlog.Info("created realtime-test-pipeline",
|
|
"vad", rtVAD, "stt", rtSTT, "llm", rtLLM, "tts", rtTTS)
|
|
}
|
|
}
|
|
|
|
// Import model configs from an external directory (e.g. real model YAMLs
|
|
// and weights mounted into a container). Symlinks avoid copying large files.
|
|
// Both files and directories are symlinked — multi-file backends like
|
|
// sherpa-onnx TTS expect their tokens.txt / lexicon.txt sidecars in the
|
|
// same directory as the .onnx, so we need whole-directory imports.
|
|
if rtModels := os.Getenv("REALTIME_MODELS_PATH"); rtModels != "" {
|
|
entries, err := os.ReadDir(rtModels)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
for _, entry := range entries {
|
|
src := filepath.Join(rtModels, entry.Name())
|
|
dst := filepath.Join(modelsPath, entry.Name())
|
|
if _, err := os.Stat(dst); err == nil {
|
|
continue // don't overwrite mock configs
|
|
}
|
|
Expect(os.Symlink(src, dst)).To(Succeed())
|
|
}
|
|
}
|
|
|
|
// Set up system state. When REALTIME_BACKENDS_PATH is set, use it so the
|
|
// application can discover real backend binaries for real-model tests.
|
|
systemOpts := []system.SystemStateOptions{
|
|
system.WithModelPath(modelsPath),
|
|
}
|
|
if realBackends := os.Getenv("REALTIME_BACKENDS_PATH"); realBackends != "" {
|
|
systemOpts = append(systemOpts, system.WithBackendPath(realBackends))
|
|
} else {
|
|
systemOpts = append(systemOpts, system.WithBackendPath(backendPath))
|
|
}
|
|
|
|
// Cloud-proxy backend e2e setup. The cloud-proxy binary lives next
|
|
// to mock-backend and is registered under its canonical "cloud-proxy"
|
|
// name. Fake upstreams come up first so the model YAMLs can encode
|
|
// their URLs at startup time. Build is best-effort — when the binary
|
|
// isn't present, the cloud-proxy specs Skip and the rest of the
|
|
// suite is unaffected.
|
|
cloudProxyCandidates := []string{
|
|
filepath.Join("..", "e2e", "mock-backend", "cloud-proxy"),
|
|
filepath.Join("tests", "e2e", "mock-backend", "cloud-proxy"),
|
|
filepath.Join("..", "..", "tests", "e2e", "mock-backend", "cloud-proxy"),
|
|
}
|
|
for _, p := range cloudProxyCandidates {
|
|
if _, err := os.Stat(p); err == nil {
|
|
cloudProxyPath = p
|
|
break
|
|
}
|
|
}
|
|
if cloudProxyPath != "" {
|
|
Expect(os.Chmod(cloudProxyPath, 0755)).To(Succeed())
|
|
|
|
cpOpenAIUpstream = newFakeOpenAIUpstream()
|
|
cpAnthropicUpstream = newFakeAnthropicUpstream()
|
|
|
|
// API keys are read from env vars — set placeholder values so
|
|
// the cloud-proxy backend's Load() doesn't fail with "unset".
|
|
// The fake upstreams accept any auth header.
|
|
Expect(os.Setenv("CLOUD_PROXY_E2E_OPENAI_KEY", "sk-e2e-openai")).To(Succeed())
|
|
Expect(os.Setenv("CLOUD_PROXY_E2E_ANTHROPIC_KEY", "sk-ant-e2e")).To(Succeed())
|
|
|
|
cloudProxyConfigs := []map[string]any{
|
|
{
|
|
"name": "cp-passthrough-openai",
|
|
"backend": "cloud-proxy",
|
|
"parameters": map[string]any{
|
|
"model": "cloud-proxy-passthrough-openai.bin",
|
|
},
|
|
"proxy": map[string]any{
|
|
"mode": "passthrough",
|
|
"provider": "openai",
|
|
"upstream_url": cpOpenAIUpstream.URL() + "/v1/chat/completions",
|
|
"api_key_env": "CLOUD_PROXY_E2E_OPENAI_KEY",
|
|
},
|
|
},
|
|
{
|
|
"name": "cp-passthrough-anthropic",
|
|
"backend": "cloud-proxy",
|
|
"parameters": map[string]any{
|
|
"model": "cloud-proxy-passthrough-anthropic.bin",
|
|
},
|
|
"proxy": map[string]any{
|
|
"mode": "passthrough",
|
|
"provider": "anthropic",
|
|
"upstream_url": cpAnthropicUpstream.URL() + "/v1/messages",
|
|
"api_key_env": "CLOUD_PROXY_E2E_ANTHROPIC_KEY",
|
|
},
|
|
},
|
|
{
|
|
"name": "cp-translate-openai",
|
|
"backend": "cloud-proxy",
|
|
"parameters": map[string]any{
|
|
"model": "cloud-proxy-translate-openai.bin",
|
|
},
|
|
"proxy": map[string]any{
|
|
"mode": "translate",
|
|
"provider": "openai",
|
|
"upstream_url": cpOpenAIUpstream.URL() + "/v1/chat/completions",
|
|
"api_key_env": "CLOUD_PROXY_E2E_OPENAI_KEY",
|
|
},
|
|
// Wire the in-process pattern detector so the streaming PII
|
|
// filter has something to redact in translate mode. The NER
|
|
// tier needs the privacy-filter model (unavailable in this
|
|
// hermetic suite); the pattern tier runs in-process with no
|
|
// backend load, so it's what exercises the streaming plumbing.
|
|
"pii": map[string]any{
|
|
"detectors": []string{"e2e-secret-filter"},
|
|
},
|
|
},
|
|
{
|
|
// In-process pattern detector: backend "pattern" is resolved by
|
|
// the PII NER resolver directly from this config (no backend is
|
|
// ever loaded). It matches high-entropy anchored secrets;
|
|
// cp-translate-openai references it above via pii.detectors.
|
|
"name": "e2e-secret-filter",
|
|
"backend": "pattern",
|
|
"known_usecases": []string{"token_classify"},
|
|
"pii_detection": map[string]any{
|
|
"default_action": "block",
|
|
"builtins": []string{"anthropic_api_key", "openai_api_key"},
|
|
},
|
|
},
|
|
{
|
|
"name": "cp-translate-anthropic",
|
|
"backend": "cloud-proxy",
|
|
"parameters": map[string]any{
|
|
"model": "cloud-proxy-translate-anthropic.bin",
|
|
},
|
|
"proxy": map[string]any{
|
|
"mode": "translate",
|
|
"provider": "anthropic",
|
|
"upstream_url": cpAnthropicUpstream.URL() + "/v1/messages",
|
|
"api_key_env": "CLOUD_PROXY_E2E_ANTHROPIC_KEY",
|
|
},
|
|
},
|
|
}
|
|
for _, cfg := range cloudProxyConfigs {
|
|
data, err := yaml.Marshal(cfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, cfg["name"].(string)+".yaml"), data, 0644)).To(Succeed())
|
|
}
|
|
}
|
|
|
|
// Live PII NER tier. When PII_NER_MODEL_GGUF points at a downloaded
|
|
// privacy-filter GGUF, register two detector models that drive the real
|
|
// gRPC TokenClassify path on the privacy-filter backend (discovered via
|
|
// REALTIME_BACKENDS_PATH). Two models so we can exercise both policy
|
|
// outcomes against the same weights: mask (redact) and block (reject).
|
|
// NOTE: no pii_detection.builtins/patterns here — that would flip the
|
|
// detector to the in-process regex tier instead of the GGUF NER tier.
|
|
if gguf := os.Getenv("PII_NER_MODEL_GGUF"); gguf != "" {
|
|
piiNERModel = "privacy-filter-ner"
|
|
piiNERBlockModel = "privacy-filter-ner-block"
|
|
nerModelConfig := func(name, defaultAction string) map[string]any {
|
|
return map[string]any{
|
|
"name": name,
|
|
"backend": "privacy-filter",
|
|
"embeddings": true, // required: TOKEN_CLS pooling loads via the embeddings flag
|
|
"known_usecases": []string{"token_classify"},
|
|
"parameters": map[string]any{"model": gguf},
|
|
"pii_detection": map[string]any{
|
|
"min_score": 0.5,
|
|
"default_action": defaultAction,
|
|
},
|
|
}
|
|
}
|
|
for _, cfg := range []map[string]any{
|
|
nerModelConfig(piiNERModel, "mask"),
|
|
nerModelConfig(piiNERBlockModel, "block"),
|
|
} {
|
|
data, err := yaml.Marshal(cfg)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
Expect(os.WriteFile(filepath.Join(modelsPath, cfg["name"].(string)+".yaml"), data, 0644)).To(Succeed())
|
|
}
|
|
xlog.Info("wired live PII NER models", "gguf", gguf, "models", []string{piiNERModel, piiNERBlockModel})
|
|
}
|
|
|
|
systemState, err := system.GetSystemState(systemOpts...)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
|
|
// Create application
|
|
appCtx, appCancel = context.WithCancel(context.Background())
|
|
|
|
// Create application instance (GeneratedContentDir so sound-generation/TTS can write files the handler sends)
|
|
generatedDir := filepath.Join(tmpDir, "generated")
|
|
Expect(os.MkdirAll(generatedDir, 0750)).To(Succeed())
|
|
localAIApp, err = localaiapp.New(
|
|
config.WithContext(appCtx),
|
|
config.WithSystemState(systemState),
|
|
config.WithDebug(true),
|
|
config.WithGeneratedContentDir(generatedDir),
|
|
)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
|
|
// Register mock backend (always available for non-realtime tests).
|
|
localAIApp.ModelLoader().SetExternalBackend("mock-backend", mockBackendPath)
|
|
localAIApp.ModelLoader().SetExternalBackend("opus", mockBackendPath)
|
|
if cloudProxyPath != "" {
|
|
localAIApp.ModelLoader().SetExternalBackend("cloud-proxy", cloudProxyPath)
|
|
}
|
|
|
|
// Create HTTP app
|
|
app, err = httpapi.API(localAIApp)
|
|
Expect(err).ToNot(HaveOccurred())
|
|
|
|
// Get free port
|
|
port, err := freeport.GetFreePort()
|
|
Expect(err).ToNot(HaveOccurred())
|
|
apiPort = port
|
|
apiURL = fmt.Sprintf("http://127.0.0.1:%d/v1", apiPort)
|
|
// Anthropic SDK appends /v1/messages to base URL; use base without /v1 so requests go to /v1/messages
|
|
anthropicBaseURL = fmt.Sprintf("http://127.0.0.1:%d", apiPort)
|
|
// Ollama client uses base URL directly
|
|
ollamaBaseURL = fmt.Sprintf("http://127.0.0.1:%d", apiPort)
|
|
|
|
// Start server in goroutine
|
|
go func() {
|
|
if err := app.Start(fmt.Sprintf("127.0.0.1:%d", apiPort)); err != nil && err != http.ErrServerClosed {
|
|
xlog.Error("server error", "error", err)
|
|
}
|
|
}()
|
|
|
|
// Wait for server to be ready
|
|
client = openai.NewClient(option.WithBaseURL(apiURL))
|
|
|
|
Eventually(func() error {
|
|
_, err := client.Models.List(context.TODO())
|
|
return err
|
|
}, "2m").ShouldNot(HaveOccurred())
|
|
})
|
|
|
|
var _ = AfterSuite(func() {
|
|
// Synchronous shutdown — the context-cancel goroutine in application.New
|
|
// runs the same cleanup asynchronously, which races test-binary exit and
|
|
// orphans spawned mock-backend children to init.
|
|
if localAIApp != nil {
|
|
if err := localAIApp.Shutdown(); err != nil {
|
|
xlog.Error("error shutting down application", "error", err)
|
|
}
|
|
}
|
|
if appCancel != nil {
|
|
appCancel()
|
|
}
|
|
if app != nil {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
Expect(app.Shutdown(ctx)).To(Succeed())
|
|
}
|
|
if mcpServerShutdown != nil {
|
|
mcpServerShutdown()
|
|
}
|
|
if cpOpenAIUpstream != nil {
|
|
cpOpenAIUpstream.Close()
|
|
}
|
|
if cpAnthropicUpstream != nil {
|
|
cpAnthropicUpstream.Close()
|
|
}
|
|
if tmpDir != "" {
|
|
os.RemoveAll(tmpDir)
|
|
}
|
|
})
|
|
|
|
func TestLocalAI(t *testing.T) {
|
|
RegisterFailHandler(Fail)
|
|
RunSpecs(t, "LocalAI E2E test suite")
|
|
}
|