From 8ca4060b7e30e76baaa5670ee1ec1b760b7f8552 Mon Sep 17 00:00:00 2001 From: Tarek Dakhran Date: Wed, 28 Oct 2026 01:26:12 +0100 Subject: [PATCH] server : add realtime conversation and transcription on /v1/realtime (#31269) * server : add realtime conversation and transcription on /v1/realtime Co-authored-by: Xuan Son Nguyen Assisted-by: Claude Opus 5.5 * examples : default the realtime client to the server port 9931 Assisted-by: Claude Opus 5.5 * server : sample realtime responses right after their own decode Slots respond in turns on one context: sampling at the start of the next step read the logits of whichever slot decoded last, so concurrent sessions mixed their outputs. Assisted-by: Claude Opus 5.5 * move more stuff to server-realtime, making it self-contained * examples : fix the type check of the realtime client Assisted-by: Claude Opus 5.5 * examples : use the model card prompts in the realtime client Assisted-by: Claude Opus 5.5 --------- Co-authored-by: Xuan Son Nguyen --- examples/realtime_client.py | 172 ++++++ tools/server/CMakeLists.txt | 2 + tools/server/README.md | 36 ++ tools/server/server-chat.cpp | 19 + tools/server/server-chat.h | 4 + tools/server/server-context.cpp | 122 +++- tools/server/server-realtime.cpp | 949 +++++++++++++++++++++++++++++++ tools/server/server-realtime.h | 94 +++ tools/server/server-task.cpp | 10 + tools/server/server-task.h | 29 + 10 files changed, 1423 insertions(+), 14 deletions(-) create mode 100644 examples/realtime_client.py create mode 100644 tools/server/server-realtime.cpp create mode 100644 tools/server/server-realtime.h diff --git a/examples/realtime_client.py b/examples/realtime_client.py new file mode 100644 index 000000000..5bf59cc4c --- /dev/null +++ b/examples/realtime_client.py @@ -0,0 +1,172 @@ +#!/usr/bin/env python3 +"""Example client for the llama-server /v1/realtime websocket, with the openai SDK. + +Start a server with an audio model, e.g. LFM3-Audio: + + llama-server -hf LiquidAI/LFM3-Audio-2B-GGUF + +Spoken conversation, the audio of each wav file is one user turn. It is sent in chunks as it would come from a +microphone, the server prefills it as it arrives so the response starts right after the last chunk: + + python examples/realtime_client.py --instructions "Respond with interleaved text and audio." question.wav + +Output modalities: left out, the model picks; "text", "audio" or "text,audio" (interleaved) force the response. +Without --instructions, a forced mode comes with the system prompt of the LFM3-Audio or LFM3-Omni model card: + + python examples/realtime_client.py --output-modalities text,audio --out answer.wav question.wav + +Text turns go in with --text, transcription sessions transcribe each file: + + python examples/realtime_client.py --text "What is the capital of France?" + python examples/realtime_client.py --type transcription --language en speech.wav + +Live transcription models (Voxtral Realtime, LFM3-Audio live ASR) transcribe while the audio is sent, a transcription +session opens with ?intent=transcription: + + python examples/realtime_client.py --type transcription speech.wav + +Requirements: pip install "openai[realtime]" numpy +""" + +import argparse +import asyncio +import base64 +import time +import wave + +import numpy as np +from openai import AsyncOpenAI + + +# system prompts of the LiquidAI/LFM3-Audio-2B and LiquidAI/LFM3-Omni-2B model cards for a forced output mode +MODE_INSTRUCTIONS = { + "audio": {"text,audio": "Respond with interleaved text and audio."}, + "omni": {"text": "Respond in text.", "text,audio": "Respond in interleaved text and audio."}, +} + + +def read_wav(path, rate): + # PCM16 wav -> mono int16 at the given rate + with wave.open(path, "rb") as f: + assert f.getsampwidth() == 2, "the example reads PCM16 wav files" + x = np.frombuffer(f.readframes(f.getnframes()), dtype=np.int16).astype(np.float32) + x = x.reshape(-1, f.getnchannels()).mean(axis=1) + sr = f.getframerate() + if sr != rate: + n = int(round(len(x) * rate / sr)) + x = np.interp(np.arange(n) * sr / rate, np.arange(len(x)), x) + return np.clip(x, -32768, 32767).astype(np.int16) + + +def write_wav(path, pcm, rate): + with wave.open(path, "wb") as f: + f.setnchannels(1) + f.setsampwidth(2) + f.setframerate(rate) + f.writeframes(pcm.tobytes()) + + +async def main(args): + # the SDK derives a wss:// URL from base_url, a plain http server needs ws:// + client = AsyncOpenAI(base_url=args.url, api_key=args.api_key, websocket_base_url=args.url.replace("http", "ws", 1)) + query = {"intent": "transcription"} if args.type == "transcription" else {} + async with client.realtime.connect(model=args.model, extra_query=query) as conn: + ev = await conn.recv() + if ev.type == "error": + print("error:", ev.error.message) + return + # the audio goes in at the rate of the model, PCM16 mono + session_info = ev.model_dump(warnings=False)["session"] + audio_fmt = session_info["audio"] + card = "omni" if "omni" in str(session_info.get("model", "")).lower() else "audio" + in_rate = audio_fmt["input"]["format"]["rate"] + out_rate = (audio_fmt.get("output") or {}).get("format", {}).get("rate", 24000) + + session = {"type": args.type} + if args.type == "realtime": + instructions = args.instructions + if instructions is None: + instructions = MODE_INSTRUCTIONS[card].get(args.output_modalities) + if instructions is not None: + session["instructions"] = instructions + if args.output_modalities: + session["output_modalities"] = args.output_modalities.split(",") + elif args.language: + session["audio"] = {"input": {"transcription": {"language": args.language}}} + await conn.session.update(session=session) # ty: ignore[invalid-argument-type] + ev = await conn.recv() + if ev.type == "error": + print("error:", ev.error.message) + return + + turns = [("audio", path) for path in args.wav] + [("text", text) for text in args.text] + for i, (kind, src) in enumerate(turns): + t_last = 0.0 + + # the input is sent while the events are read: live transcription arrives while the audio is sent + async def send_turn(): + nonlocal t_last + if kind == "audio": + pcm = read_wav(src, in_rate) + step = in_rate * args.chunk_ms // 1000 + for j in range(0, len(pcm), step): + chunk = pcm[j:j + step] + await conn.input_audio_buffer.append(audio=base64.b64encode(chunk.tobytes()).decode()) + if args.realtime: + await asyncio.sleep(len(chunk) / in_rate) + if args.type == "transcription": + # "final" ends the input, models that transcribe with a delay then complete the last item + await conn.send({"type": "input_audio_buffer.commit", "final": i == len(args.wav) - 1}) # ty: ignore[invalid-argument-type] + else: + await conn.conversation.item.create(item={ + "type": "message", "role": "user", "content": [{"type": "input_text", "text": src}]}) + if args.type == "realtime": + # the audio not committed yet ends the user turn + await conn.response.create() + t_last = time.perf_counter() + + print(f"--- turn {i} ({kind}: {src})") + sender = asyncio.create_task(send_turn()) + t_first = None + audio = [] + async for ev in conn: + if ev.type == "response.output_text.delta" or ev.type == "conversation.item.input_audio_transcription.delta": + t_first = t_first or time.perf_counter() + print(ev.delta, end="", flush=True) + elif ev.type == "response.output_audio.delta": + t_first = t_first or time.perf_counter() + audio.append(np.frombuffer(base64.b64decode(ev.delta), dtype=np.int16)) + elif ev.type in ("response.done", "conversation.item.input_audio_transcription.completed"): + break + elif ev.type == "error": + print("\nerror:", ev.error.message) + return + await sender + print() + if t_first is not None and t_first > t_last: + print(f"first output {1000 * (t_first - t_last):.0f} ms after the end of the input") + elif t_first is not None: + print(f"output started {1000 * (t_last - t_first):.0f} ms before the end of the input, " + f"done {1000 * (time.perf_counter() - t_last):.0f} ms after it") + if audio: + pcm_out = np.concatenate(audio) + path = f"{args.out.removesuffix('.wav')}_{i}.wav" if len(turns) > 1 else args.out + write_wav(path, pcm_out, out_rate) + print(f"audio: {len(pcm_out) / out_rate:.2f} s -> {path}") + + +if __name__ == "__main__": + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("wav", nargs="*", help="PCM16 wav files, one user turn each") + p.add_argument("--text", action="append", default=[], help="a user turn as text, after the audio turns, repeatable") + p.add_argument("--type", choices=["realtime", "transcription"], default="realtime", help="session type") + p.add_argument("--instructions", help="system prompt of the conversation") + p.add_argument("--output-modalities", choices=["text", "audio", "text,audio"], help="default: the model picks") + p.add_argument("--language", help="transcription language, e.g. en") + p.add_argument("--chunk-ms", type=int, default=100, help="audio chunk size") + p.add_argument("--realtime", action=argparse.BooleanOptionalAction, default=True, help="send the audio in real time") + p.add_argument("--out", default="response.wav", help="where the audio of the responses goes") + p.add_argument("--url", default="http://127.0.0.1:9931/v1") + p.add_argument("--api-key", default="none") + p.add_argument("--model", default="default", help="model name, used by the router") + asyncio.run(main(p.parse_args())) diff --git a/tools/server/CMakeLists.txt b/tools/server/CMakeLists.txt index 43d9027df..4f3fa2458 100644 --- a/tools/server/CMakeLists.txt +++ b/tools/server/CMakeLists.txt @@ -17,6 +17,8 @@ add_library(${TARGET} STATIC server-decision.h server-context.cpp server-context.h + server-realtime.cpp + server-realtime.h server-stream.cpp server-stream.h server-tools.cpp diff --git a/tools/server/README.md b/tools/server/README.md index de01eaeab..42a679202 100644 --- a/tools/server/README.md +++ b/tools/server/README.md @@ -1587,6 +1587,10 @@ Returns a JSON object with a boolean `success` field, and an optional `message` Transcribes streamed audio, for models with a streaming audio encoder. Compatible with OAI `/v1/realtime` schema. +Models that write the text while they listen use this endpoint: Voxtral Realtime, and live ASR models, which carry the prompt of their transcription in a chat template named `asr` (e.g. the LFM3-Audio live ASR checkpoints, the text of every 0.96 s of audio follows it). A live ASR model restarts its context at a pause after 15 s and always after 30 s, the transcript then goes on in a new line. A commit completes the audio with silence, so that its text comes out. + +Live ASR models can also hold a conversation, so they transcribe live only in sessions opened with `/v1/realtime?intent=transcription`, as OpenAI transcription sessions are; other sessions are a [realtime conversation](#realtime-conversation). + 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:* @@ -1608,6 +1612,38 @@ Closing the websocket also ends the input. Note that words held back by the model's delay at the time of a commit are attributed to the next item. +#### Realtime conversation + +Models that answer after the audio, like LFM3-Audio and LFM3-Omni, hold a conversation on the same endpoint (with an `asr` template, unless the session is opened with `?intent=transcription`). The audio is prefilled as it arrives, so the response starts right after the last chunk. Audio input is PCM16 mono at the model sample rate (`audio.input.format.rate` of `session.created`), audio output is PCM16 mono at `audio.output.format.rate`. + +`session.type` selects the session, it can be changed with `session.update` before the first audio or message: + +- `realtime` (default): a conversation, every response stays in context for the next turns +- `transcription`: each committed audio buffer is transcribed on its own, `audio.input.transcription.language` picks the language + +*Client events:* + +- `session.update`: `type`, `instructions` (the system prompt, before the first audio or message), `output_modalities`, `max_output_tokens` (an integer or `"inf"`, the default) +- `input_audio_buffer.append`: `audio` holds base64 PCM16 samples +- `input_audio_buffer.commit`: ends the audio of the user turn, in a transcription session the transcript follows +- `conversation.item.create`: a user message with `input_text` content +- `response.create`: ends the user turn and generates the response, `response.output_modalities` and `response.max_output_tokens` override the session +- `response.cancel`: stops the response being generated + +LFM3-Omni is not trained to reason, run it with `--reasoning off` so that its turns open with an empty think block. + +`output_modalities` picks the mode of the responses: left out, the model switches between text and audio on its own; `["text"]` keeps the response in text, `["audio"]` makes it speech and `["text", "audio"]` interleaves text with the speech. + +*Server events:* + +- `session.created` and `session.updated`: carry the `session` object +- `input_audio_buffer.committed`, `conversation.item.added`, `conversation.item.done`: the user item +- `response.created`, `response.output_item.added`, `response.content_part.added`, then `response.output_text.delta` and `response.output_audio.delta` (base64 PCM16) as they are generated, then the `.done` events and `response.done` +- `conversation.item.input_audio_transcription.delta` and `.completed`: in a transcription session +- `error`: `error.message` describes the problem + +See [examples/realtime_client.py](../../examples/realtime_client.py) for a client with the `openai` library. + ### POST `/v1/responses`: OpenAI-compatible Responses API *Options:* diff --git a/tools/server/server-chat.cpp b/tools/server/server-chat.cpp index 6d1c56c0d..e0f00b25d 100644 --- a/tools/server/server-chat.cpp +++ b/tools/server/server-chat.cpp @@ -721,3 +721,22 @@ json convert_transcriptions_to_chatcmpl( return chatcmpl_body; } + +llama_tokens server_chat_get_asr_prompt(const llama_model * model) { + const char * src = llama_model_chat_template(model, "asr"); + if (src == nullptr) { + return {}; + } + const llama_vocab * vocab = llama_model_get_vocab(model); + const llama_token bos = llama_vocab_bos(vocab); + const common_chat_template tmpl(src, bos == LLAMA_TOKEN_NULL ? "" : common_token_to_piece(vocab, bos, true), ""); + jinja::context ctx(tmpl.source()); + jinja::global_from_json(ctx, json { + {"bos_token", tmpl.bos_token()}, + {"messages", json::array()}, + {"add_generation_prompt", true}, + }, false); + jinja::runtime runtime(ctx); + const std::string prompt = jinja::runtime::gather_string_parts(runtime.execute(tmpl.prog))->as_string().str(); + return common_tokenize(vocab, prompt, false, true); +} diff --git a/tools/server/server-chat.h b/tools/server/server-chat.h index 86b842650..6ff9e34d0 100644 --- a/tools/server/server-chat.h +++ b/tools/server/server-chat.h @@ -21,4 +21,8 @@ json convert_transcriptions_to_chatcmpl( const std::map & in_files, std::vector & out_files); +// prompt of the "asr" chat template, live ASR models (e.g. LFM3-Audio live ASR) transcribe with it +// empty if the model has no such template, throws on a template error +llama_tokens server_chat_get_asr_prompt(const llama_model * model); + json server_chat_msg_diff_to_json_oaicompat(const common_chat_msg_diff & diff); diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp index 3ca27762c..2bd1ab9ec 100644 --- a/tools/server/server-context.cpp +++ b/tools/server/server-context.cpp @@ -8,6 +8,7 @@ #include "server-http.h" #include "server-task.h" #include "server-queue.h" +#include "server-realtime.h" #include "server-schema.h" #include "server-stream.h" @@ -402,6 +403,9 @@ struct server_slot { bool audio_wait = false; // the stream needs more audio before it can produce rows std::deque audio_commit_rows; // end row of each non-final commit not yet completed + // realtime conversation, see SERVER_TASK_TYPE_REALTIME + std::unique_ptr rt; + server_slot_stats stats; // accepted tokens per draft position @@ -460,6 +464,7 @@ struct server_slot { audio_wait = false; audio_commit_rows.clear(); inp_embd.clear(); + rt.reset(); } void init_sampler() const { @@ -660,7 +665,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() || audio_stream) { + if (task->is_child() || task->is_stateless() || audio_stream || task->type == SERVER_TASK_TYPE_REALTIME) { prompt_clear(); } @@ -1023,6 +1028,9 @@ public: mtmd_helper_init_opt init_opt = mtmd_helper_init_opt_default(); const llama_vocab * vocab = nullptr; + // prompt of the "asr" chat template, for live ASR + llama_tokens asr_prompt; + bool has_cap_tts() const { return mtmd_cap.audio_gen; } @@ -1705,6 +1713,13 @@ private: return false; } + try { + asr_prompt = server_chat_get_asr_prompt(model_tgt); + } catch (const std::exception & e) { + SRV_ERR("%s: \"asr\" chat template error: %s\n", __func__, e.what()); + return false; + } + // IMPORTANT: chat_params is reused across sleeping / resuming states, // never store llama_context/llama_model pointers in chat_params, // as they may be invalidated after sleeping @@ -2020,6 +2035,23 @@ private: } } + // realtime transcription: the rows come from the audio stream, the helper may bias some tokens + if (task.type == SERVER_TASK_TYPE_TRANSCRIBE_RT) { + auto sparams = mtmd_audio_stream_params_default(); + sparams.prompt = asr_prompt.data(); + sparams.n_prompt = asr_prompt.size(); + slot.audio_stream.reset(mtmd_audio_stream_init(mctx, sparams)); + if (!slot.audio_stream) { + send_error(task, "failed to initialize the audio stream", ERROR_TYPE_SERVER); + return false; + } + if (!asr_prompt.empty()) { + for (const auto & b : mtmd_helper::audio_stream_logit_bias(mctx, model_tgt)) { + task.params.sampling.logit_bias.push_back(b); + } + } + } + // initialize samplers if (task.need_sampling()) { try { @@ -2075,12 +2107,6 @@ private: // 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; @@ -2090,6 +2116,17 @@ private: send_partial_response(slot, {}, false, true); } + // no prompt either, the session appends text and audio and asks for responses with its controls + if (slot.task->type == SERVER_TASK_TYPE_REALTIME) { + slot.rt = std::make_unique(ctx_tgt, mctx, slot.id, slot.n_ctx, slot.task->params.sampling, + queue_results, slot.task->id, slot.task->index); + slot.prompt_clear(); + slot.stats.update_prompt_start(); + slot.state = SLOT_STATE_GENERATING; + slot.audio_wait = true; + send_partial_response(slot, {}, false, true); + } + // reset server kill-switch counter n_empty_consecutive = 0; @@ -2972,6 +3009,7 @@ private: case SERVER_TASK_TYPE_COMPLETION: case SERVER_TASK_TYPE_INFILL: case SERVER_TASK_TYPE_TRANSCRIBE_RT: + case SERVER_TASK_TYPE_REALTIME: case SERVER_TASK_TYPE_EMBEDDING: case SERVER_TASK_TYPE_RERANK: case SERVER_TASK_TYPE_TTS: @@ -3084,6 +3122,20 @@ private: // act on the live slot mid generation, never defer common_sampler_reasoning_budget_force(slot->smpl.get()); res->success = true; + } else if (slot->rt) { + // a cancel acts now, everything else runs after what is already queued + if (task.params.control_action == SERVER_CONTROL_ACTION_RESPONSE_CANCEL) { + slot->rt->cancel(); + } else { + slot->rt->push({ + task.params.control_action, + std::move(task.params.control_text), + std::move(task.params.control_audio), + task.params.control_response, + }); + } + slot->audio_wait = false; + 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) { @@ -3449,6 +3501,18 @@ private: } if (n > 0) { + // new sequence (e.g. live ASR restart): drop the old one, the text goes on in a new line + if (mtmd_audio_stream_is_seq_start(slot.audio_stream.get())) { + slot.prompt_clear(); + if (slot.n_sent_text > 0) { + completion_token_output sep; + sep.tok = LLAMA_TOKEN_NULL; + sep.text_to_send = "\n"; + slot.generated.append(common_chat_input(sep.text_to_send)); + slot.n_sent_text += sep.text_to_send.size(); + send_partial_response(slot, sep, false); + } + } 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; @@ -3534,6 +3598,14 @@ private: abort_all_slots("process_tts_slots() failed: " + std::string(e.what())); } + // so do the realtime conversations + try { + process_realtime_slots(); + } catch (const std::exception & e) { + SRV_ERR("process_realtime_slots() failed: %s\n", e.what()); + abort_all_slots("process_realtime_slots() failed: " + std::string(e.what())); + } + GGML_ASSERT(batch.slot_batched || batch.size() == 0); if (batch.slot_batched) { @@ -3596,6 +3668,21 @@ private: } } + // realtime conversations (SERVER_TASK_TYPE_REALTIME) drive their own decode loop, see server_realtime_slot + void process_realtime_slots() { + iterate(slots, [&](server_slot & slot) { + if (!slot.is_processing() || !slot.rt) { + return; + } + if (!slot.rt->update()) { + send_error(slot, slot.rt->error, slot.rt->error_type); + slot.release(); + return; + } + slot.audio_wait = slot.rt->is_idle(); + }); + } + void process_tts_slots() { iterate(slots, [&](server_slot & slot) { if (!slot.is_processing() || slot.task->type != SERVER_TASK_TYPE_TTS) { @@ -3675,8 +3762,8 @@ private: // apply context-shift if needed // TODO: simplify and improve iterate(slots, [&](server_slot & slot) { - if (slot.task && slot.task->type == SERVER_TASK_TYPE_TTS) { - // TTS slots drive their own decode loop in process_tts_slots(), never enter the shared batch + if (slot.task && (slot.task->type == SERVER_TASK_TYPE_TTS || slot.task->type == SERVER_TASK_TYPE_REALTIME)) { + // TTS and realtime conversation slots drive their own decode loop, never enter the shared batch return; } if (slot.state == SLOT_STATE_GENERATING && slot.prompt.n_tokens() + 1 >= slot.n_ctx) { @@ -3751,7 +3838,7 @@ private: // determine which slots are generating and drafting iterate(slots, [&](server_slot & slot) { - if (slot.state != SLOT_STATE_GENERATING || slot.task->type == SERVER_TASK_TYPE_TTS) { + if (slot.state != SLOT_STATE_GENERATING || slot.task->type == SERVER_TASK_TYPE_TTS || slot.task->type == SERVER_TASK_TYPE_REALTIME) { return; } @@ -3897,7 +3984,7 @@ private: return; // batch is full, skip remaining slots } - if (!slot.is_processing() || slot.task->type == SERVER_TASK_TYPE_TTS) { + if (!slot.is_processing() || slot.task->type == SERVER_TASK_TYPE_TTS || slot.task->type == SERVER_TASK_TYPE_REALTIME) { return; } @@ -5150,8 +5237,7 @@ server_context_meta server_context::get_meta() const { /* has_inp_video */ impl->chat_params.allow_video, /* has_cap_chat */ impl->has_cap_chat(), /* has_cap_tts */ impl->has_cap_tts(), - // only realtime transcription models (no chat) stream without a prompt - /* has_audio_stream */ impl->mtmd_cap.audio_stream && !impl->mtmd_cap.chat, + /* has_audio_stream */ impl->mtmd_cap.audio_stream, /* json_ui_settings */ impl->json_ui_settings, /* slot_n_ctx */ impl->n_ctx_slot(), /* pooling_type */ llama_pooling_type(impl->ctx_tgt), @@ -6025,7 +6111,7 @@ void server_routes::init_routes() { // 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) { + this->ws_realtime = [this](const server_http_req & 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)); @@ -6042,6 +6128,14 @@ void server_routes::init_routes() { }); }; + // models that answer after the audio hold a conversation, see server-realtime.h + // live transcription: Voxtral Realtime (no chat), or an "asr" model in a session opened with ?intent=transcription + const bool live_asr = !ctx_server.asr_prompt.empty() && req.get_param("intent") == "transcription"; + if (ctx_server.mtmd_cap.audio_stream && ctx_server.mtmd_cap.chat && !live_asr) { + server_realtime_conversation({queue_tasks, queue_results, *meta, params.sampling, ctx_server.mctx, ctx_server.mtmd_cap, ctx_server.vocab}, conn); + return; + } + if (!meta->has_audio_stream) { send_error_event("The current model does not support realtime transcription.", "invalid_request_error"); conn.close(); diff --git a/tools/server/server-realtime.cpp b/tools/server/server-realtime.cpp new file mode 100644 index 000000000..4f3c08b7c --- /dev/null +++ b/tools/server/server-realtime.cpp @@ -0,0 +1,949 @@ +#include "server-realtime.h" +#include "server-common.h" +#include "server-context.h" +#include "server-task.h" + +#include "base64.hpp" +#include "chat.h" +#include "common.h" +#include "mtmd.h" +#include "mtmd-helper.h" + +#include +#include +#include +#include +#include +#include +#include + +// +// a realtime conversation, modeled after the OpenAI realtime API (GA) +// +// one slot holds the conversation for the whole session: text and audio go into it as they arrive, so a response +// starts right after the last audio. session.type "realtime" holds a conversation, "transcription" transcribes each +// committed audio buffer on its own. +// + +namespace { + +// output_modalities of a response, unset: the model switches between text and audio itself +bool parse_output_modalities(const json & j, mtmd_helper_gen_audio_out_mode & mode, std::string & err) { + if (j.is_null()) { + mode = MTMD_HELPER_GEN_AUDIO_OUT_AUTO; + return true; + } + if (!j.is_array() || j.empty()) { + err = "output_modalities must be a non-empty array"; + return false; + } + bool text = false; + bool audio = false; + for (const auto & m : j) { + if (m == "text") { + text = true; + } else if (m == "audio") { + audio = true; + } else { + err = "output_modalities: unknown modality " + m.dump(); + return false; + } + } + if (text && audio) { + mode = MTMD_HELPER_GEN_AUDIO_OUT_MIXED; + } else { + mode = audio ? MTMD_HELPER_GEN_AUDIO_OUT_AUDIO : MTMD_HELPER_GEN_AUDIO_OUT_TEXT; + } + return true; +} + +bool parse_max_output_tokens(const json & j, int32_t & n) { + if (j == "inf") { + n = -1; + return true; + } + if (!j.is_number_integer() || j.get() < 1) { + return false; + } + n = (int32_t) std::min(j.get(), INT32_MAX); + return true; +} + +json output_modalities_json(mtmd_helper_gen_audio_out_mode mode) { + switch (mode) { + case MTMD_HELPER_GEN_AUDIO_OUT_TEXT: return json::array({"text"}); + case MTMD_HELPER_GEN_AUDIO_OUT_AUDIO: return json::array({"audio"}); + case MTMD_HELPER_GEN_AUDIO_OUT_MIXED: return json::array({"text", "audio"}); + default: return nullptr; + } +} + +struct realtime_conversation { + const server_realtime_deps & deps; + server_http_context::ws_conn & conn; + + std::string id = "sess_" + random_string(); + + // session config + std::string type = "realtime"; + std::string instructions; + std::string language; // transcription + mtmd_helper_gen_audio_out_mode out_mode = MTMD_HELPER_GEN_AUDIO_OUT_AUTO; + int32_t max_output_tokens = -1; // "inf" + int input_rate = 16000; + int output_rate = 24000; + + // the text around a turn, from the chat template + std::string first_header; // system prompt + first user header + std::string user_header; + std::string user_footer; + std::string assistant_header; + std::string assistant_footer; + + + // conversation state, owned by the reader thread + bool started = false; // something went into the slot, the session config is fixed + bool turn_open = false; // a user turn is being filled + bool has_audio = false; // the open turn has audio not committed yet + int n_turns = 0; + std::string next_header; // text that goes in before the next user turn + + // the item or response the results belong to, shared with the result loop + std::mutex mutex; + std::string last_item_id; + std::string transcribe_item_id; // transcription: the item being transcribed + std::string transcript; + std::string response_id; + std::string response_item_id; + std::string response_text; + int idx_text = -1; + int idx_audio = -1; + json parts = json::array(); + + std::atomic closed{false}; + + realtime_conversation(const server_realtime_deps & deps, server_http_context::ws_conn & conn) : deps(deps), conn(conn) {} + + // + // events + // + + void send_event(json ev) { + ev["event_id"] = "event_" + random_string(); + conn.send(safe_json_to_str(ev)); + } + + void send_error(const std::string & message, const std::string & type_err = "invalid_request_error") { + send_event({ + {"type", "error"}, + {"error", {{"type", type_err}, {"code", nullptr}, {"message", message}, {"param", nullptr}}}, + }); + } + + json session_json() const { + json audio_in = {{"format", {{"type", "audio/pcm"}, {"rate", input_rate}}}, {"turn_detection", nullptr}}; + if (type == "transcription") { + audio_in["transcription"] = {{"model", deps.meta.model_name}, {"language", language.empty() ? json(nullptr) : json(language)}}; + return { + {"id", id}, + {"object", "realtime.transcription_session"}, + {"type", "transcription"}, + {"audio", {{"input", audio_in}}}, + }; + } + json audio = {{"input", audio_in}}; + if (deps.cap.audio_gen) { + audio["output"] = {{"format", {{"type", "audio/pcm"}, {"rate", output_rate}}}}; + } + return { + {"id", id}, + {"object", "realtime.session"}, + {"type", "realtime"}, + {"model", deps.meta.model_name}, + {"instructions", instructions}, + {"output_modalities", output_modalities_json(out_mode)}, + {"max_output_tokens", max_output_tokens < 0 ? json("inf") : json(max_output_tokens)}, + {"audio", audio}, + }; + } + + // + // slot controls, run on the slot in the order they are posted + // + + bool control(server_control_action action, std::string text = {}, std::vector audio = {}, + server_control_response response = {}) { + server_response_reader ctl(deps.queue_tasks, deps.queue_results, 1); + server_task task(SERVER_TASK_TYPE_CONTROL); + task.id = ctl.get_new_id(); + task.params.control_action = action; + task.params.control_cmpl_id = id; + task.params.control_text = std::move(text); + task.params.control_audio = std::move(audio); + task.params.control_response = response; + ctl.post_task(std::move(task)); + auto result = ctl.next([this]() { return closed.load() || !conn.is_open(); }); + auto * res = result ? dynamic_cast(result.get()) : nullptr; + return res && res->success; + } + + // + // chat template + // + + std::string render(const std::vector & messages, bool add_generation_prompt = false) const { + const auto & chat = deps.meta.chat_params; + common_chat_templates_inputs inputs; + inputs.messages = messages; + inputs.add_generation_prompt = add_generation_prompt; + inputs.use_jinja = chat.use_jinja; + inputs.enable_thinking = false; + inputs.chat_template_kwargs = chat.chat_template_kwargs; + return common_chat_templates_apply(chat.tmpls.get(), inputs).prompt.str(); + } + + // the text around a user and an assistant turn, from placeholder messages + bool derive_turn_text() { + const std::string user = "@@USER@@"; + const std::string reply = "@@REPLY@@"; + auto msg = [](const char * role, const std::string & content) { + common_chat_msg m; + m.role = role; + m.content = content; + return m; + }; + std::vector sys; + if (!instructions.empty()) { + sys.push_back(msg("system", instructions)); + } + auto with = [&](std::vector rest, bool add_generation_prompt = false) { + std::vector all = sys; + all.insert(all.end(), rest.begin(), rest.end()); + return render(all, add_generation_prompt); + }; + const std::string a = with({msg("user", user)}); + const std::string g = with({msg("user", user)}, true); + const std::string c = with({msg("user", user), msg("assistant", reply)}); + const std::string d = with({msg("user", user), msg("assistant", reply), msg("user", user)}); + + const size_t pa = a.find(user); + const size_t pc = c.find(reply); + const size_t pd = d.find(user, c.size()); + if (pa == std::string::npos || pc == std::string::npos || pd == std::string::npos || + c.compare(0, a.size(), a) != 0 || d.compare(0, c.size(), c) != 0 || g.compare(0, a.size(), a) != 0 || pc < a.size()) { + return false; + } + // the response opens with the generation prompt, a past turn may render differently (e.g. its think block) + first_header = a.substr(0, pa); + user_footer = a.substr(pa + user.size()); + assistant_header = g.substr(a.size()); + assistant_footer = c.substr(pc + reply.size()); + user_header = d.substr(c.size(), pd - c.size()); + return true; + } + + // + // conversation + // + + // the header of a user turn goes in before its first audio or text + bool ensure_turn() { + if (turn_open) { + return true; + } + const std::string header = n_turns == 0 ? first_header : next_header + user_header; + if (!control(SERVER_CONTROL_ACTION_TEXT_APPEND, header)) { + return false; + } + next_header.clear(); + started = true; + turn_open = true; + return true; + } + + void on_append(const json & ev) { + std::string pcm; + try { + pcm = base64::decode(json_value(ev, "audio", std::string())); + } catch (const std::exception &) { + send_error("invalid base64 audio"); + return; + } + 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 (samples.empty() || !ensure_turn()) { + return; + } + if (control(SERVER_CONTROL_ACTION_AUDIO_APPEND, {}, std::move(samples))) { + has_audio = true; + } + } + + // ends the audio of the open turn, it becomes a user item + bool commit() { + if (!has_audio) { + send_error("input_audio_buffer.commit: the buffer is empty", "input_audio_buffer_commit_empty"); + return false; + } + if (!control(SERVER_CONTROL_ACTION_AUDIO_END)) { + return false; + } + has_audio = false; + + std::lock_guard lock(mutex); + const json prev = last_item_id.empty() ? json(nullptr) : json(last_item_id); + last_item_id = "item_" + random_string(); + send_event({{"type", "input_audio_buffer.committed"}, {"previous_item_id", prev}, {"item_id", last_item_id}}); + const json item = {{"id", last_item_id}, {"object", "realtime.item"}, {"type", "message"}, {"role", "user"}, + {"status", "completed"}, {"content", json::array({{{"type", "input_audio"}}})}}; + send_event({{"type", "conversation.item.added"}, {"previous_item_id", prev}, {"item", item}}); + send_event({{"type", "conversation.item.done"}, {"previous_item_id", prev}, {"item", item}}); + return true; + } + + void on_commit() { + if (!commit() || type != "transcription") { + return; + } + // transcription: the committed audio is transcribed on its own, then the slot is cleared for the next one + { + std::lock_guard lock(mutex); + transcribe_item_id = last_item_id; + transcript.clear(); + } + server_control_response response; + response.n_predict = max_output_tokens; + response.out_mode = MTMD_HELPER_GEN_AUDIO_OUT_TEXT; + if (!control(SERVER_CONTROL_ACTION_RESPONSE_CREATE, user_footer + assistant_header, {}, response)) { + return; + } + control(SERVER_CONTROL_ACTION_CLEAR); + turn_open = false; + n_turns = 0; + } + + void on_item_create(const json & ev) { + if (type == "transcription") { + send_error("conversation.item.create: not supported in a transcription session"); + return; + } + const json item = ev.value("item", json::object()); + if (json_value(item, "type", std::string()) != "message" || json_value(item, "role", std::string()) != "user") { + send_error("conversation.item.create: only user messages are supported"); + return; + } + std::string text; + for (const auto & part : item.value("content", json::array())) { + const std::string t = json_value(part, "type", std::string()); + if (t != "input_text") { + send_error("conversation.item.create: unsupported content type: " + t); + return; + } + text += json_value(part, "text", std::string()); + } + if (!ensure_turn() || !control(SERVER_CONTROL_ACTION_TEXT_APPEND, text)) { + return; + } + std::lock_guard lock(mutex); + const json prev = last_item_id.empty() ? json(nullptr) : json(last_item_id); + last_item_id = item.contains("id") && item["id"].is_string() ? item["id"].get() : "item_" + random_string(); + const json out = {{"id", last_item_id}, {"object", "realtime.item"}, {"type", "message"}, {"role", "user"}, + {"status", "completed"}, {"content", json::array({{{"type", "input_text"}, {"text", text}}})}}; + send_event({{"type", "conversation.item.added"}, {"previous_item_id", prev}, {"item", out}}); + send_event({{"type", "conversation.item.done"}, {"previous_item_id", prev}, {"item", out}}); + } + + void on_response_create(const json & ev) { + if (type == "transcription") { + send_error("response.create: not supported in a transcription session"); + return; + } + if (!turn_open) { + send_error("response.create: commit audio or create a message first", "conversation_empty"); + return; + } + mtmd_helper_gen_audio_out_mode mode = out_mode; + const json resp = ev.value("response", json::object()); + if (resp.contains("output_modalities")) { + std::string err; + if (!parse_output_modalities(resp["output_modalities"], mode, err)) { + send_error("response.create: " + err); + return; + } + } + if (!deps.cap.audio_gen && mode != MTMD_HELPER_GEN_AUDIO_OUT_TEXT && mode != MTMD_HELPER_GEN_AUDIO_OUT_AUTO) { + send_error("response.create: the model cannot generate audio"); + return; + } + if (!deps.cap.text_gen && mode != MTMD_HELPER_GEN_AUDIO_OUT_AUDIO && mode != MTMD_HELPER_GEN_AUDIO_OUT_AUTO) { + send_error("response.create: the model cannot generate text"); + return; + } + // audio not committed yet belongs to this turn + if (has_audio && !commit()) { + return; + } + + // the gen-audio pipeline opens the response in the asked mode + const std::string prompt = user_footer + assistant_header; + server_control_response response; + response.n_predict = max_output_tokens; + if (resp.contains("max_output_tokens") && !parse_max_output_tokens(resp["max_output_tokens"], response.n_predict)) { + send_error("response.create: max_output_tokens must be an integer or \"inf\""); + return; + } + response.out_mode = mode; + + { + std::lock_guard lock(mutex); + response_id = "resp_" + random_string(); + response_item_id = "item_" + random_string(); + response_text.clear(); + parts = json::array(); + idx_text = -1; + idx_audio = -1; + send_event({{"type", "response.created"}, {"response", {{"id", response_id}, {"object", "realtime.response"}, + {"status", "in_progress"}, {"output_modalities", output_modalities_json(mode)}, {"output", json::array()}}}}); + send_event({{"type", "response.output_item.added"}, {"response_id", response_id}, {"output_index", 0}, + {"item", {{"id", response_item_id}, {"object", "realtime.item"}, {"type", "message"}, {"role", "assistant"}, + {"status", "in_progress"}, {"content", json::array()}}}}); + } + if (!control(SERVER_CONTROL_ACTION_RESPONSE_CREATE, prompt, {}, response)) { + return; + } + // the response ends the turn, its tokens are in the slot; the footer goes in before the next turn + next_header = assistant_footer; + turn_open = false; + n_turns++; + } + + void on_session_update(const json & ev) { + const json session = ev.value("session", json::object()); + if (session.contains("type") && session["type"].is_string() && session["type"] != type) { + const std::string t = session["type"]; + if (t != "realtime" && t != "transcription") { + send_error("session.update: unknown session type " + t); + return; + } + if (started) { + send_error("session.update: the session type cannot change after the first audio or message"); + return; + } + type = t; + } + if (session.contains("instructions") && session["instructions"].is_string()) { + if (started) { + send_error("session.update: instructions cannot change after the first audio or message"); + } else { + instructions = session["instructions"]; + } + } + if (session.contains("output_modalities")) { + std::string err; + if (!parse_output_modalities(session["output_modalities"], out_mode, err)) { + send_error("session.update: " + err); + return; + } + } + if (session.contains("max_output_tokens") && !parse_max_output_tokens(session["max_output_tokens"], max_output_tokens)) { + send_error("session.update: max_output_tokens must be an integer or \"inf\""); + return; + } + const json audio = session.value("audio", json::object()); + const json in = audio.value("input", json::object()); + if (in.contains("format") && in["format"].is_object() && json_value(in["format"], "rate", input_rate) != input_rate) { + send_error("session.update: the input audio must be PCM16 at " + std::to_string(input_rate) + " Hz"); + } + if (in.contains("transcription") && in["transcription"].is_object()) { + language = json_value(in["transcription"], "language", language); + } + if (!started && !configure()) { + send_error("the chat template does not fit the realtime turn format", "server_error"); + return; + } + send_event({{"type", "session.updated"}, {"session", session_json()}}); + } + + // turn text of the session type, before anything went into the slot + bool configure() { + const std::string saved = instructions; + if (type == "transcription") { + const auto preset = common_chat_get_asr_prompt(deps.meta.chat_params.tmpls.get(), language); + instructions = preset.system; + } + const bool ok = derive_turn_text(); + instructions = saved; + return ok; + } + + void handle(const std::string & msg) { + json ev; + try { + ev = json::parse(msg); + } catch (const std::exception &) { + send_error("invalid JSON"); + return; + } + const std::string t = json_value(ev, "type", std::string()); + if (t == "session.update" || t == "transcription_session.update") { + on_session_update(ev); + } else if (t == "input_audio_buffer.append") { + on_append(ev); + } else if (t == "input_audio_buffer.commit") { + on_commit(); + } else if (t == "input_audio_buffer.clear") { + // the audio already went into the conversation + send_event({{"type", "input_audio_buffer.cleared"}}); + } else if (t == "conversation.item.create") { + on_item_create(ev); + } else if (t == "response.create") { + on_response_create(ev); + } else if (t == "response.cancel") { + control(SERVER_CONTROL_ACTION_RESPONSE_CANCEL); + } else { + send_error("unknown event type: " + t); + } + } + + // + // results of the slot + // + + static std::string encode_pcm16(const std::vector & samples) { + std::string bytes(samples.size() * sizeof(int16_t), '\0'); + for (size_t i = 0; i < samples.size(); i++) { + const int16_t v = (int16_t) std::lround(std::max(-1.0f, std::min(1.0f, samples[i])) * 32767.0f); + memcpy(&bytes[2 * i], &v, sizeof(v)); + } + return base64::encode(bytes); + } + + // a part of the response is opened on its first delta, in order of appearance + int open_part(const char * part_type) { + json part = std::string(part_type) == "output_text" ? json{{"type", part_type}, {"text", ""}} : json{{"type", part_type}, {"transcript", ""}}; + parts.push_back(part); + send_event({{"type", "response.content_part.added"}, {"response_id", response_id}, {"item_id", response_item_id}, + {"output_index", 0}, {"content_index", parts.size() - 1}, {"part", part}}); + return (int) parts.size() - 1; + } + + void on_result(const server_task_result_realtime & res) { + std::lock_guard lock(mutex); + if (!transcribe_item_id.empty()) { + if (!res.text.empty()) { + transcript += res.text; + send_event({{"type", "conversation.item.input_audio_transcription.delta"}, {"item_id", transcribe_item_id}, + {"content_index", 0}, {"delta", res.text}}); + } + if (res.response_done) { + send_event({{"type", "conversation.item.input_audio_transcription.completed"}, {"item_id", transcribe_item_id}, + {"content_index", 0}, {"transcript", transcript}}); + transcribe_item_id.clear(); + } + return; + } + + const json base = {{"response_id", response_id}, {"item_id", response_item_id}, {"output_index", 0}}; + if (!res.text.empty()) { + if (idx_text < 0) { + idx_text = open_part("output_text"); + } + response_text += res.text; + json ev = base; + ev["type"] = "response.output_text.delta"; + ev["content_index"] = idx_text; + ev["delta"] = res.text; + send_event(ev); + } + if (!res.audio.empty()) { + if (idx_audio < 0) { + idx_audio = open_part("output_audio"); + } + json ev = base; + ev["type"] = "response.output_audio.delta"; + ev["content_index"] = idx_audio; + ev["delta"] = encode_pcm16(res.audio); + send_event(ev); + } + if (!res.response_done) { + return; + } + + const std::string status = res.cancelled ? "cancelled" : "completed"; + for (size_t i = 0; i < parts.size(); i++) { + json ev = base; + ev["content_index"] = i; + if ((int) i == idx_text) { + parts[i]["text"] = response_text; + ev["type"] = "response.output_text.done"; + ev["text"] = response_text; + } else { + ev["type"] = "response.output_audio.done"; + } + send_event(ev); + ev.erase("text"); + ev["type"] = "response.content_part.done"; + ev["part"] = parts[i]; + send_event(ev); + } + const json item = {{"id", response_item_id}, {"object", "realtime.item"}, {"type", "message"}, {"role", "assistant"}, + {"status", status}, {"content", parts}}; + send_event({{"type", "response.output_item.done"}, {"response_id", response_id}, {"output_index", 0}, {"item", item}}); + send_event({{"type", "response.done"}, {"response", {{"id", response_id}, {"object", "realtime.response"}, {"status", status}, + {"output", json::array({item})}, {"usage", {{"output_tokens", res.n_decoded}}}}}}); + last_item_id = response_item_id; + } + + // + // session + // + + void run() { + input_rate = mtmd_get_audio_sample_rate(deps.mctx); + if (deps.cap.audio_gen) { + output_rate = mtmd_gen_audio_get_info(deps.mctx).sample_rate; + } + if (!configure()) { + send_error("the chat template does not fit the realtime turn format", "server_error"); + conn.close(); + return; + } + + server_response_reader rd(deps.queue_tasks, deps.queue_results, 1); + { + server_task task(SERVER_TASK_TYPE_REALTIME); + task.id = rd.get_new_id(); + task.params.sampling = deps.sampling; + task.params.stream = true; + task.params.oaicompat_cmpl_id = id; + task.params.oaicompat_model = deps.meta.model_name; + rd.post_task(std::move(task)); + } + // the slot answers once it holds the session + { + auto result = rd.next([this]() { return !conn.is_open(); }); + if (!result || result->is_error()) { + if (result) { + send_error(json_value(result->to_json(), "message", std::string("server error")), "server_error"); + } + conn.close(); + return; + } + } + send_event({{"type", "session.created"}, {"session", session_json()}}); + + // client events are handled on this thread, results are sent from the loop below + std::thread reader([this]() { + std::string msg; + while (conn.read(msg) != 0) { + handle(msg); + } + closed = true; + conn.close(); + }); + + while (true) { + auto result = rd.next([this]() { return !conn.is_open(); }); + if (!result) { + break; + } + if (result->is_error()) { + send_error(json_value(result->to_json(), "message", std::string("server error")), "server_error"); + break; + } + if (auto * res = dynamic_cast(result.get())) { + on_result(*res); + } + } + + closed = true; + conn.close(); + reader.join(); + } +}; + +} // namespace + +void server_realtime_conversation(const server_realtime_deps & deps, server_http_context::ws_conn & conn) { + realtime_conversation session(deps, conn); + session.run(); +} + +// +// server_realtime_slot +// + +server_realtime_slot::server_realtime_slot(llama_context * ctx, mtmd_context * mctx, llama_seq_id seq_id, int32_t n_ctx, + const common_params_sampling & sampling, server_response & queue_results, int id_task, int index) + : ctx(ctx), model(llama_get_model(ctx)), vocab(llama_model_get_vocab(llama_get_model(ctx))), mctx(mctx), seq_id(seq_id), + n_ctx(n_ctx), sampling(sampling), queue_results(queue_results), id_task(id_task), index(index) { + if (mtmd_gen_audio_get_info(mctx).type != MTMD_GEN_AUDIO_TYPE_NONE) { + gen.init(ctx, mctx); + } +} + +void server_realtime_slot::push(server_realtime_op && op) { + ops.push_back(std::move(op)); +} + +void server_realtime_slot::cancel() { + cancelled = responding; +} + +bool server_realtime_slot::is_idle() const { + return ops.empty() && !responding; +} + +bool server_realtime_slot::fail(const std::string & msg, enum error_type type) { + error = msg; + error_type = type; + return false; +} + +void server_realtime_slot::send_result(std::string text, std::vector audio, bool response_done) { + auto res = std::make_unique(); + res->id = id_task; + res->index = index; + res->text = std::move(text); + res->audio = std::move(audio); + res->response_done = response_done; + res->cancelled = response_done && cancelled; + res->n_decoded = n_decoded; + queue_results.send(std::move(res)); +} + +// appends tokens or embedding rows to the conversation, logits of the last one if asked +bool server_realtime_slot::decode(const llama_token * tokens, const float * rows, int32_t n, bool logits_last) { + const int32_t n_embd = llama_model_n_embd_inp(model); + const int32_t n_batch = llama_n_batch(ctx); + if (n_past + n >= n_ctx) { + return fail("the conversation does not fit in the context", ERROR_TYPE_EXCEED_CONTEXT_SIZE); + } + for (int32_t i0 = 0; i0 < n; i0 += n_batch) { + const int32_t n_cur = std::min(n_batch, n - i0); + llama_batch b = llama_batch_init(n_cur, rows ? n_embd : 0, 1); + for (int32_t i = 0; i < n_cur; i++) { + if (rows) { + memcpy(b.embd + (size_t) i * n_embd, rows + (size_t) (i0 + i) * n_embd, n_embd * sizeof(float)); + } else { + b.token[i] = tokens[i0 + i]; + } + b.pos[i] = n_past + i0 + i; + b.n_seq_id[i] = 1; + b.seq_id[i][0] = seq_id; + b.logits[i] = logits_last && i0 + i == n - 1; + } + b.n_tokens = n_cur; + const int ret = llama_decode(ctx, b); + llama_batch_free(b); + if (ret != 0) { + return fail("failed to decode the conversation"); + } + } + n_past += n; + return true; +} + +// decodes the rows the audio stream has ready +bool server_realtime_slot::audio_rows() { + while (true) { + const int32_t n = mtmd_audio_stream_step(stream.get(), LLAMA_TOKEN_NULL); + if (n == -2) { + return fail("audio stream error"); + } + if (n <= 0) { + return true; + } + if (!decode(nullptr, mtmd_audio_stream_get_output_embd(stream.get()), n, false)) { + return false; + } + } +} + +// samples the next text token of a response from the logits of the slot's own last decode; several slots +// respond in turns on the same context, so this runs before any other slot decodes +void server_realtime_slot::sample_next() { + sampled = LLAMA_TOKEN_NULL; + if (gen.valid() && !gen.next_is_text()) { + return; + } + sampled = common_sampler_sample(smpl.get(), ctx, -1); + common_sampler_accept(smpl.get(), sampled, true); +} + +bool server_realtime_slot::response_start(const server_realtime_op & op) { + auto sparams = sampling; + n_predict = op.response.n_predict; + n_decoded = 0; + cancelled = false; + + if (gen.valid()) { + // the gen-audio pipeline continues the conversation and follows the model's text/audio switching + mtmd_helper::gen_audio::inp inp; + inp.set_prompt(op.text); + inp.data.seq_id = seq_id; + inp.data.n_past = n_past; + inp.data.seed = sampling.seed; + inp.data.stream = true; + inp.data.out_type = MTMD_HELPER_GEN_AUDIO_OUTTYPE_PCM; + inp.data.out_mode = op.response.out_mode; + if (gen.set_input(inp.get()) != 0) { + return fail("failed to start the response"); + } + // the pipeline may bias some text tokens, e.g. the switch to audio of a text-only response + for (const auto & b : gen.logit_bias()) { + sparams.logit_bias.push_back(b); + } + int32_t ret; + while ((ret = gen.step_prompt(llama_n_batch(ctx))) > 0) { + } + if (ret < 0) { + return fail("failed to start the response"); + } + const float * h = llama_get_embeddings_ith(ctx, -1); + h_state.assign(h, h + llama_model_n_embd(model)); + } else { + const llama_tokens toks = common_tokenize(vocab, op.text, false, true); + if (toks.empty()) { + return fail("failed to start the response"); + } + if (!decode(toks.data(), nullptr, (int32_t) toks.size(), true)) { + return false; + } + } + smpl.reset(common_sampler_init(model, sparams)); + // the logits of the last decode are this slot's only until another slot decodes + sample_next(); + responding = true; + return true; +} + +// one text token or audio frame of the response being generated +bool server_realtime_slot::response_step() { + bool stop = false; + std::string text; + std::vector audio; + + if (gen.valid()) { + const float * h_next = nullptr; + if (gen.step_gen(sampled, h_state.data(), &h_next, &stop) != 0) { + return fail("response generation failed"); + } + if (sampled != LLAMA_TOKEN_NULL && !stop && !llama_vocab_is_control(vocab, sampled)) { + text = common_token_to_piece(vocab, sampled, false); + } + if (h_next) { + h_state.assign(h_next, h_next + llama_model_n_embd(model)); + } + } else { + stop = llama_vocab_is_eog(vocab, sampled); + if (!stop) { + if (!llama_vocab_is_control(vocab, sampled)) { + text = common_token_to_piece(vocab, sampled, false); + } + if (!decode(&sampled, nullptr, 1, true)) { + return false; + } + } + } + n_decoded++; + + // the pipeline decodes on its own, the response ends before the context is full + const bool ctx_full = gen.valid() ? + llama_memory_seq_pos_max(llama_get_memory(ctx), seq_id) + 2 >= n_ctx : + n_past + 1 >= n_ctx; + const bool done = stop || cancelled || ctx_full || (n_predict > 0 && n_decoded >= n_predict); + if (!done) { + sample_next(); + } + if (gen.valid()) { + if (done) { + gen.flush(); + } + int32_t sample_rate = 0; + const char * data = nullptr; + size_t data_len = 0; + if (gen.get_output(&sample_rate, &data, &data_len) == 0 && data_len > 0) { + audio.assign((const float *) data, (const float *) data + data_len / sizeof(float)); + } + } + if (!text.empty() || !audio.empty()) { + send_result(std::move(text), std::move(audio)); + } + if (done) { + if (gen.valid()) { + // the pipeline decoded the response, the sequence ends where it stopped + n_past = llama_memory_seq_pos_max(llama_get_memory(ctx), seq_id) + 1; + } + send_result({}, {}, true); + responding = false; + cancelled = false; + } + return true; +} + +bool server_realtime_slot::update() { + llama_set_embeddings(ctx, true); + + if (responding) { + return response_step(); + } + + while (!ops.empty() && !responding) { + server_realtime_op op = std::move(ops.front()); + ops.pop_front(); + switch (op.action) { + case SERVER_CONTROL_ACTION_TEXT_APPEND: + { + const llama_tokens toks = common_tokenize(vocab, op.text, false, true); + if (!toks.empty() && !decode(toks.data(), nullptr, (int32_t) toks.size(), false)) { + return false; + } + } break; + case SERVER_CONTROL_ACTION_AUDIO_APPEND: + { + if (!stream) { + stream.reset(mtmd_audio_stream_init(mctx, mtmd_audio_stream_params_default())); + if (!stream) { + return fail("failed to initialize the audio stream"); + } + } + mtmd_audio_stream_push(stream.get(), op.audio.data(), op.audio.size()); + if (!audio_rows()) { + return false; + } + } break; + case SERVER_CONTROL_ACTION_AUDIO_COMMIT: + case SERVER_CONTROL_ACTION_AUDIO_END: + { + // the end of the audio of a turn + if (stream) { + mtmd_audio_stream_flush(stream.get(), true); + if (!audio_rows()) { + return false; + } + stream.reset(); + } + } break; + case SERVER_CONTROL_ACTION_RESPONSE_CREATE: + { + if (!response_start(op)) { + return false; + } + } break; + case SERVER_CONTROL_ACTION_CLEAR: + { + llama_memory_seq_rm(llama_get_memory(ctx), seq_id, -1, -1); + stream.reset(); + n_past = 0; + } break; + default: + break; + } + } + return true; +} diff --git a/tools/server/server-realtime.h b/tools/server/server-realtime.h new file mode 100644 index 000000000..21d1812bd --- /dev/null +++ b/tools/server/server-realtime.h @@ -0,0 +1,94 @@ +#pragma once + +#include "server-http.h" +#include "server-queue.h" +#include "server-task.h" + +#include "llama.h" +#include "mtmd.h" +#include "mtmd-helper.h" +#include "sampling.h" + +#include +#include +#include + +struct server_context_meta; +struct common_params_sampling; +struct mtmd_context; +struct mtmd_helper_cap; +struct llama_vocab; + +struct server_realtime_deps { + server_queue & queue_tasks; + server_response & queue_results; + const server_context_meta & meta; + const common_params_sampling & sampling; + mtmd_context * mctx; + const mtmd_helper_cap & cap; + const llama_vocab * vocab; +}; + +// a realtime conversation over websocket, for models that answer after the audio (e.g. LFM3-Audio): +// the audio is prefilled as it arrives, responses follow output_modalities or the model's own switching +void server_realtime_conversation(const server_realtime_deps & deps, server_http_context::ws_conn & conn); + +// one control of a realtime conversation +struct server_realtime_op { + server_control_action action; + std::string text; + std::vector audio; + server_control_response response; +}; + +// the conversation of a slot that holds a realtime session (SERVER_TASK_TYPE_REALTIME), it runs on the server thread +// the controls run in order; the sequence is decoded by its own calls, not in the shared batch +struct server_realtime_slot { + server_realtime_slot(llama_context * ctx, mtmd_context * mctx, llama_seq_id seq_id, int32_t n_ctx, + const common_params_sampling & sampling, server_response & queue_results, int id_task, int index); + + void push(server_realtime_op && op); + void cancel(); // stops the response being generated + + // runs the queued controls, or one step of the response; false on error, see error and error_type + bool update(); + + // nothing to run until the next control + bool is_idle() const; + + std::string error; + enum error_type error_type = ERROR_TYPE_SERVER; + +private: + llama_context * ctx; + const llama_model * model; + const llama_vocab * vocab; + mtmd_context * mctx; + llama_seq_id seq_id; + int32_t n_ctx; + common_params_sampling sampling; + server_response & queue_results; + int id_task; + int index; + + std::deque ops; + llama_pos n_past = 0; // positions of the conversation in the sequence + bool responding = false; + bool cancelled = false; + int32_t n_predict = 0; + int32_t n_decoded = 0; + llama_token sampled = LLAMA_TOKEN_NULL; // next text token of the response, sampled right after the slot's own decode + + mtmd::audio_stream_ptr stream; // the audio of the open turn + mtmd_helper::gen_audio gen; // set if the model generates audio + std::vector h_state; // hidden state that the gen-audio pipeline steps from + common_sampler_ptr smpl; + + bool fail(const std::string & msg, enum error_type type = ERROR_TYPE_SERVER); + void send_result(std::string text, std::vector audio, bool response_done = false); + bool decode(const llama_token * tokens, const float * rows, int32_t n, bool logits_last); + bool audio_rows(); + void sample_next(); + bool response_start(const server_realtime_op & op); + bool response_step(); +}; diff --git a/tools/server/server-task.cpp b/tools/server/server-task.cpp index d7ac2a1dc..1a0d434a1 100644 --- a/tools/server/server-task.cpp +++ b/tools/server/server-task.cpp @@ -1511,6 +1511,16 @@ json server_task_result_rerank::to_json() { // // server_task_result_tts // +json server_task_result_realtime::to_json() { + return json { + {"text", text}, + {"n_samples", audio.size()}, + {"response_done", response_done}, + {"cancelled", cancelled}, + {"n_decoded", n_decoded}, + }; +} + json server_task_result_tts::to_json() { return json { {"sample_rate", sample_rate}, diff --git a/tools/server/server-task.h b/tools/server/server-task.h index ab128f3bd..4c934db36 100644 --- a/tools/server/server-task.h +++ b/tools/server/server-task.h @@ -20,6 +20,7 @@ enum server_task_type { SERVER_TASK_TYPE_DECISION, SERVER_TASK_TYPE_INFILL, SERVER_TASK_TYPE_TRANSCRIBE_RT, + SERVER_TASK_TYPE_REALTIME, // a conversation over /v1/realtime, see server-realtime.cpp SERVER_TASK_TYPE_CANCEL, SERVER_TASK_TYPE_CONTROL, SERVER_TASK_TYPE_NEXT_RESPONSE, @@ -57,6 +58,17 @@ enum server_control_action { 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 + // realtime conversations (SERVER_TASK_TYPE_REALTIME), applied in order with the audio actions + SERVER_CONTROL_ACTION_TEXT_APPEND, // append control_text to the conversation + SERVER_CONTROL_ACTION_RESPONSE_CREATE, // generate after control_text, see control_response + SERVER_CONTROL_ACTION_RESPONSE_CANCEL, // stop the response being generated + SERVER_CONTROL_ACTION_CLEAR, // drop the conversation +}; + +// a response of a realtime conversation +struct server_control_response { + mtmd_helper_gen_audio_out_mode out_mode = MTMD_HELPER_GEN_AUDIO_OUT_AUTO; + int32_t n_predict = -1; // text tokens and audio frames, -1: until the context is full }; struct task_params { @@ -100,6 +112,8 @@ struct task_params { 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 + std::string control_text; // raw text, special tokens are parsed + server_control_response control_response; // reported in generation_settings, parsing itself is owned by the chat session common_chat_format chat_format = COMMON_CHAT_FORMAT_CONTENT_ONLY; @@ -538,6 +552,21 @@ struct server_task_result_embd : server_task_result { json to_json_oaicompat(); }; +// one event of a realtime conversation: text and audio of the response being generated, or its end +struct server_task_result_realtime : server_task_result { + std::string text; + std::vector audio; // PCM F32 at the sample rate of mtmd_gen_audio_get_info() + bool response_done = false; + bool cancelled = false; + int32_t n_decoded = 0; + + virtual bool is_stop() override { + return false; // the session ends when the client leaves, the task is cancelled then + } + + virtual json to_json() override; +}; + struct server_task_result_tts : server_task_result { std::string audio; // raw bytes for this chunk (WAV or PCM, per request's out_type) int32_t sample_rate = 0;