Commit 1242be66f for llama.cpp
commit 1242be66f65d5d8a3d996ba56d6de106ced98cf6
Author: Xuan-Son Nguyen <son@huggingface.co>
Date: Wed Oct 21 00:04:15 2026 +0200
server: add realtime transcription API (#30636)
* server: add realtime transcription API
* drop transcription_sessions
* proxy ws
* add docs
* server: let the API key middleware guard the websocket
httplib runs the pre-routing handler before the websocket upgrade, so
the API key middleware answers 401 at handshake like any other route.
This covers the router proxy, whose children run keyless, and accepts
the same key forms as HTTP (Bearer, raw key, X-Api-Key).
* server: keep realtime slots out of the prompt loop
A realtime slot waiting for audio is GENERATING with nothing to decode,
so it fell through the prompt loop and took slot_batched. A request on
a higher slot then failed can_batch_with on the task type and starved
for the whole session. Realtime slots have no prompt: the loop sets
them up once and always leaves them.
* server: start the realtime session once its slot is live
The audio stream was created in the prompt loop, which skips a slot
that cannot batch with the one generating, and session.created went out
before the task was posted: early appends failed and the reader quit,
leaving the socket open and the slot held until the HTTP timeout.
The slot now sets up its stream at launch and signals it with the begin
result, session.created waits for it, and a reader that stops closes
the socket so the session ends with it.
* server: name the committed item in input_audio_buffer.committed
Per the Realtime spec, item_id is the item made from the committed audio
and previous_item_id the one committed before it, null for the first.
committed now comes before the transcript that completes that item, and
both go out with the item switch under one lock, so a delta cannot land
between them.
* server: use the GA transcription session shape
The session object follows the GA Realtime schema: type transcription,
with the input format, transcription model, turn detection and noise
reduction under audio.input. The rate stays the model sample rate, which
is what the client must send for now.
---------
Co-authored-by: Pascal <admin@serveurperso.com>
diff --git a/tools/server/README.md b/tools/server/README.md
index 8d9f53569..2977fcd4d 100644
--- a/tools/server/README.md
+++ b/tools/server/README.md
@@ -1491,6 +1491,31 @@ Acts on an in-flight completion identified by its `id` (the `id` field streamed
Returns a JSON object with a boolean `success` field, and an optional `message` field describing the reason when `success` is `false`.
+### WebSocket `/v1/realtime`: Realtime transcription
+
+Transcribes streamed audio, for models with a streaming audio encoder. Compatible with OAI `/v1/realtime` schema.
+
+Audio must be PCM16, mono, at the model sample rate. The sample rate is returned in `session.created` as `audio.input.format.rate`. Each `input_audio_buffer.append` carries a base64-encoded chunk of PCM16 samples.
+
+*Client events:*
+
+- `session.update`: accepted for compatibility, the server replies with `session.updated`
+- `input_audio_buffer.append`: `audio` holds base64 PCM16 samples
+- `input_audio_buffer.commit`: flushes the audio received so far, so the words still held back by the model's delay are produced. The server replies with `input_audio_buffer.committed` and starts a new item. With `"final": true`, the input ends instead and the last item is completed once the stream drains.
+- `input_audio_buffer.clear`: accepted for compatibility, audio already sent cannot be discarded
+
+Closing the websocket also ends the input.
+
+*Server events:*
+
+- `session.created` and `session.updated`: carry the `session` object
+- `conversation.item.input_audio_transcription.delta`: `delta` holds the new text for `item_id`
+- `conversation.item.input_audio_transcription.completed`: `transcript` holds the full text of `item_id`
+- `input_audio_buffer.committed`: `item_id` of the committed item, `previous_item_id` of the item committed before it (`null` for the first one)
+- `error`: `error.message` describes the problem
+
+Note that words held back by the model's delay at the time of a commit are attributed to the next item.
+
### POST `/v1/responses`: OpenAI-compatible Responses API
*Options:*
@@ -1992,6 +2017,8 @@ For **GET** endpoints (`/props`, `/metrics`, etc.) The router uses the `model` q
GET /props?model=ggml-org%2Fgemma-3-4b-it-GGUF%3AQ4_K_M
```
+For the **WebSocket** endpoint (`/v1/realtime`) The router also uses the `model` query parameter.
+
By default, the model will be loaded automatically if it's not loaded. To disable this, add `--no-models-autoload` when starting the server. Additionally, you can include `?autoload=true|false` in the query param to control this behavior per-request.
### GET `/models`: List available models
diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp
index d66aae426..6d8f8e7be 100644
--- a/tools/server/server-context.cpp
+++ b/tools/server/server-context.cpp
@@ -1,4 +1,7 @@
#include "server-context.h"
+#include <atomic>
+#include <thread>
+#include "base64.hpp"
#include "server-chat.h"
#include "server-common.h"
#include "server-decision.h"
@@ -360,8 +363,13 @@ struct server_slot {
// for TTS models, this is the embd generated from prev step, decode this to generate next hidden state
// corresponding to one token position (size = n_embd)
+ // for realtime transcription, these are the rows from the last audio stream step (size = n_rows * n_embd)
std::vector<float> inp_embd;
+ // realtime transcription, see SERVER_TASK_TYPE_TRANSCRIBE_RT
+ mtmd::audio_stream_ptr audio_stream;
+ bool audio_wait = false; // the stream needs more audio before it can produce rows
+
server_slot_stats stats;
// accepted tokens per draft position
@@ -413,6 +421,9 @@ struct server_slot {
// clear multimodal state
mbatch.reset();
+ audio_stream.reset();
+ audio_wait = false;
+ inp_embd.clear();
}
void init_sampler() const {
@@ -542,16 +553,28 @@ struct server_slot {
// add sampled token of this slot to the batch, optionally add the speculative draft tokens if any
void handle_last_sampled_token(server_batch & batch) {
bool add_ok = true;
+ // the audio stream has no token before the first step, track the position with a dummy token
+ const llama_token tok_track = sampled == LLAMA_TOKEN_NULL ? 0 : sampled;
if (spec_draft.empty()) {
// no speculative decoding
i_batch = batch.size();
if (!inp_embd.empty()) {
- // text embeddings broadcast the same position across the M-RoPE sections
- const llama_pos p = prompt.tokens.pos_next();
- const llama_pos pos[GGML_MROPE_SECTIONS] = { p, p, p, 0 };
- GGML_ASSERT((int32_t) inp_embd.size() == batch.n_embd);
- add_ok &= batch.add_embd(id, inp_embd.data(), pos, true, false);
+ const int32_t n_rows = (int32_t) inp_embd.size() / batch.n_embd;
+ GGML_ASSERT((int32_t) inp_embd.size() == n_rows * batch.n_embd);
+ i_batch = batch.size() + n_rows - 1;
+ for (int32_t i = 0; i < n_rows; i++) {
+ // text embeddings broadcast the same position across the M-RoPE sections
+ const llama_pos p = prompt.tokens.pos_next();
+ const llama_pos pos[GGML_MROPE_SECTIONS] = { p, p, p, 0 };
+ add_ok &= batch.add_embd(id, inp_embd.data() + (size_t) i * batch.n_embd, pos, i == n_rows - 1, false);
+ if (i < n_rows - 1) {
+ prompt.tokens.push_back(tok_track);
+ }
+ }
+ if (audio_stream) {
+ inp_embd.clear(); // rows are consumed, the next step produces new ones
+ }
} else {
add_ok &= batch.add(id, sampled, prompt.tokens.pos_next(), true, false);
}
@@ -579,7 +602,7 @@ struct server_slot {
GGML_ASSERT(add_ok && "batch must be large enough to hold the sampled and draft tokens");
- prompt.tokens.push_back(sampled);
+ prompt.tokens.push_back(tok_track);
prompt.tokens.insert(spec_draft);
}
@@ -595,7 +618,7 @@ struct server_slot {
// do not keep context of the child slots - the parent's context is enough
// nor of the stateless slots - no task ever reuses it
- if (task->is_child() || task->is_stateless()) {
+ if (task->is_child() || task->is_stateless() || audio_stream) {
prompt_clear();
}
@@ -1967,6 +1990,23 @@ private:
? SLOT_STATE_WAIT_OTHER // wait for the parent to process prompt
: SLOT_STATE_STARTED;
+ // no prompt, the rows come from the audio stream, which audio controls reach from now on
+ if (slot.task->type == SERVER_TASK_TYPE_TRANSCRIBE_RT) {
+ slot.audio_stream.reset(mtmd_audio_stream_init(mctx, mtmd_audio_stream_params_default()));
+ if (!slot.audio_stream) {
+ send_error(slot, "failed to initialize the audio stream", ERROR_TYPE_SERVER);
+ slot.release();
+ return false;
+ }
+ slot.prompt_clear();
+ slot.init_sampler();
+ slot.sampled = LLAMA_TOKEN_NULL;
+ slot.i_batch = -1;
+ slot.stats.update_prompt_start();
+ slot.state = SLOT_STATE_GENERATING;
+ send_partial_response(slot, {}, false, true);
+ }
+
// reset server kill-switch counter
n_empty_consecutive = 0;
@@ -2829,6 +2869,7 @@ private:
switch (task.type) {
case SERVER_TASK_TYPE_COMPLETION:
case SERVER_TASK_TYPE_INFILL:
+ case SERVER_TASK_TYPE_TRANSCRIBE_RT:
case SERVER_TASK_TYPE_EMBEDDING:
case SERVER_TASK_TYPE_RERANK:
case SERVER_TASK_TYPE_DECISION:
@@ -2916,15 +2957,15 @@ private:
server_slot * slot = get_slot_by_cmpl_id(task.params.control_cmpl_id);
if (slot == nullptr) {
- SRV_WRN("control %s on unknown completion id=%s, no live slot\n",
- task.params.control_action.c_str(), task.params.control_cmpl_id.c_str());
+ SRV_WRN("control %d on unknown completion id=%s, no live slot\n",
+ (int) task.params.control_action, task.params.control_cmpl_id.c_str());
res->success = false;
res->message = "no active completion for this id";
queue_results.send(std::move(res));
break;
}
- if (task.params.control_action == "reasoning_end") {
+ if (task.params.control_action == SERVER_CONTROL_ACTION_REASONING_END) {
// the budget sampler only exists when reasoning control was armed
if (!slot->task->params.sampling.reasoning_control) {
res->success = false;
@@ -2935,6 +2976,22 @@ private:
// act on the live slot mid generation, never defer
common_sampler_reasoning_budget_force(slot->smpl.get());
res->success = true;
+ } else if (task.params.control_action == SERVER_CONTROL_ACTION_AUDIO_APPEND
+ || task.params.control_action == SERVER_CONTROL_ACTION_AUDIO_COMMIT
+ || task.params.control_action == SERVER_CONTROL_ACTION_AUDIO_END) {
+ if (!slot->audio_stream) {
+ res->success = false;
+ res->message = "not a realtime transcription slot";
+ } else {
+ auto * stream = slot->audio_stream.get();
+ if (task.params.control_action == SERVER_CONTROL_ACTION_AUDIO_APPEND) {
+ mtmd_audio_stream_push(stream, task.params.control_audio.data(), task.params.control_audio.size());
+ } else {
+ mtmd_audio_stream_flush(stream, task.params.control_action == SERVER_CONTROL_ACTION_AUDIO_END);
+ }
+ slot->audio_wait = false;
+ res->success = true;
+ }
} else {
res->success = false;
res->message = "unknown control action";
@@ -3254,6 +3311,39 @@ private:
};
#endif
+ // fill slot.inp_embd with the next rows of the audio stream
+ // returns false if the slot has nothing to decode now
+ bool slot_audio_step(server_slot & slot) {
+ if (!slot.inp_embd.empty()) {
+ return true; // rows from a previous step, not decoded yet
+ }
+
+ const int32_t n = mtmd_audio_stream_step(slot.audio_stream.get(), slot.sampled);
+ if (n > 0) {
+ const float * rows = mtmd_audio_stream_get_output_embd(slot.audio_stream.get());
+ slot.inp_embd.assign(rows, rows + (size_t) n * batch.n_embd);
+ slot.audio_wait = false;
+ return true;
+ }
+ if (n == 0) {
+ slot.audio_wait = true;
+ return false;
+ }
+ if (n == -2) {
+ send_error(slot, "audio stream error", ERROR_TYPE_SERVER);
+ slot.release();
+ return false;
+ }
+
+ // end of stream
+ slot.stop = STOP_TYPE_EOS;
+ slot.has_next_token = false;
+ slot.print_timings();
+ send_final_response(slot);
+ slot.release();
+ return false;
+ }
+
void update_slots() {
#ifdef DEBUG_TIMINGS
static int64_t t_prev = 0;
@@ -3269,11 +3359,12 @@ private:
#endif
// check if all slots are idle
+ // a realtime slot waiting for audio has no work, the next audio control task wakes the loop
{
bool all_idle = true;
for (auto & slot : slots) {
- if (slot.is_processing()) {
+ if (slot.is_processing() && !slot.audio_wait) {
all_idle = false;
break;
}
@@ -3448,6 +3539,10 @@ private:
return;
}
+ if (slot.audio_stream && !slot_audio_step(slot)) {
+ return; // waiting for audio, or the stream ended
+ }
+
// check if we can batch this slot with the previous one
if (!slot_batched) {
slot_batched = &slot;
@@ -3601,6 +3696,11 @@ private:
return;
}
+ // no prompt, the rows come from the audio stream
+ if (slot.task->type == SERVER_TASK_TYPE_TRANSCRIBE_RT) {
+ return;
+ }
+
// this slot still has a prompt to be processed
if (slot.state == SLOT_STATE_PROCESSING_PROMPT || slot.state == SLOT_STATE_STARTED) {
const auto & input_tokens = slot.task->tokens;
@@ -4831,6 +4931,7 @@ server_context_meta server_context::get_meta() const {
/* has_inp_image */ impl->chat_params.allow_image,
/* has_inp_audio */ impl->chat_params.allow_audio,
/* has_inp_video */ impl->chat_params.allow_video,
+ /* has_audio_stream */ impl->mctx != nullptr && mtmd_support_audio_stream(impl->mctx),
/* json_ui_settings */ impl->json_ui_settings,
/* slot_n_ctx */ impl->n_ctx_slot(),
/* pooling_type */ llama_pooling_type(impl->ctx_tgt),
@@ -5611,7 +5712,7 @@ void server_routes::init_routes() {
server_task task(SERVER_TASK_TYPE_CONTROL);
task.id = rd.get_new_id();
task.params.control_cmpl_id = cmpl_id;
- task.params.control_action = action;
+ task.params.control_action = SERVER_CONTROL_ACTION_REASONING_END;
rd.post_task(std::move(task));
}
@@ -5686,6 +5787,197 @@ void server_routes::init_routes() {
session);
};
+ // realtime transcription over websocket, modeled after the OpenAI realtime API
+ // audio is PCM16 mono at the model sample rate, sent as base64 in input_audio_buffer.append events
+ // input_audio_buffer.commit with "final": true ends the input, closing the socket also ends it
+ this->ws_realtime = [this](const server_http_req &, server_http_context::ws_conn & conn) {
+ auto send_event = [&conn](json ev) {
+ ev["event_id"] = "event_" + random_string();
+ conn.send(safe_json_to_str(ev));
+ };
+ auto send_error_event = [&send_event](const std::string & message, const std::string & type) {
+ send_event(json {
+ {"type", "error"},
+ {"error", {
+ {"type", type},
+ {"code", nullptr},
+ {"message", message},
+ {"param", nullptr},
+ }},
+ });
+ };
+
+ if (!meta->has_audio_stream) {
+ send_error_event("The current model does not support realtime transcription.", "invalid_request_error");
+ conn.close();
+ return;
+ }
+
+ const std::string session_id = "sess_" + random_string();
+ const json session_config = json {
+ {"id", session_id},
+ {"object", "realtime.transcription_session"},
+ {"type", "transcription"},
+ {"audio", {{"input", {
+ {"format", {{"type", "audio/pcm"}, {"rate", mtmd_get_audio_sample_rate(ctx_server.mctx)}}},
+ {"transcription", {{"model", meta->model_name}}},
+ {"turn_detection", nullptr},
+ {"noise_reduction", nullptr},
+ }}}},
+ };
+
+ auto res = create_response();
+ auto & rd = res->rd;
+ {
+ server_task task(SERVER_TASK_TYPE_TRANSCRIBE_RT);
+ task.id = rd.get_new_id();
+ task.params.sampling = params.sampling;
+ task.params.stream = true;
+ task.params.oaicompat_cmpl_id = session_id;
+ task.params.oaicompat_model = meta->model_name;
+ rd.post_task(std::move(task));
+ }
+
+ // the slot answers once its audio stream is live, the session starts there
+ {
+ auto result = rd.next([&conn]() { return !conn.is_open(); });
+ if (!result || result->is_error()) {
+ if (result) {
+ send_error_event(json_value(result->to_json(), "message", std::string("server error")), "server_error");
+ }
+ conn.close();
+ return;
+ }
+ }
+ send_event(json {{"type", "session.created"}, {"session", session_config}});
+
+ // one item per committed audio buffer, deltas are attributed to the current item
+ std::mutex mutex_item;
+ std::string item_id = "item_" + random_string();
+ std::string item_prev; // the committed item before item_id, empty for the first one
+ std::string item_text;
+
+ // called with mutex_item held
+ auto send_completed = [&]() {
+ send_event(json {
+ {"type", "conversation.item.input_audio_transcription.completed"},
+ {"item_id", item_id},
+ {"content_index", 0},
+ {"transcript", item_text},
+ });
+ item_text.clear();
+ };
+
+ auto post_control = [&](server_control_action action, std::vector<float> && audio) {
+ server_response_reader ctl(queue_tasks, queue_results, HTTP_POLLING_SECONDS);
+ server_task task(SERVER_TASK_TYPE_CONTROL);
+ task.id = ctl.get_new_id();
+ task.params.control_action = action;
+ task.params.control_cmpl_id = session_id;
+ task.params.control_audio = std::move(audio);
+ ctl.post_task(std::move(task));
+ auto result = ctl.next([&conn]() { return !conn.is_open(); });
+ auto * control = result ? dynamic_cast<server_task_result_control *>(result.get()) : nullptr;
+ return control && control->success;
+ };
+
+ // client events are read on this thread, the transcript is sent from the thread below
+ std::atomic<bool> input_ended{false};
+ std::thread reader([&]() {
+ std::string msg;
+ while (conn.read(msg) != 0) {
+ json ev;
+ try {
+ ev = json::parse(msg);
+ } catch (const std::exception &) {
+ send_error_event("invalid JSON", "invalid_request_error");
+ continue;
+ }
+ const std::string type = json_value(ev, "type", std::string());
+ if (type == "session.update" || type == "transcription_session.update") {
+ send_event(json {{"type", "session.updated"}, {"session", session_config}});
+ } else if (type == "input_audio_buffer.append") {
+ std::string pcm;
+ try {
+ pcm = base64::decode(json_value(ev, "audio", std::string()));
+ } catch (const std::exception &) {
+ send_error_event("invalid base64 audio", "invalid_request_error");
+ continue;
+ }
+ std::vector<float> samples(pcm.size() / 2);
+ for (size_t i = 0; i < samples.size(); i++) {
+ int16_t v;
+ memcpy(&v, pcm.data() + 2 * i, sizeof(v));
+ samples[i] = v / 32768.0f;
+ }
+ if (!post_control(SERVER_CONTROL_ACTION_AUDIO_APPEND, std::move(samples))) {
+ break;
+ }
+ } else if (type == "input_audio_buffer.commit") {
+ // "final": true ends the input, the last item is completed when the stream drains
+ if (json_value(ev, "final", false)) {
+ input_ended = true;
+ post_control(SERVER_CONTROL_ACTION_AUDIO_END, {});
+ continue;
+ }
+ if (!post_control(SERVER_CONTROL_ACTION_AUDIO_COMMIT, {})) {
+ break;
+ }
+ // the current item is the committed one, its transcript completes it and the next item opens
+ std::lock_guard<std::mutex> lock(mutex_item);
+ send_event(json {
+ {"type", "input_audio_buffer.committed"},
+ {"previous_item_id", item_prev.empty() ? json(nullptr) : json(item_prev)},
+ {"item_id", item_id},
+ });
+ send_completed();
+ item_prev = item_id;
+ item_id = "item_" + random_string();
+ } else if (type == "input_audio_buffer.clear") {
+ send_event(json {{"type", "input_audio_buffer.cleared"}});
+ } else {
+ send_error_event("unknown event type: " + type, "invalid_request_error");
+ }
+ }
+ if (!input_ended.exchange(true)) {
+ post_control(SERVER_CONTROL_ACTION_AUDIO_END, {});
+ }
+ // the result loop waits on the socket, nobody reads it from here on
+ conn.close();
+ });
+
+ while (true) {
+ auto result = rd.next([&conn]() { return !conn.is_open(); });
+ if (!result) {
+ break; // connection closed
+ }
+ if (result->is_error()) {
+ send_error_event(json_value(result->to_json(), "message", std::string("server error")), "server_error");
+ break;
+ }
+ if (auto * partial = dynamic_cast<server_task_result_cmpl_partial *>(result.get())) {
+ if (!partial->content.text.empty()) {
+ std::lock_guard<std::mutex> lock(mutex_item);
+ item_text += partial->content.text;
+ send_event(json {
+ {"type", "conversation.item.input_audio_transcription.delta"},
+ {"item_id", item_id},
+ {"content_index", 0},
+ {"delta", partial->content.text},
+ });
+ }
+ } else if (result->is_stop()) {
+ std::lock_guard<std::mutex> lock(mutex_item);
+ send_completed();
+ break;
+ }
+ }
+
+ input_ended = true;
+ conn.close();
+ reader.join();
+ };
+
this->post_anthropic_messages = [this](const server_http_req & req) {
auto res = create_response();
std::vector<raw_buffer> files;
diff --git a/tools/server/server-context.h b/tools/server/server-context.h
index ed23f7d4e..da75e5327 100644
--- a/tools/server/server-context.h
+++ b/tools/server/server-context.h
@@ -24,6 +24,7 @@ struct server_context_meta {
bool has_inp_image;
bool has_inp_audio;
bool has_inp_video;
+ bool has_audio_stream; // realtime transcription, see mtmd_support_audio_stream()
json json_ui_settings;
int slot_n_ctx;
enum llama_pooling_type pooling_type;
@@ -144,6 +145,7 @@ struct server_routes {
server_http_context::handler_t post_responses_oai;
server_http_context::handler_t post_responses_tok_oai;
server_http_context::handler_t post_transcriptions_oai;
+ server_http_context::ws_handler_t ws_realtime;
server_http_context::handler_t post_anthropic_messages;
server_http_context::handler_t post_anthropic_count_tokens;
server_http_context::handler_t post_apply_template;
diff --git a/tools/server/server-http.cpp b/tools/server/server-http.cpp
index 33cbb156c..cf147c7e7 100644
--- a/tools/server/server-http.cpp
+++ b/tools/server/server-http.cpp
@@ -11,6 +11,7 @@
#include <memory>
#include <string>
#include <thread>
+#include <mutex>
//
// HTTP implementation using cpp-httplib
@@ -744,6 +745,56 @@ void server_http_context::post(const std::string & path, const server_http_conte
}
}
+void server_http_context::ws(const std::string & path, const server_http_context::ws_handler_t & handler) const {
+ const std::string full_path = path_prefix + path;
+
+ auto callback = [handler](const httplib::Request & req, httplib::ws::WebSocket & sock) {
+ struct conn_impl : server_http_context::ws_conn {
+ httplib::ws::WebSocket & sock;
+ std::mutex mutex_send;
+ conn_impl(httplib::ws::WebSocket & sock) : sock(sock) {}
+ int read(std::string & msg) override {
+ switch (sock.read(msg)) {
+ case httplib::ws::Text: return 1;
+ case httplib::ws::Binary: return 2;
+ default: return 0;
+ }
+ }
+ bool send(const std::string & msg) override {
+ std::lock_guard<std::mutex> lock(mutex_send);
+ return sock.send(msg);
+ }
+ bool send_binary(const std::string & msg) override {
+ std::lock_guard<std::mutex> lock(mutex_send);
+ return sock.send(msg.data(), msg.size());
+ }
+ bool is_open() const override {
+ return sock.is_open();
+ }
+ void close() override {
+ sock.close();
+ }
+ };
+ conn_impl conn(sock);
+ const std::function<bool()> should_stop = [&sock]() {
+ return !sock.is_open();
+ };
+ server_http_req request{
+ get_params(req),
+ get_headers(req),
+ req.path,
+ build_query_string(req),
+ req.body,
+ {},
+ should_stop
+ };
+ handler(request, conn);
+ };
+ for (const auto & srv : pimpl->servers) {
+ srv->WebSocket(full_path, callback);
+ }
+}
+
void server_http_context::del(const std::string & path, const server_http_context::handler_t & handler) const {
handlers.emplace(path, handler);
auto callback = [handler](const httplib::Request & req, httplib::Response & res) {
diff --git a/tools/server/server-http.h b/tools/server/server-http.h
index 63fd2c186..d123535e8 100644
--- a/tools/server/server-http.h
+++ b/tools/server/server-http.h
@@ -74,6 +74,20 @@ struct server_http_context {
using handler_t = std::function<server_http_res_ptr(const server_http_req & req)>;
mutable std::unordered_map<std::string, handler_t> handlers;
+ // websocket connection, see ws()
+ struct ws_conn {
+ // blocking read, returns 1 for a text message, 2 for a binary message, 0 when the connection is closed
+ virtual int read(std::string & msg) = 0;
+ // thread-safe, can be called while another thread is in read()
+ virtual bool send(const std::string & msg) = 0;
+ virtual bool send_binary(const std::string & msg) = 0;
+ virtual bool is_open() const = 0;
+ virtual void close() = 0;
+ virtual ~ws_conn() = default;
+ };
+ // the handler runs on its own HTTP thread for the whole lifetime of the connection
+ using ws_handler_t = std::function<void(const server_http_req & req, ws_conn & conn)>;
+
std::string path_prefix;
int port = 9931;
bool is_ssl = false;
@@ -89,6 +103,7 @@ struct server_http_context {
void get(const std::string & path, const handler_t & handler) const;
void post(const std::string & path, const handler_t & handler) const;
void del(const std::string & path, const handler_t & handler) const;
+ void ws(const std::string & path, const ws_handler_t & handler) const;
// Register the Google Cloud Platform (Vertex AI) compat (AIP_PREDICT_ROUTE env var, or /predict)
// Must be called AFTER all other API routes are registered
diff --git a/tools/server/server-models.cpp b/tools/server/server-models.cpp
index 95cbdd77a..c8e235129 100644
--- a/tools/server/server-models.cpp
+++ b/tools/server/server-models.cpp
@@ -1899,6 +1899,81 @@ server_http_res_ptr server_models::proxy_request(const server_http_req & req, co
return proxy;
}
+void server_models::proxy_ws(const server_http_req & req, server_http_context::ws_conn & conn, const std::string & name) {
+ auto send_error = [&conn](const std::string & message) {
+ conn.send(safe_json_to_str(json {
+ {"type", "error"},
+ {"error", {{"type", "server_error"}, {"code", nullptr}, {"message", message}, {"param", nullptr}}},
+ }));
+ conn.close();
+ };
+
+ auto meta = get_meta(name);
+ if (!meta.has_value() || !meta->is_running()) {
+ send_error("model name=" + name + " is not running");
+ return;
+ }
+ {
+ std::unique_lock<std::mutex> lk(mutex);
+ mapping[name].meta.last_used = ggml_time_ms();
+ mapping[name].req_count++;
+ }
+ SRV_INF("proxying websocket to model %s on port %d\n", name.c_str(), meta->port);
+
+ std::string url = std::string("ws://") + CHILD_ADDR + ":" + std::to_string(meta->port) + req.path;
+ if (!req.query_string.empty()) {
+ url += '?' + req.query_string;
+ }
+ // the client library adds its own upgrade headers
+ httplib::Headers headers;
+ for (const auto & [key, value] : req.headers) {
+ std::string lower = key;
+ std::transform(lower.begin(), lower.end(), lower.begin(), ::tolower);
+ if (lower == "host" || lower == "connection" || lower == "upgrade" || lower.compare(0, 14, "sec-websocket-") == 0) {
+ continue;
+ }
+ headers.emplace(key, value);
+ }
+
+ {
+ httplib::ws::WebSocketClient upstream(url, headers);
+ if (!upstream.connect()) {
+ send_error("failed to connect to model name=" + name);
+ } else {
+ // client to child
+ std::thread pump([&]() {
+ std::string msg;
+ while (true) {
+ const int res = conn.read(msg);
+ if (res == 0 || !(res == 1 ? upstream.send(msg) : upstream.send(msg.data(), msg.size()))) {
+ break;
+ }
+ }
+ upstream.close();
+ });
+ // child to client
+ std::string msg;
+ while (true) {
+ const auto res = upstream.read(msg);
+ if (res == httplib::ws::Fail || !(res == httplib::ws::Text ? conn.send(msg) : conn.send_binary(msg))) {
+ break;
+ }
+ }
+ conn.close();
+ pump.join();
+ }
+ }
+
+ std::unique_lock<std::mutex> lk(mutex);
+ auto it = mapping.find(name);
+ if (it != mapping.end() && it->second.req_count > 0) {
+ it->second.req_count--;
+ if (it->second.req_count == 0) {
+ sched->tick(lk);
+ }
+ }
+}
+
void server_models::handle_child_state(const std::string & name, const std::string & raw_input) {
server_state state;
json payload;
@@ -2277,6 +2352,24 @@ void server_models_routes::init_routes() {
return models.proxy_request(req, method, name, false);
};
+ this->proxy_ws = [this](const server_http_req & req, server_http_context::ws_conn & conn) {
+ std::string name = req.get_param("model");
+ bool autoload = is_autoload(params, req);
+ auto error_res = std::make_unique<server_http_res>();
+ if (!router_validate_model(name, models, autoload, error_res)) {
+ // error_res holds {"error": {...}}, send it as a websocket error event
+ json err = json::parse(error_res->data);
+ err["type"] = "error";
+ conn.send(safe_json_to_str(err));
+ conn.close();
+ return;
+ }
+ if (autoload) {
+ models.ensure_model_ready(name, req.should_stop);
+ }
+ models.proxy_ws(req, conn, name);
+ };
+
this->proxy_post = [this](const server_http_req & req) {
std::string method = "POST";
json body = json::parse(req.body);
diff --git a/tools/server/server-models.h b/tools/server/server-models.h
index b04fc7384..500ce9fb3 100644
--- a/tools/server/server-models.h
+++ b/tools/server/server-models.h
@@ -366,6 +366,8 @@ public:
// proxy an HTTP request to the model instance
server_http_res_ptr proxy_request(const server_http_req & req, const std::string & method, const std::string & name, bool update_last_used, bool detached = false);
+ // forward a websocket connection to the child, returns when either side closes
+ void proxy_ws(const server_http_req & req, server_http_context::ws_conn & conn, const std::string & name);
// handle message sent from server_child::notify_to_router()
// raw input must starts with CMD_CHILD_TO_ROUTER_STATE, followed by a JSON string
@@ -433,6 +435,8 @@ struct server_models_routes {
server_http_context::handler_t get_router_props;
server_http_context::handler_t proxy_get;
server_http_context::handler_t proxy_post;
+ server_http_context::ws_handler_t proxy_ws;
+
server_http_context::handler_t get_router_models;
server_http_context::handler_t post_router_models_load;
server_http_context::handler_t post_router_models_unload;
diff --git a/tools/server/server-task.h b/tools/server/server-task.h
index e7bb04398..de39ce3b8 100644
--- a/tools/server/server-task.h
+++ b/tools/server/server-task.h
@@ -18,6 +18,7 @@ enum server_task_type {
SERVER_TASK_TYPE_RERANK,
SERVER_TASK_TYPE_DECISION,
SERVER_TASK_TYPE_INFILL,
+ SERVER_TASK_TYPE_TRANSCRIBE_RT,
SERVER_TASK_TYPE_CANCEL,
SERVER_TASK_TYPE_CONTROL,
SERVER_TASK_TYPE_NEXT_RESPONSE,
@@ -48,6 +49,14 @@ enum stop_type {
STOP_TYPE_LIMIT,
};
+enum server_control_action {
+ SERVER_CONTROL_ACTION_NONE,
+ SERVER_CONTROL_ACTION_REASONING_END,
+ SERVER_CONTROL_ACTION_AUDIO_APPEND, // push control_audio into the slot's audio stream
+ SERVER_CONTROL_ACTION_AUDIO_COMMIT, // flush the audio stream, more audio can follow
+ SERVER_CONTROL_ACTION_AUDIO_END, // flush the audio stream and end it
+};
+
struct task_params {
bool stream = false;
bool include_usage = false;
@@ -86,8 +95,9 @@ struct task_params {
std::string oaicompat_cmpl_id;
// realtime control (SERVER_TASK_TYPE_CONTROL)
- std::string control_action;
- std::string control_cmpl_id;
+ server_control_action control_action = SERVER_CONTROL_ACTION_NONE;
+ std::string control_cmpl_id;
+ std::vector<float> control_audio; // PCM F32 samples for SERVER_CONTROL_ACTION_AUDIO_APPEND
// reported in generation_settings, parsing itself is owned by the chat session
common_chat_format chat_format = COMMON_CHAT_FORMAT_CONTENT_ONLY;
@@ -233,6 +243,7 @@ struct server_task {
switch (type) {
case SERVER_TASK_TYPE_COMPLETION:
case SERVER_TASK_TYPE_INFILL:
+ case SERVER_TASK_TYPE_TRANSCRIBE_RT:
return true;
case SERVER_TASK_TYPE_DECISION:
return !decision.labels.empty();
@@ -245,6 +256,7 @@ struct server_task {
switch (type) {
case SERVER_TASK_TYPE_COMPLETION:
case SERVER_TASK_TYPE_INFILL:
+ case SERVER_TASK_TYPE_TRANSCRIBE_RT:
return true;
default:
return false;
diff --git a/tools/server/server.cpp b/tools/server/server.cpp
index b16824107..41f29ded5 100644
--- a/tools/server/server.cpp
+++ b/tools/server/server.cpp
@@ -247,6 +247,7 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
routes.post_control = models_routes->proxy_post;
routes.post_responses_oai = models_routes->proxy_post;
routes.post_transcriptions_oai = models_routes->proxy_post;
+ routes.ws_realtime = models_routes->proxy_ws;
routes.post_anthropic_messages = models_routes->proxy_post;
routes.post_anthropic_count_tokens = models_routes->proxy_post;
routes.post_infill = models_routes->proxy_post;
@@ -292,6 +293,7 @@ int llama_server(common_params & params, int argc, char ** argv, server_child &
ctx_http.post("/responses", ex_wrapper(routes.post_responses_oai));
ctx_http.post("/v1/audio/transcriptions", ex_wrapper(routes.post_transcriptions_oai));
ctx_http.post("/audio/transcriptions", ex_wrapper(routes.post_transcriptions_oai));
+ ctx_http.ws ("/v1/realtime", routes.ws_realtime);
ctx_http.post("/v1/messages", ex_wrapper(routes.post_anthropic_messages)); // anthropic messages API
ctx_http.post("/infill", ex_wrapper(routes.post_infill));
ctx_http.post("/embedding", ex_wrapper(routes.post_embeddings)); // legacy