From 4a2a6180a2eec8ef48fdc501dfce79f1588a029b Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Sun, 27 Sep 2026 00:32:13 +0000 Subject: [PATCH] test(localai-proxy): proxy APIs and realtime stages end to end The e2e suite now registers the localai-proxy binary and points proxy models back at the test server itself, so a request leaves LocalAI through the backend, returns over REST and is answered by a mock model. Chat, embeddings, TTS and transcription through the proxy return the upstream model's answer; a chain whose proxy target's upstream model fails to load serves from the local target; and a realtime pipeline whose LLM stage is a chain on a remote target completes a turn, then switches to the local target with a localai.model.failover trip event when a gate in front of the upstream starts answering 503. The docs describe the localai-proxy backend next to cloud-proxy and add a per-stage remote LocalAI example to the failover page. Assisted-by: Claude:claude-opus-5-5 Signed-off-by: Ettore Di Giacinto --- docs/content/features/model-failover.md | 75 ++++++++- docs/content/operations/cloud-proxy.md | 90 ++++++++++ tests/e2e/e2e_failover_test.go | 23 +-- tests/e2e/e2e_localai_proxy_test.go | 215 ++++++++++++++++++++++++ tests/e2e/e2e_suite_test.go | 21 +++ tests/e2e/realtime_ws_test.go | 63 +++++++ 6 files changed, 461 insertions(+), 26 deletions(-) create mode 100644 tests/e2e/e2e_localai_proxy_test.go diff --git a/docs/content/features/model-failover.md b/docs/content/features/model-failover.md index 32e91af96..8f62a0314 100644 --- a/docs/content/features/model-failover.md +++ b/docs/content/features/model-failover.md @@ -20,7 +20,7 @@ provider, and to fall back to a local model when the remote one is down. name: assistant-llm failover: targets: - - model: argus-llm # for example a cloud-proxy model + - model: argus-llm # for example a localai-proxy or cloud-proxy model - model: gemma-local warm: true # keep it loaded ``` @@ -50,9 +50,9 @@ Rules: - A chain cannot also set `alias` or `backend`. - Responses name the chain as the model. The `X-LocalAI-Served-Model` header names the target that served the request. -- A remote (`cloud-proxy`) target receives its own model name, never the chain - name: `proxy.upstream_model`, or the target name when `upstream_model` is - empty. The health check looks for the same name. +- A remote (`localai-proxy` or `cloud-proxy`) target receives its own model + name, never the chain name: `proxy.upstream_model`, or the target name when + `upstream_model` is empty. The health check looks for the same name. ## How the target is chosen @@ -89,7 +89,7 @@ targets were down. | Target | Regular check | Check before moving back | |---|---|---| -| Remote (`cloud-proxy`) | `GET /v1/models` on the upstream lists the model | one small real request, for example a 1-token completion | +| Remote (`localai-proxy`, `cloud-proxy`) | `GET /v1/models` on the upstream lists the model | one small real request, for example a 1-token completion | | Local, `warm: true` | the backend answers a health check. A check never loads the model: while it is not loaded, the check passes and real requests judge it | one small real request. While the model is not loaded, the target is used again after `min_dwell` | | Local, not warm | none: judged only by real requests; it is never loaded only to check it | none: the target is used again after `min_dwell` | @@ -107,8 +107,9 @@ count toward the active backend limit (`--max-active-backends`) like any pinned model: LocalAI never evicts them to make room, and if they fill the limit, a new model still loads rather than being blocked. -`warm` applies only to local targets. On a remote (`cloud-proxy`) target it -has no effect, and LocalAI logs a warning when it loads the chain. +`warm` applies only to local targets. On a remote (`localai-proxy` or +`cloud-proxy`) target it has no effect, and LocalAI logs a warning when it +loads the chain. ## Realtime pipelines @@ -136,6 +137,66 @@ it starts (`reason: initial`) and each time a chain switches: "from":"argus-llm","to":"gemma-local","state":"fallback","reason":"trip"} ``` +### Example: stages on a remote LocalAI + +This pipeline runs its transcription, LLM and TTS stages on a remote LocalAI +(`argus`) through [`localai-proxy`]({{% relref "operations/cloud-proxy" %}}) +models, and uses local models when the remote instance is down. Each stage has +its own chain, so one stage can fail over while the others stay remote. + +```yaml +# Remote targets: each one names the model on the upstream LocalAI. +name: argus-stt +backend: localai-proxy +known_usecases: [transcript] +options: + - realtime_pipeline:asr-pipeline # upstream pipeline for live transcription +proxy: + upstream_url: http://argus.lan:8080 + upstream_model: parakeet +--- +name: argus-llm +backend: localai-proxy +known_usecases: [chat] +proxy: + upstream_url: http://argus.lan:8080 + upstream_model: gemma-3-12b +--- +name: argus-tts +backend: localai-proxy +known_usecases: [tts] +proxy: + upstream_url: http://argus.lan:8080 + upstream_model: kokoro +--- +# One chain per stage, remote first, local second. +name: stt-chain +failover: + targets: [{model: argus-stt}, {model: whisper-local, warm: true}] +--- +name: llm-chain +failover: + targets: [{model: argus-llm}, {model: gemma-local, warm: true}] +--- +name: tts-chain +failover: + targets: [{model: argus-tts}, {model: piper-local}] +--- +name: assistant +pipeline: + vad: silero-vad + transcription: stt-chain + llm: llm-chain + tts: tts-chain +``` + +The example shows the configs as one YAML stream; put each config in its own +file in the models directory. When `argus` stops +answering, the next call of each stage fails over to the local model and the +session receives a `localai.model.failover` event for that stage. A remote +target that does not support a call (it returns `Unimplemented`) is skipped for +that call and is not marked down. + Limits: - After a `session.update` that changes the pipeline, `localai.model.failover` diff --git a/docs/content/operations/cloud-proxy.md b/docs/content/operations/cloud-proxy.md index bd1d806d4..ec178e2f8 100644 --- a/docs/content/operations/cloud-proxy.md +++ b/docs/content/operations/cloud-proxy.md @@ -242,6 +242,96 @@ ACLs, and the cloud-proxy fork all run against the resolved target. See [Middleware: PII filtering and intelligent routing]({{< relref "middleware.md" >}}) for the full router and PII-filter reference. +## Proxying to another LocalAI (`localai-proxy`) + +`cloud-proxy` forwards chat and Messages requests only. To serve a model from +another LocalAI instance for every API it has, use `backend: localai-proxy`. +The backend receives the request from the local pipeline like any other +backend and sends it to the REST API of the upstream LocalAI. Because it is a +normal backend, a `localai-proxy` model can be a stage of a realtime pipeline +or a target of a [failover chain]({{% relref "features/model-failover" %}}). + +```yaml +name: remote-llm +backend: localai-proxy +known_usecases: [chat] +proxy: + # Base URL of the upstream LocalAI. Do not add /v1 or an endpoint path: + # the backend adds the path for each API. + upstream_url: https://argus.lan:8080 + # The model name on the upstream. When empty, the name of this config. + upstream_model: gemma-3-12b + # Optional. The upstream API key, from an environment variable + # (or api_key_file). Sent as "Authorization: Bearer ". + api_key_env: ARGUS_API_KEY + # Optional. Time limit for each non-streaming request. Streams have no limit. + request_timeout_seconds: 120 +``` + +A model that does live transcription in a realtime pipeline also names a +realtime pipeline on the upstream. The backend opens a transcription session +on the upstream `/v1/realtime` endpoint with that pipeline: + +```yaml +name: remote-stt +backend: localai-proxy +known_usecases: [transcript] +options: + - realtime_pipeline:asr-pipeline +proxy: + upstream_url: https://argus.lan:8080 + upstream_model: parakeet +``` + +Set `known_usecases` on every `localai-proxy` model. Failover uses it to match +targets, and LocalAI cannot guess the usecases of a remote model. For a chat +model, `known_usecases: [chat]` has one more effect: LocalAI sends the chat +messages to the upstream `/v1/chat/completions` endpoint, and the upstream +applies its own chat template, tool parsing and reasoning parsing. Without +`chat`, or when the config has its own templates, LocalAI renders the prompt +locally and sends it to `/v1/completions`. `proxy.mode` and `proxy.provider` +have no effect on this backend. + +Supported APIs: + +- Text: chat and completions (also streamed), embeddings, rerank, tokenize, + detokenize, score. +- Audio: TTS (also streamed), sound generation, transcription (also streamed), + live transcription (with `realtime_pipeline`), diarization, VAD, sound + classification, audio transformations. +- Image, video and 3D: image generation, upscaling, video generation, 3D + generation and animation. The backend downloads the files that the upstream + generates. +- Vision: object detection, depth, face verification and analysis, voice + verification, analysis and embeddings. +- Stores: set, get, delete, find. + +Methods that have no REST API on the upstream return the gRPC error +`Unimplemented` ("localai-proxy: has no upstream counterpart"): +audio encoding and decoding, audio-to-audio streams, token classification +(PII NER), model metadata, fine-tuning, quantization and model export. A +failover chain skips a target that returns `Unimplemented` and tries the next +target, but does not mark the target down. + +Errors from the upstream: a 5xx response or a connection failure becomes +`Unavailable`, and a failover chain marks the target down. A 4xx response +becomes `InvalidArgument`, and LocalAI returns it to the client without a +retry. + +Known limits: + +- Voice-profile paths pass through unresolved. When LocalAI resolves a TTS + voice to a local file (for example a voice clone reference), the backend + sends that path to the upstream, where it does not exist. Use voices that + the upstream knows by name. +- Depth exports are not supported. The upstream writes them to its own disk, + so a depth request with exports or a destination file returns + `Unimplemented`. Depth maps and points without exports work. +- The REST transcription API has no end-of-utterance (`eou`) flag, so + transcriptions through the proxy never set it. Live transcription through + `realtime_pipeline` sets `eou` at the end of each utterance. +- Sound generation from a source audio file is not supported. + ## Limitations - **Passthrough does no wire-shape translation.** Use `mode: translate` (with diff --git a/tests/e2e/e2e_failover_test.go b/tests/e2e/e2e_failover_test.go index b8e1632d2..22a4226f6 100644 --- a/tests/e2e/e2e_failover_test.go +++ b/tests/e2e/e2e_failover_test.go @@ -6,23 +6,13 @@ import ( "io" "mime/multipart" "net/http" - "os" - "path/filepath" "time" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" - "gopkg.in/yaml.v3" ) var _ = Describe("Failover chains", Label("failover"), func() { - postJSON := func(path string, body map[string]any) *http.Response { - b, err := json.Marshal(body) - Expect(err).ToNot(HaveOccurred()) - resp, err := http.Post(apiURL+path, "application/json", bytes.NewReader(b)) - Expect(err).ToNot(HaveOccurred()) - return resp - } expectServedByMock := func(resp *http.Response) { defer func() { _ = resp.Body.Close() }() body, _ := io.ReadAll(resp.Body) @@ -34,7 +24,7 @@ var _ = Describe("Failover chains", Label("failover"), func() { // The entry name is the chain suffix: chain- is written by the suite. DescribeTable("retries every endpoint family on the next target", func(path string, body func(model string) map[string]any) { - expectServedByMock(postJSON(path, body("chain-"+CurrentSpecReport().LeafNodeText))) + expectServedByMock(postJSONTo(path, body("chain-"+CurrentSpecReport().LeafNodeText))) }, Entry("chat", "/chat/completions", func(m string) map[string]any { return map[string]any{"model": m, "messages": []map[string]string{{"role": "user", "content": "hi"}}} @@ -99,7 +89,7 @@ var _ = Describe("Failover chains", Label("failover"), func() { }) up2.SetScript(chatReply) - resp := postJSON("/chat/completions", map[string]any{"model": "chain-remote", "messages": []map[string]string{{"role": "user", "content": "hi"}}}) + resp := postJSONTo("/chat/completions", map[string]any{"model": "chain-remote", "messages": []map[string]string{{"role": "user", "content": "hi"}}}) defer func() { _ = resp.Body.Close() }() body, _ := io.ReadAll(resp.Body) Expect(resp.StatusCode).To(Equal(200), string(body)) @@ -116,7 +106,7 @@ var _ = Describe("Failover chains", Label("failover"), func() { Eventually(func() string { return chainActive("chain-remote") }, 30*time.Second, 500*time.Millisecond). Should(Equal("up-1")) - resp2 := postJSON("/chat/completions", map[string]any{"model": "chain-remote", "messages": []map[string]string{{"role": "user", "content": "hi"}}}) + resp2 := postJSONTo("/chat/completions", map[string]any{"model": "chain-remote", "messages": []map[string]string{{"role": "user", "content": "hi"}}}) defer func() { _ = resp2.Body.Close() }() Expect(resp2.StatusCode).To(Equal(200)) Expect(resp2.Header.Get("X-LocalAI-Served-Model")).To(Equal("up-1")) @@ -178,12 +168,7 @@ func registerFailoverRemoteModels(url1, url2 string) { "recovery": map[string]any{"probes": 2, "min_dwell": "2s"}, }, } - for _, cfg := range []map[string]any{proxyModel("up-1", url1), proxyModel("up-2", url2), chain} { - data, err := yaml.Marshal(cfg) - Expect(err).ToNot(HaveOccurred()) - Expect(os.WriteFile(filepath.Join(modelsPath, cfg["name"].(string)+".yaml"), data, 0644)).To(Succeed()) - } - Expect(localAIApp.ModelConfigLoader().LoadModelConfigsFromPath(modelsPath)).To(Succeed()) + registerModelConfigs(proxyModel("up-1", url1), proxyModel("up-2", url2), chain) Eventually(func() string { return chainActive("chain-remote") }, 10*time.Second, 200*time.Millisecond). Should(Equal("up-1")) } diff --git a/tests/e2e/e2e_localai_proxy_test.go b/tests/e2e/e2e_localai_proxy_test.go new file mode 100644 index 000000000..fc86f612e --- /dev/null +++ b/tests/e2e/e2e_localai_proxy_test.go @@ -0,0 +1,215 @@ +package e2e_test + +import ( + "bytes" + "encoding/json" + "io" + "mime/multipart" + "net/http" + "net/http/httptest" + "net/http/httputil" + "net/url" + "os" + "path/filepath" + "sync/atomic" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "gopkg.in/yaml.v3" +) + +// The localai-proxy specs point the backend at this same test server: a +// request to an lp-* model leaves LocalAI through the localai-proxy process, +// comes back in over REST and is answered by the mock model it names, so the +// whole round trip (core -> gRPC -> REST upstream -> gRPC reply) is real. +var _ = Describe("localai-proxy backend", Label("failover"), Ordered, func() { + BeforeAll(func() { + if localAIProxyPath == "" { + Skip("localai-proxy backend binary not built (make build-localai-proxy-backend)") + } + registerModelConfigs( + localAIProxyModel("lp-chat", anthropicBaseURL, "mock-model", "chat"), + localAIProxyModel("lp-embeddings", anthropicBaseURL, "mock-model", "embeddings"), + localAIProxyModel("lp-tts", anthropicBaseURL, "mock-model", "tts"), + localAIProxyModel("lp-transcription", anthropicBaseURL, "mock-model", "transcript"), + // The upstream serves this target from a model whose load always + // fails, so every request through it errors with a 5xx. + localAIProxyModel("lp-broken", anthropicBaseURL, "fail-chat", "chat"), + map[string]any{ + "name": "chain-lp", + "failover": map[string]any{ + "targets": []map[string]any{{"model": "lp-broken"}, {"model": "mock-model"}}, + }, + }, + ) + Eventually(func() string { return chainActive("chain-lp") }, 10*time.Second, 200*time.Millisecond). + Should(Equal("lp-broken")) + }) + + // sameBody posts the same request to the proxied model and to the model + // the upstream serves it from, and returns both bodies: the proxy answered + // with the upstream's answer when they match. + sameBody := func(path string, body func(model string) map[string]any, proxied, upstream string) (string, string) { + get := func(model string) string { + resp := postJSONTo(path, body(model)) + defer func() { _ = resp.Body.Close() }() + b, err := io.ReadAll(resp.Body) + Expect(err).ToNot(HaveOccurred()) + Expect(resp.StatusCode).To(Equal(http.StatusOK), "%s: %s", model, b) + return string(b) + } + return get(proxied), get(upstream) + } + + It("answers chat with the upstream model's reply", func() { + chat := func(m string) map[string]any { + return map[string]any{"model": m, "messages": []map[string]string{{"role": "user", "content": "hi"}}} + } + got, want := sameBody("/chat/completions", chat, "lp-chat", "mock-model") + Expect(chatContent(got)).ToNot(BeEmpty()) + Expect(chatContent(got)).To(Equal(chatContent(want))) + }) + + It("answers embeddings with the upstream model's vector", func() { + embed := func(m string) map[string]any { return map[string]any{"model": m, "input": "hello"} } + got, want := sameBody("/embeddings", embed, "lp-embeddings", "mock-model") + Expect(embeddingVector(got)).ToNot(BeEmpty()) + Expect(embeddingVector(got)).To(Equal(embeddingVector(want))) + }) + + It("answers TTS with the upstream model's audio", func() { + speech := func(m string) map[string]any { return map[string]any{"model": m, "input": "hello", "voice": "default"} } + got, want := sameBody("/audio/speech", speech, "lp-tts", "mock-model") + Expect(got).To(HavePrefix("RIFF")) + Expect(got).To(Equal(want)) + }) + + It("answers transcription with the upstream model's text", func() { + transcribe := func(model string) string { + var body bytes.Buffer + mw := multipart.NewWriter(&body) + Expect(mw.WriteField("model", model)).To(Succeed()) + fw, err := mw.CreateFormFile("file", "a.wav") + Expect(err).ToNot(HaveOccurred()) + _, err = fw.Write(wavFromPCM(make([]byte, 6400), 16000)) + Expect(err).ToNot(HaveOccurred()) + Expect(mw.Close()).To(Succeed()) + resp, err := http.Post(apiURL+"/audio/transcriptions", mw.FormDataContentType(), &body) + Expect(err).ToNot(HaveOccurred()) + defer func() { _ = resp.Body.Close() }() + b, _ := io.ReadAll(resp.Body) + Expect(resp.StatusCode).To(Equal(http.StatusOK), "%s: %s", model, b) + var out struct { + Text string `json:"text"` + } + Expect(json.Unmarshal(b, &out)).To(Succeed(), string(b)) + return out.Text + } + got := transcribe("lp-transcription") + Expect(got).To(HavePrefix("transcribed:")) + Expect(got).To(Equal(transcribe("mock-model"))) + }) + + It("fails over from a proxy target whose upstream model fails to the local target", func() { + resp := postJSONTo("/chat/completions", map[string]any{ + "model": "chain-lp", + "messages": []map[string]string{{"role": "user", "content": "hi"}}, + }) + defer func() { _ = resp.Body.Close() }() + b, _ := io.ReadAll(resp.Body) + Expect(resp.StatusCode).To(Equal(http.StatusOK), string(b)) + Expect(resp.Header.Get("X-LocalAI-Served-Model")).To(Equal("mock-model")) + Expect(resp.Header.Get("X-LocalAI-Failover")).To(Equal("fallback")) + Expect(chainActive("chain-lp")).To(Equal("mock-model")) + }) +}) + +// postJSONTo posts body as JSON to an /v1 path of the test server. +func postJSONTo(path string, body map[string]any) *http.Response { + b, err := json.Marshal(body) + ExpectWithOffset(1, err).ToNot(HaveOccurred()) + resp, err := http.Post(apiURL+path, "application/json", bytes.NewReader(b)) + ExpectWithOffset(1, err).ToNot(HaveOccurred()) + return resp +} + +func chatContent(body string) string { + var out struct { + Choices []struct { + Message struct { + Content string `json:"content"` + } `json:"message"` + } `json:"choices"` + } + ExpectWithOffset(1, json.Unmarshal([]byte(body), &out)).To(Succeed(), body) + ExpectWithOffset(1, out.Choices).ToNot(BeEmpty(), body) + return out.Choices[0].Message.Content +} + +func embeddingVector(body string) []float32 { + var out struct { + Data []struct { + Embedding []float32 `json:"embedding"` + } `json:"data"` + } + ExpectWithOffset(1, json.Unmarshal([]byte(body), &out)).To(Succeed(), body) + ExpectWithOffset(1, out.Data).ToNot(BeEmpty(), body) + return out.Data[0].Embedding +} + +// localAIProxyModel is a localai-proxy config serving upstreamModel from the +// LocalAI at baseURL. known_usecases is what failover matches targets on, and +// "chat" also makes the proxy send structured messages upstream. +func localAIProxyModel(name, baseURL, upstreamModel string, usecases ...string) map[string]any { + return map[string]any{ + "name": name, + "backend": "localai-proxy", + "known_usecases": usecases, + "parameters": map[string]any{"model": name + ".bin"}, + "proxy": map[string]any{ + "upstream_url": baseURL, + "upstream_model": upstreamModel, + }, + } +} + +// registerModelConfigs writes model YAMLs after startup and has the loader +// re-read the models directory, for configs that embed runtime URLs. The +// failover manager picks new chains up on its next tick. +func registerModelConfigs(cfgs ...map[string]any) { + for _, cfg := range cfgs { + data, err := yaml.Marshal(cfg) + ExpectWithOffset(1, err).ToNot(HaveOccurred()) + ExpectWithOffset(1, os.WriteFile(filepath.Join(modelsPath, cfg["name"].(string)+".yaml"), data, 0644)).To(Succeed()) + } + ExpectWithOffset(1, localAIApp.ModelConfigLoader().LoadModelConfigsFromPath(modelsPath)).To(Succeed()) +} + +// upstreamGate is a reverse proxy in front of the test server that a spec +// can take down: while down it answers 503, as an upstream LocalAI with no +// healthy backend would, so a localai-proxy target behind it starts failing +// without restarting its backend process. +type upstreamGate struct { + srv *httptest.Server + down atomic.Bool +} + +func newUpstreamGate(target string) *upstreamGate { + u, err := url.Parse(target) + ExpectWithOffset(1, err).ToNot(HaveOccurred()) + g := &upstreamGate{} + rp := httputil.NewSingleHostReverseProxy(u) + g.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if g.down.Load() { + http.Error(w, `{"error":{"message":"no healthy backend"}}`, http.StatusServiceUnavailable) + return + } + rp.ServeHTTP(w, r) + })) + return g +} + +func (g *upstreamGate) URL() string { return g.srv.URL } +func (g *upstreamGate) Close() { g.srv.Close() } +func (g *upstreamGate) SetDown(down bool) { g.down.Store(down) } diff --git a/tests/e2e/e2e_suite_test.go b/tests/e2e/e2e_suite_test.go index e7bb844d5..0d1f51608 100644 --- a/tests/e2e/e2e_suite_test.go +++ b/tests/e2e/e2e_suite_test.go @@ -39,6 +39,7 @@ var ( apiURL string mockBackendPath string cloudProxyPath string + localAIProxyPath string mcpServerURL string mcpServerShutdown func() localAIApp *localaiapp.Application @@ -646,6 +647,23 @@ var _ = BeforeSuite(func() { } } + // localai-proxy backend: its models point back at this server, whose URL + // exists only once it listens, so the specs register them at runtime. + // Like cloud-proxy, a missing binary makes those specs Skip. + for _, p := range []string{ + filepath.Join("..", "e2e", "mock-backend", "localai-proxy"), + filepath.Join("tests", "e2e", "mock-backend", "localai-proxy"), + filepath.Join("..", "..", "tests", "e2e", "mock-backend", "localai-proxy"), + } { + if _, err := os.Stat(p); err == nil { + localAIProxyPath = p + break + } + } + if localAIProxyPath != "" { + Expect(os.Chmod(localAIProxyPath, 0755)).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 @@ -703,6 +721,9 @@ var _ = BeforeSuite(func() { if cloudProxyPath != "" { localAIApp.ModelLoader().SetExternalBackend("cloud-proxy", cloudProxyPath) } + if localAIProxyPath != "" { + localAIApp.ModelLoader().SetExternalBackend("localai-proxy", localAIProxyPath) + } // Create HTTP app app, err = httpapi.API(localAIApp) diff --git a/tests/e2e/realtime_ws_test.go b/tests/e2e/realtime_ws_test.go index 68181ef35..5556a608f 100644 --- a/tests/e2e/realtime_ws_test.go +++ b/tests/e2e/realtime_ws_test.go @@ -273,6 +273,69 @@ var _ = Describe("Realtime WebSocket API", Label("Realtime"), func() { Expect(retrieveItem(conn, firstReplyID)).To(HaveKeyWithValue("id", firstReplyID)) }) + It("serves the LLM stage from a remote LocalAI and switches to the local target when it fails", func() { + if localAIProxyPath == "" { + Skip("localai-proxy backend binary not built (make build-localai-proxy-backend)") + } + // The remote target reaches this server through a gate the spec + // can take down, as if the remote LocalAI lost its backends. + gate := newUpstreamGate(anthropicBaseURL) + DeferCleanup(gate.Close) + registerModelConfigs( + localAIProxyModel("lp-rt-llm", gate.URL(), "mock-llm", "chat"), + map[string]any{ + "name": "chain-rt-lp", + "failover": map[string]any{ + "targets": []map[string]any{{"model": "lp-rt-llm"}, {"model": "mock-llm"}}, + }, + }, + map[string]any{ + "name": "rt-lp", + "pipeline": map[string]any{ + "vad": "mock-vad", + "transcription": "mock-stt", + "llm": "chain-rt-lp", + "tts": "mock-tts", + "disable_warmup": true, + }, + }, + ) + Eventually(func() string { return chainActive("chain-rt-lp") }, 10*time.Second, 200*time.Millisecond). + Should(Equal("lp-rt-llm")) + + conn := connectWS("rt-lp") + defer conn.Close() + + Expect(readServerEvent(conn, 30*time.Second)["type"]).To(Equal("session.created")) + initial := drainUntil(conn, "localai.model.failover", 10*time.Second) + Expect(initial).To(HaveKeyWithValue("stage", "llm")) + Expect(initial).To(HaveKeyWithValue("chain", "chain-rt-lp")) + Expect(initial).To(HaveKeyWithValue("reason", "initial")) + Expect(initial).To(HaveKeyWithValue("to", "lp-rt-llm")) + + sendClientEvent(conn, disableVADEvent()) + drainUntil(conn, "session.updated", 10*time.Second) + + // The first turn goes through the proxy to the remote mock-llm. + _, done, failovers := userTurn(conn, "Hello, how are you?") + Expect(failovers).To(BeEmpty()) + resp, _ := done["response"].(map[string]any) + Expect(resp).To(HaveKeyWithValue("status", "completed")) + Expect(chainActive("chain-rt-lp")).To(Equal("lp-rt-llm")) + + gate.SetDown(true) + _, done, failovers = userTurn(conn, "And now?") + Expect(failovers).To(ContainElement(And( + HaveKeyWithValue("stage", "llm"), + HaveKeyWithValue("from", "lp-rt-llm"), + HaveKeyWithValue("to", "mock-llm"), + HaveKeyWithValue("reason", "trip"), + ))) + resp, _ = done["response"].(map[string]any) + Expect(resp).To(HaveKeyWithValue("status", "completed")) + Expect(chainActive("chain-rt-lp")).To(Equal("mock-llm")) + }) + It("starts the session on the next target when the active one fails to warm up", func() { conn := connectWS("rt-failover-warm") defer func() { _ = conn.Close() }()