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 +#include +#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 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 && 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(result.get()) : nullptr; + return control && control->success; + }; + + // client events are read on this thread, the transcript is sent from the thread below + std::atomic 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 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 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(result.get())) { + if (!partial->content.text.empty()) { + std::lock_guard 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 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 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 #include #include +#include // // 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 lock(mutex_send); + return sock.send(msg); + } + bool send_binary(const std::string & msg) override { + std::lock_guard 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 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; mutable std::unordered_map 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; + 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 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 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(); + 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 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