mirror of
https://github.com/mudler/LocalAI.git
synced 2026-09-29 09:35:02 -04:00
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 <mudler@localai.io>
This commit is contained in:
1 parent
e66d87b887
commit
4a2a6180a2
6 files changed
+461
-26
No files matched your search
@@ -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`
|
||||
|
||||
@@ -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 <key>".
|
||||
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: <method> 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
|
||||
|
||||
@@ -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-<name> 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"))
|
||||
}
|
||||
@@ -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) }
|
||||
@@ -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)
|
||||
|
||||
@@ -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() }()
|
||||
|
||||
Reference in new issue
Block a user