char-fork/llama.cpp
git clone https://git.t4t.associates/char-fork/llama.cpp
8300f1411
char/aliceai-t5-moe
1#include "server-context.h" 2#include "server-chat.h" 3#include "server-common.h" 4#include "server-http.h" 5#include "server-task.h" 6#include "server-queue.h" 7#include "server-schema.h" 8#include "server-stream.h" 9 10#include "build-info.h" 11#include "common.h" 12#include "fit.h" 13#include "llama.h" 14#include "log.h" 15#include "sampling.h" 16#include "speculative.h" 17#include "mtmd.h" 18#include "mtmd-helper.h" 19 20#include <algorithm> 21#include <cstddef> 22#include <cinttypes> 23#include <exception> 24#include <memory> 25#include <filesystem> 26#include <random> 27#include <utility> 28#include <fstream> 29 30// fix problem with std::min and std::max 31#if defined(_WIN32 ) 32#define WIN32_LEAN_AND_MEAN 33#ifndef NOMINMAX 34# define NOMINMAX 35#endif 36#include <windows.h> 37#endif 38 39constexpr int HTTP_POLLING_SECONDS = 1 ; 40 41static common_speculative_output_limits server_output_limits (const common_params & params ) { 42if (params .embedding || 43 (params .pooling_type != LLAMA_POOLING_TYPE_UNSPECIFIED && params .pooling_type != LLAMA_POOLING_TYPE_NONE )) { 44return {params .n_batch ,1 }; 45 } 46 47auto result = common_speculative_get_output_limits ( 48params .n_batch ,params .n_parallel ,common_speculative_n_max (& params .speculative )); 49 50result .total = std::max < int32_t > (1 ,result .total ); 51result .per_seq = std::max < int32_t > (1 ,result .per_seq ); 52return result ; 53} 54 55// synthetic draft verification for benchmarking - accept draft tokens at random instead of by match with the target 56// on replay the draft was already accepted before a context checkpoint restore, so repeat the same decisions 57static std::vector < llama_token > server_sample_and_accept_synth ( 58common_sampler * smpl , 59llama_context * ctx , 60const std::vector < int32_t > & idxs , 61const llama_tokens & draft , 62const std::vector < double > & synth_probs , 63 std::mt19937 & rng , 64bool is_replay ) { 65GGML_ASSERT (idxs .size ()== draft .size ()+ 1 ); 66GGML_ASSERT (synth_probs .size () >=draft .size ()); 67 68 std::vector < llama_token > result ; 69result .reserve (idxs .size ()); 70 71const llama_vocab * vocab = llama_model_get_vocab (llama_get_model (ctx )); 72 std::uniform_real_distribution < double > dist (0.0 ,1.0 ); 73for (size_t i = 0 ;i < draft .size ();++ i ) { 74const llama_token id = common_sampler_sample (smpl ,ctx ,idxs [i ]); 75const bool accept = is_replay || dist (rng )< synth_probs [i ]; 76// do not accept a drafted EOG token - it would end the generation early 77// on replay the last token is from the target and can be EOG, so skip this check 78if (accept && (is_replay || !llama_vocab_is_eog (vocab ,draft [i ]))) { 79// synthetic draft tokens do not advance grammar or reasoning state 80// the last replay token is from the target and must advance both 81const bool is_replay_target = is_replay && i + 1 == draft .size (); 82common_sampler_accept (smpl ,draft [i ],is_replay_target ); 83result .push_back (draft [i ]); 84continue ; 85 } 86 87common_sampler_accept (smpl ,id , true); 88result .push_back (id ); 89return result ; 90 } 91 92const llama_token id = common_sampler_sample (smpl ,ctx ,idxs [draft .size ()]); 93common_sampler_accept (smpl ,id , true); 94result .push_back (id ); 95 96return result ; 97} 98 99// state diagram: https://github.com/ggml-org/llama.cpp/pull/9283 100enum slot_state { 101SLOT_STATE_IDLE , 102SLOT_STATE_WAIT_OTHER ,// after assigning a task, but waiting for parent slot to process prompt 103SLOT_STATE_STARTED ,// after assigning a task and about to process prompt 104SLOT_STATE_PROCESSING_PROMPT , 105SLOT_STATE_DONE_PROMPT , 106SLOT_STATE_GENERATING , 107}; 108 109struct server_slot ;// forward declaration 110 111struct server_batch { 112llama_batch batch ; 113bool batch_rendered = false; 114 115struct token { 116int32_t id_slot ; 117llama_token token ; 118llama_pos pos ; 119bool output ; 120bool is_prompt ;// for stats tracking 121 }; 122 std::vector < token > tokens ; 123int32_t n_tokens_alloc = 0 ; 124int32_t n_embd = 0 ; 125 126// track if given slot can be batched with slots already in the batch 127server_slot * slot_batched = nullptr ; 128 129// in embd mode, we temporarily swap out the tokens arr and restore it on clear() 130bool has_embd = false; 131llama_token * tokens_ptr = nullptr ; 132 std::vector < float > embd ; 133 134float alora_scale = -1.0f ; 135size_t alora_disabled_id = 0 ; 136 137server_batch () { 138batch .pos = nullptr ;// sentinel: uninitialized batch 139 } 140 141 ~server_batch () { 142if (batch .pos != nullptr ) { 143clear (); 144llama_batch_free (batch ); 145 } 146 } 147 148void init (int32_t n_tokens_alloc ,int32_t n_embd ) { 149this -> n_tokens_alloc = n_tokens_alloc ; 150this -> n_embd = n_embd ; 151batch = llama_batch_init (n_tokens_alloc ,0 ,1 ); 152tokens_ptr = batch .token ; 153tokens .reserve (n_tokens_alloc ); 154 } 155 156bool add (int32_t id_slot ,llama_token token ,llama_pos pos ,bool output ,bool is_prompt ) { 157GGML_ASSERT (!has_embd );// cannot mix tokens + embd in same batch 158GGML_ASSERT (batch .pos != nullptr ); 159if ((int32_t )tokens .size () >=n_tokens_alloc ) { 160return false; 161 } 162tokens .push_back ({id_slot ,token ,pos ,output ,is_prompt }); 163return true; 164 } 165 166bool add (int32_t id_slot ,const std::vector < float > & embd_in ,llama_pos pos ,bool output ,bool is_prompt ) { 167GGML_ASSERT (batch .pos != nullptr ); 168if ((int32_t )tokens .size () >=n_tokens_alloc ) { 169return false; 170 } 171tokens .push_back ({id_slot ,LLAMA_TOKEN_NULL ,pos ,output ,is_prompt }); 172has_embd = true; 173embd .insert (embd .end (),embd_in .begin (),embd_in .end ()); 174return true; 175 } 176 177void clear () { 178tokens .clear (); 179embd .clear (); 180common_batch_clear (batch ); 181slot_batched = nullptr ; 182alora_scale = -1.0f ; 183alora_disabled_id = 0 ; 184batch_rendered = false; 185has_embd = false; 186if (batch .token == nullptr ) { 187batch .token = tokens_ptr ; 188batch .embd = nullptr ; 189 } 190 } 191 192int32_t size ()const { 193return (int32_t )tokens .size (); 194 } 195 196void set_output (int32_t idx ,bool output ) { 197GGML_ASSERT (idx >=0 && idx < (int32_t )tokens .size ()); 198tokens [idx ].output = output ; 199 } 200 201void render () { 202GGML_ASSERT (!batch_rendered ); 203GGML_ASSERT (batch .pos != nullptr ); 204common_batch_clear (batch ); 205for (int32_t i = 0 ;i < size ();i ++ ) { 206const auto & t = tokens [i ]; 207common_batch_add (batch ,t .token ,t .pos , {t .id_slot },t .output ); 208 } 209if (has_embd ) { 210batch .token = nullptr ;// will be restored on clear() 211batch .embd = embd .data (); 212 } 213batch_rendered = true; 214 } 215 216llama_batch get_view (int32_t off ,int32_t n_tokens )const { 217GGML_ASSERT (batch .pos != nullptr ); 218GGML_ASSERT (batch_rendered ); 219GGML_ASSERT (off >=0 && off < size ()); 220GGML_ASSERT (n_tokens > 0 && off + n_tokens <=size ()); 221 222auto * token = batch .token ?batch .token + off :nullptr ; 223auto * embd = batch .embd ?batch .embd + off * n_embd :nullptr ; 224 225llama_batch view = { 226n_tokens , 227token , 228embd , 229batch .pos + off , 230batch .n_seq_id + off , 231batch .seq_id + off , 232batch .logits + off , 233 }; 234 235return view ; 236 } 237}; 238 239struct server_slot { 240int id ; 241 242llama_context * ctx_tgt = nullptr ; 243llama_context * ctx_dft = nullptr ; 244 245common_memory mem ; 246 247// multimodal 248mtmd_context * mctx = nullptr ; 249 mtmd::batch_ptr mbatch = nullptr ; 250 251// speculative decoding 252common_speculative * spec ; 253 254llama_tokens spec_draft ; 255llama_tokens spec_prompt ; 256 std::vector < int32_t > spec_i_batch ; 257common_prompt_checkpoint spec_ckpt ; 258bool spec_is_replay = false; 259 std::mt19937 spec_synth_rng ; 260 261// TODO: move members that belong to the task (such as `generated_text`, `has_new_line`) to task_results_state 262// see https://github.com/ggml-org/llama.cpp/pull/18283#issuecomment-3710175837 263 std::unique_ptr < const server_task > task ; 264 std::unique_ptr < const server_task > task_prev ;// used for debugging 265 266// used to determine the slot that has been used the longest 267int64_t t_last_used = -1 ; 268 269// generation props 270int32_t n_ctx = 0 ;// context size per slot 271int32_t n_keep = 0 ; 272int32_t i_batch = -1 ; 273 274// effective generation limit for the current task, -1 means unlimited 275int32_t n_predict_max = -1 ; 276 277size_t last_nl_pos = 0 ; 278 279 std::string generated_text ; 280 std::string debug_generated_text ; 281llama_tokens generated_tokens ; 282size_t n_sent_text = 0 ;// number of sent text character (i.e. handle partial UTF-8 on streaming) 283 284 std::vector < completion_token_output > generated_token_probs ; 285 286bool has_next_token = true; 287bool has_new_line = false; 288bool truncated = false; 289 290stop_type stop ; 291 292 std::string stopping_word ; 293 294// state 295slot_state state = SLOT_STATE_IDLE ; 296 297server_prompt prompt ; 298 299bool prompt_save (server_prompt_cache & prompt_cache )const { 300if (prompt .tokens .size ()== 0 ) { 301return false; 302 } 303 304const size_t cur_size_tgt = llama_state_seq_get_size_ext (ctx_tgt ,id ,LLAMA_STATE_SEQ_FLAGS_NONE ); 305const size_t cur_size_dft = ctx_dft ?llama_state_seq_get_size_ext (ctx_dft ,id ,LLAMA_STATE_SEQ_FLAGS_NONE ) :0 ; 306 307const size_t cur_size = cur_size_tgt + cur_size_dft ; 308 309SRV_TRC (" - saving prompt with length %d, total state size = %.3f MiB (draft: %.3f MiB)\n" , 310 (int )prompt .tokens .size (),cur_size / (1024.0 * 1024.0 ),cur_size_dft / (1024.0 * 1024.0 )); 311 312auto * cur = prompt_cache .alloc (prompt ,cur_size_tgt ,cur_size_dft ); 313if (cur == nullptr ) { 314return false; 315 } 316 317llama_state_seq_get_data_ext (ctx_tgt ,cur -> data .main .data (),cur_size_tgt ,id ,LLAMA_STATE_SEQ_FLAGS_NONE ); 318if (ctx_dft ) { 319llama_state_seq_get_data_ext (ctx_dft ,cur -> data .drft .data (),cur_size_dft ,id ,LLAMA_STATE_SEQ_FLAGS_NONE ); 320 } 321 322return true; 323 } 324 325bool prompt_load (server_prompt_cache & prompt_cache ,const server_tokens & tokens ) { 326bool res = prompt_cache .load (prompt ,tokens ,ctx_tgt ,ctx_dft ,id ); 327if (!res ) { 328SLT_WRN (* this ,"%s" ,"failed to load prompt from cache\n" ); 329 } 330 331return res ; 332 } 333 334void prompt_clear () { 335SLT_TRC (* this ,"clearing prompt with %zu tokens\n" ,prompt .tokens .size ()); 336 337mem .seq_rm (id ,-1 ,-1 ); 338 339prompt .clear (); 340 } 341 342 std::vector < common_adapter_lora_info > lora ; 343int32_t alora_invocation_start = -1 ; 344 345// sampling 346json json_schema ; 347 348common_sampler_ptr smpl ; 349 350llama_token sampled ;// in speculative mode, this is the last accepted token 351 352// for TTS models, this is the embd generated from prev step, decode this to generate next hidden state 353// corresponding to one token position (size = n_embd) 354 std::vector < float > inp_embd ; 355 356server_slot_stats stats ; 357 358// accepted tokens per draft position 359// not in server_slot_stats to avoid copying to every task result 360 std::vector < uint64_t > n_accepted_per_pos ; 361 362 std::function < void (int /* id_slot */ )> callback_on_release ; 363 std::function < void (const server_slot & )> callback_on_reset ;// called before reset() 364 365// this is for printing timings with slot progress, not part of metrics 366int64_t t_print_last = 0 ; 367int32_t n_gen_last = 0 ; 368 369void reset () { 370SLT_DBG (* this ,"%s" ,"\n" ); 371 372spec_is_replay = false; 373 374last_nl_pos = 0 ; 375generated_text = "" ; 376has_new_line = false; 377truncated = false; 378stop = STOP_TYPE_NONE ; 379stopping_word = "" ; 380n_sent_text = 0 ; 381 382if (can_speculate ()) { 383spec_draft .clear (); 384spec_i_batch .clear (); 385spec_ckpt .clear (); 386 } 387generated_tokens .clear (); 388generated_token_probs .clear (); 389json_schema = json (); 390 391task_prev = std::move (task ); 392task .reset (); 393 394// note: callback_on_reset() must have run before this, see release() 395stats = {}; 396n_accepted_per_pos .clear (); 397 398n_predict_max = -1 ; 399 400llama_set_sampler (ctx_tgt ,id ,nullptr ); 401 402// clear alora start 403alora_invocation_start = -1 ; 404 405// clear multimodal state 406mbatch .reset (); 407 } 408 409void init_sampler ()const { 410common_sampler_reset (smpl .get ()); 411 412if (!task -> need_sampling ()) { 413return ; 414 } 415 416const int64_t t_start = ggml_time_us (); 417 418int n_text = 0 ; 419 420for (int i = 0 ;i < (int )prompt .tokens .size ();i ++ ) { 421const llama_token id = prompt .tokens [i ]; 422 423if (id != LLAMA_TOKEN_NULL ) { 424common_sampler_accept (smpl .get (),id , false); 425n_text ++ ; 426 } 427 } 428 429SLT_TRC (* this ,"init sampler, took %0.2f ms, tokens: text = %d, total = %d\n" , 430 (ggml_time_us ()- t_start ) /1000.0 ,n_text , (int )prompt .tokens .size ()); 431 } 432 433bool need_embd ()const { 434GGML_ASSERT (task ); 435return task -> need_embd (); 436 } 437 438// if the context does not have a memory module then all embeddings have to be computed within a single ubatch 439// also we cannot split if the pooling would require any past tokens 440// (MTP supports splitting — uses task->need_embd() not need_embd()) 441bool can_split ()const { 442GGML_ASSERT (task ); 443 444if (llama_model_has_encoder (llama_get_model (ctx_tgt ))&& llama_model_has_decoder (llama_get_model (ctx_tgt ))) { 445return false; 446 } 447 448return 449 !task -> need_embd ()|| 450 (llama_get_memory (ctx_tgt )&& llama_pooling_type (ctx_tgt )== LLAMA_POOLING_TYPE_LAST ); 451 } 452 453bool can_batch_with (server_slot & other_slot )const { 454GGML_ASSERT (task ); 455 456return task -> type == other_slot .task -> type 457&& inp_embd .size ()== other_slot .inp_embd .size () 458&& are_lora_equal (lora ,other_slot .lora ); 459 } 460 461// returns -1 if the generation is limitless 462int32_t n_remaining ()const { 463return n_predict_max == -1 ?-1 :n_predict_max - (int32_t )stats .n_gen ; 464 } 465 466bool has_budget ()const { 467return n_predict_max == -1 || n_remaining ()> 0 ; 468 } 469 470bool is_processing ()const { 471return state != SLOT_STATE_IDLE ; 472 } 473 474bool can_speculate ()const { 475return !!spec ; 476 } 477 478void add_token (const completion_token_output & token ) { 479if (!is_processing ()) { 480SLT_WRN (* this ,"%s" ,"slot is not processing\n" ); 481return ; 482 } 483 484generated_token_probs .push_back (token ); 485 } 486 487int get_n_draft_max ()const { 488GGML_ASSERT (task ); 489 490if (!can_speculate ()) { 491return 0 ; 492 } 493 494// determine the max draft that fits the current slot state 495// note: slot.prompt is not yet expanded with the `id` token sampled above 496// also, need to leave space for 1 extra token to allow context shifts 497int n_draft_max = n_ctx - prompt .n_tokens ()- 2 ; 498 499if (n_remaining ()> 0 ) { 500n_draft_max = std::min (n_draft_max ,n_remaining ()- 1 ); 501 } 502 503SLT_DBG (* this ,"max possible draft: %d\n" ,n_draft_max ); 504 505return n_draft_max ; 506 } 507 508// add sampled token of this slot to the batch, optionally add the speculative draft tokens if any 509void handle_last_sampled_token (server_batch & batch ) { 510bool add_ok = true; 511if (spec_draft .empty ()) { 512// no speculative decoding 513i_batch = batch .size (); 514 515if (!inp_embd .empty ()) { 516add_ok &=batch .add (id ,inp_embd ,prompt .tokens .pos_next (), true, false); 517 }else { 518add_ok &=batch .add (id ,sampled ,prompt .tokens .pos_next (), true, false); 519 } 520 521SLT_DBG (* this ,"slot decode token, id=%d, n_ctx = %d, n_tokens = %d, truncated = %d\n" , 522sampled ,n_ctx ,prompt .n_tokens (),truncated ); 523 }else { 524SLT_DBG (* this ,"generate_draft: id=%d, #tokens=%zu, #draft=%zu, pos_next=%d\n" , 525sampled ,prompt .tokens .size (),spec_draft .size (),prompt .tokens .pos_next ()); 526 527GGML_ASSERT (spec_i_batch .empty ()); 528 529spec_i_batch .push_back (batch .size ()); 530for (size_t i = 0 ;i < spec_draft .size ();i ++ ) { 531spec_i_batch .push_back (batch .size ()+ i + 1 ); 532 } 533 534auto pos0 = prompt .tokens .pos_next (); 535 536add_ok &=batch .add (id ,sampled ,pos0 ++ , true, false); 537for (auto token :spec_draft ) { 538add_ok &=batch .add (this -> id ,token ,pos0 ++ , true, false); 539 } 540 } 541 542GGML_ASSERT (add_ok && "batch must be large enough to hold the sampled and draft tokens" ); 543 544prompt .tokens .push_back (sampled ); 545prompt .tokens .insert (spec_draft ); 546 } 547 548void release () { 549if (is_processing ()) { 550GGML_ASSERT (task ); 551 552SLT_INF (* this ,"stop processing: n_tokens = %d, truncated = %d\n" ,prompt .n_tokens (),truncated ); 553 554t_last_used = ggml_time_us (); 555 556state = SLOT_STATE_IDLE ; 557 558// do not keep context of the child slots - the parent's context is enough 559if (task -> is_child ()) { 560prompt_clear (); 561 } 562 563callback_on_reset (* this ); 564 565reset (); 566 567callback_on_release (id ); 568 } 569 } 570 571size_t find_stopping_strings (const std::string & text ,const size_t last_token_size ,bool is_full_stop ) { 572GGML_ASSERT (task ); 573 574size_t stop_pos = std::string::npos ; 575 576for (const std::string & word :task -> params .antiprompt ) { 577size_t pos ; 578 579if (is_full_stop ) { 580const size_t tmp = word .size ()+ last_token_size ; 581const size_t from_pos = text .size ()> tmp ?text .size ()- tmp :0 ; 582 583pos = text .find (word ,from_pos ); 584 }else { 585// otherwise, partial stop 586pos = string_find_partial_stop (text ,word ); 587 } 588 589if (pos != std::string::npos && (stop_pos == std::string::npos || pos < stop_pos )) { 590if (is_full_stop ) { 591stop = STOP_TYPE_WORD ; 592stopping_word = word ; 593has_next_token = false; 594 } 595stop_pos = pos ; 596 } 597 } 598 599return stop_pos ; 600 } 601 602void print_timings_tg () { 603if (stats .n_gen < 100 ) { 604return ; 605 } 606 607const int64_t t_now = ggml_time_us (); 608 609if (t_now - t_print_last < 3 * 1000 * 1000 ) { 610return ; 611 } 612 613const double n_gen_second = stats .n_gen_tps (); 614const double n_gen_second_win = 1e6 / (t_now - t_print_last )* (stats .n_gen - n_gen_last ); 615 616t_print_last = t_now ; 617n_gen_last = stats .n_gen ; 618 619SLT_INF (* this ,"n_gen = %6d, tg = %6.2f t/s, tg_3s = %6.2f t/s\n" , (int )stats .n_gen ,n_gen_second ,n_gen_second_win ); 620 } 621 622void print_timings_pp ()const { 623const double t_prompt_total = stats .t_prompt_ms (); 624 625if (t_prompt_total < 3000.0 ) { 626return ; 627 } 628 629const double n_prompt_second = stats .n_prompt_tps (); 630const double f_progress = task -> n_tokens ()> 0 ? (double )prompt .n_tokens () /task -> n_tokens () :0.0 ; 631 632SLT_INF (* this ,"prompt processing, n_tokens = %6d, progress = %.2f, t = %6.2f s / %.2f tokens per second\n" , 633 (int )stats .n_prompt_processed ,f_progress ,t_prompt_total /1e3 ,n_prompt_second ); 634 } 635 636void print_timings ()const { 637const double t_prompt_total = stats .t_prompt_ms (); 638const double t_gen_total = stats .t_gen_ms (); 639 640const double t_prompt = stats .t_prompt_per_token_ms (); 641const double n_prompt_second = stats .n_prompt_tps (); 642 643const double t_gen = stats .t_gen_per_token_ms (); 644const double n_gen_second = stats .n_gen_tps (); 645 646SLT_INF (* this , 647"prompt eval time = %10.2f ms / %5d tokens (%8.2f ms per token, %8.2f tokens per second)\n" , 648t_prompt_total , (int )stats .n_prompt_processed ,t_prompt ,n_prompt_second ); 649 650SLT_INF (* this , 651" eval time = %10.2f ms / %5d tokens (%8.2f ms per token, %8.2f tokens per second)\n" , 652t_gen_total , (int )stats .n_gen ,t_gen ,n_gen_second ); 653 654SLT_INF (* this , 655" total time = %10.2f ms / %5d tokens\n" , 656t_prompt_total + t_gen_total , (int ) (stats .n_prompt_processed + stats .n_gen )); 657 658SLT_INF (* this , 659" graphs reused = %10d\n" , 660llama_perf_context (ctx_tgt ).n_reused ); 661 662const int32_t n_draft_total = stats .n_draft_tokens ; 663const int32_t n_draft_accepted = stats .n_draft_accepted ; 664const int32_t n_draft_verif_steps = stats .n_draft_verif_steps ; 665 666if (n_draft_total > 0 ) { 667const float draft_ratio = (float )n_draft_accepted /n_draft_total ; 668const double mean_acc_len = n_draft_verif_steps > 0 ?1.0 + (double )n_draft_accepted / (double )n_draft_verif_steps :1.0 ; 669 670 std::string acceptance_rates_per_pos ; 671if (n_draft_verif_steps > 0 ) { 672for (size_t i = 0 ;i < n_accepted_per_pos .size ();++ i ) { 673if (i > 0 ) { 674acceptance_rates_per_pos += ", " ; 675 } 676acceptance_rates_per_pos += string_format ("%.3f" , (double )n_accepted_per_pos [i ] / (double )n_draft_verif_steps ); 677 } 678 } 679 680SLT_INF (* this , 681"draft acceptance = %0.5f (%5d accepted / %5d generated), mean len = %5.2f\n" , 682draft_ratio ,n_draft_accepted ,n_draft_total ,mean_acc_len ); 683SLT_TRC (* this , 684" acc per pos = (%s)\n" ,acceptance_rates_per_pos .c_str ()); 685 } 686 687common_speculative_print_stats (spec ); 688 } 689 690json to_json (bool only_metrics = false)const { 691json res ; 692 693res = { 694 {"id" ,id }, 695 {"n_ctx" ,n_ctx }, 696 {"speculative" ,can_speculate ()}, 697 {"is_processing" ,is_processing ()}, 698 }; 699 700const auto & ptask = task ?task :task_prev ; 701 702if (ptask ) { 703res ["id_task" ]= ptask -> id ; 704res ["n_prompt_tokens" ]= (int32_t )prompt .tokens .size (); 705res ["n_prompt_tokens_processed" ]= stats .n_prompt_processed ; 706res ["n_prompt_tokens_cache" ]= stats .n_prompt_cached ; 707res ["params" ]= ptask -> params .to_json (only_metrics ); 708res ["next_token" ]= json::array ({ 709 { 710 {"has_next_token" ,has_next_token }, 711 {"has_new_line" ,has_new_line }, 712 {"n_remain" ,n_remaining ()}, 713 {"n_decoded" ,stats .n_gen }, 714 } 715 }); 716 717if (!only_metrics ) { 718res ["prompt" ]= ptask -> tokens .detokenize (ctx_tgt , true); 719res ["generated" ]= generated_text .empty () ?debug_generated_text :generated_text ; 720 } 721 } 722 723return res ; 724 } 725 726void copy_state_to (server_slot & other )const { 727GGML_ASSERT (state == SLOT_STATE_DONE_PROMPT ); 728 729mem .seq_rm (other .id ,-1 ,-1 ); 730mem .seq_cp (id ,other .id ,-1 ,-1 ); 731 732other .i_batch = i_batch ; 733 734other .stats = stats ; 735 736other .prompt = prompt .clone (); 737other .init_sampler (); 738 } 739}; 740 741// returns 0 on success 742// caller need to update prompt.tokens after a successful call to keep track of the processing progress 743// note: this is not a member of server_slot because we want to run it inside yield_to_queue 744// slot is passed as const to avoid accidental modification of the slot state 745// some pointers are allowed to be used, they are not used by to_json() 746static int process_mtmd_chunk (const server_slot & slot , mtmd::batch_ptr & mbatch ,size_t idx ,size_t & n_tokens_out ) { 747GGML_ASSERT (slot .mctx ); 748const auto & mctx = slot .mctx ; 749const auto & input_tokens = slot .task -> tokens ; 750const auto & chunk = input_tokens .find_chunk (idx ); 751int32_t res = 0 ; 752 753auto try_decode = [& ]()-> int32_t { 754if (mbatch ) { 755float * embd = mtmd_batch_get_output_embd (mbatch .get (),chunk .get ()); 756if (embd ) { 757void * cb_data = slot .spec ; 758static auto cb = [](llama_batch batch ,void * user_data ) { 759common_speculative * spec = static_cast < common_speculative *> (user_data ); 760if (!common_speculative_process (spec ,batch )) { 761return 1 ; 762 } 763return 0 ; 764 }; 765 766llama_pos new_n_past ;// unused for now 767res = mtmd_helper_decode_image_chunk ( 768mctx , 769slot .ctx_tgt , 770chunk .get (), 771embd , 772slot .prompt .tokens .pos_next (), 773slot .id , 774llama_n_batch (slot .ctx_tgt ), 775& new_n_past , 776cb , 777cb_data 778 ); 779if (res != 0 ) { 780SLT_ERR (slot ,"failed to decode mtmd chunk, idx = %zu, res = %d\n" ,idx ,res ); 781return -1 ; 782 } 783n_tokens_out = mtmd_input_chunk_get_n_tokens (chunk .get ()); 784return 0 ;// success 785 } 786 } 787return 1 ;// (non-error) need to create & encode batch 788 }; 789 790// if the batch is already exist, try searching & encode 791res = try_decode (); 792if (res == 0 ) { 793return 0 ; 794 } 795if (res < 0 ) { 796// fatal error 797return res ; 798 } 799 800// otherwise, the batch is either uninitialized or is used up 801// we need to create & encode a new batch 802mbatch .reset (mtmd_batch_init (mctx )); 803res = mtmd_batch_add_chunk (mbatch .get (),chunk .get ()); 804GGML_ASSERT (res == 0 );// we should never have an empty batch 805 806// try batching as much as possible 807int n_added = 1 ; 808size_t idx_cur = idx ; 809while (res == 0 ) { 810auto [next_chunk ,next_idx ]= input_tokens .find_next_media_chunk (idx_cur ); 811if (next_chunk == nullptr ) { 812break ; 813 } 814res = mtmd_batch_add_chunk (mbatch .get (),next_chunk -> get ()); 815n_added += (res == 0 ?1 :0 ); 816idx_cur = next_idx ; 817SLT_DBG (slot ,"try adding media chunk idx = %zu to batch, res = %d\n" ,next_idx ,res ); 818// if res != 0, batch is full or chunk is not compatible -> this loop breaks 819 } 820 821// TODO @ngxson : move this log line to debug when it become more stable 822SLT_TRC (slot ,"encoding mtmd batch from idx = %zu, n_chunks = %d\n" ,idx ,n_added ); 823 824res = mtmd_batch_encode (mbatch .get ()); 825if (res != 0 ) { 826SLT_ERR (slot ,"failed to encode mtmd batch for chunk idx = %zu, res = %d\n" ,idx ,res ); 827return -1 ; 828 } 829 830return try_decode (); 831} 832 833// 834// server_context_impl (private implementation) 835// 836 837struct server_context_impl { 838friend struct server_context ; 839 840public : 841// only use these pointers outside of this class: 842// - when not in sleeping state 843// - and, with thread-safe APIs (e.g., tokenizer calls) 844llama_model * model_tgt = nullptr ; 845 846mtmd_context * mctx = nullptr ; 847// note: video_params.ffmpeg_bin_dir points into params_base, which outlives this struct 848mtmd_helper_init_opt init_opt = mtmd_helper_init_opt_default (); 849const llama_vocab * vocab = nullptr ; 850 851server_queue queue_tasks ; 852server_response queue_results ; 853 854// note: chat_params must not be refreshed upon existing sleeping state 855server_chat_params chat_params ; 856 857server_state_callback_t callback_state = [](server_state ,json )-> void {}; 858 859server_context_impl () { 860mtmd_helper_log_set (common_log_default_callback ,nullptr ); 861 } 862 863 ~server_context_impl () { 864if (!sleeping ) { 865// destroy() is already called when entering sleeping state 866// we don't call it again here to avoid double free 867destroy (); 868 } 869 } 870 871server_metrics get_metrics ()const { 872return metrics ; 873 } 874 875void reset_metrics_bucket () { 876metrics .reset_bucket (); 877 } 878 879private : 880// note: accessing these fields outside of this class is not thread-safe 881// use server_context methods instead 882 883common_params params_base ; 884 885// note: keep these alive - they determine the lifetime of the model, context, etc. 886common_init_result_ptr llama_init ; 887 888llama_context * ctx_tgt = nullptr ; 889 890server_batch batch ; 891 892llama_model * model_dft = nullptr ; 893llama_context * ctx_dft = nullptr ; 894 895common_speculative_init_result_ptr spec_init ; 896 897common_context_seq_rm_type ctx_tgt_seq_rm_type = COMMON_CONTEXT_SEQ_RM_TYPE_NO ; 898common_context_seq_rm_type ctx_dft_seq_rm_type = COMMON_CONTEXT_SEQ_RM_TYPE_NO ; 899 900common_speculative_ptr spec ; 901 902bool add_bos_token = true; 903bool encoder_decoder = false; 904llama_tokens decoder_prefix ; 905 906int32_t n_ctx ;// total context for all clients / slots 907 908// set to llama_model_n_swa(model) 909// if swa_full is enabled, this is set to 0 to simulate a non-SWA model 910int32_t n_swa ; 911 912// slots / clients 913 std::vector < server_slot > slots ; 914 915int trace = 0 ;// env: LLAMA_TRACE 916int slots_debug = 0 ;// env: LLAMA_SERVER_SLOTS_DEBUG 917int slots_n_diff = 0 ;// env: LLAMA_SERVER_SLOTS_N_DIFF 918 919int n_empty_consecutive = 0 ; 920 921 std::unique_ptr < server_prompt_cache > prompt_cache ; 922 923server_metrics metrics ; 924 925// queued prompt stats - llama_decode() is async, so the timing is only valid after a sync 926// note: kept out of server_metrics, which is copied as-is into the task result 927int64_t t_decode_start = 0 ;// start of the last submitted decode 928int64_t t_prompt_start = 0 ;// start of the oldest queued prompt decode 929uint64_t n_prompt_queued = 0 ; 930 931json json_ui_settings = json::object (); 932 933// Necessary similarity of prompt for slot selection 934float slot_prompt_similarity = 0.0f ; 935 936 std::string model_name ;// name of the loaded model, to be used by API 937 std::set < std::string > model_aliases ;// additional names for the model 938 std::set < std::string > model_tags ;// informational tags 939 940bool sleeping = false; 941 942int64_t t_last_load_progress_ms = 0 ; 943 944void destroy () { 945spec .reset (); 946spec_init .reset (); 947 948ctx_dft = nullptr ; 949model_dft = nullptr ; 950 951llama_init .reset (); 952 953ctx_tgt = nullptr ; 954model_tgt = nullptr ; 955 956mtmd_free (mctx ); 957mctx = nullptr ; 958 } 959 960void handle_sleeping_state (bool new_state ) { 961GGML_ASSERT (sleeping != new_state ); 962if (new_state ) { 963if (callback_state ) { 964callback_state (SERVER_STATE_SLEEPING , {}); 965// note: for sleeping == false, event is emitted by load_model() 966 } 967SRV_INF ("%s" ,"server is entering sleeping state\n" ); 968destroy (); 969 }else { 970SRV_INF ("%s" ,"server is exiting sleeping state\n" ); 971if (!load_model (params_base )) { 972GGML_ABORT ("failed to reload model after sleeping" ); 973 } 974 } 975sleeping = new_state ; 976 } 977 978struct load_progress_data { 979server_context_impl * ctx ; 980 std::string stage ; 981 std::vector < std::string > stages ; 982int64_t t_last_load_progress_ms = 0 ; 983load_progress_data (server_context_impl * ctx ,const std::string & stage ) :ctx (ctx ),stage (stage ) {} 984 }; 985static bool load_progress_callback (float progress ,void * user_data ) { 986auto * d = static_cast < load_progress_data *> (user_data ); 987GGML_ASSERT (d ); 988// always emit the first and final sample; throttle the rest to one per 200ms 989 { 990auto & t_last = d -> t_last_load_progress_ms ; 991const int64_t t_now = ggml_time_ms (); 992const bool first = t_last == 0 ; 993const bool done = progress >=1.0f ; 994const bool throttled = !first && !done && (t_now - t_last )< 200 ; 995if (throttled ) { 996return true; 997 } 998t_last = t_now ; 999 } 1000if (d -> ctx -> callback_state ) { 1001d -> ctx -> callback_state (SERVER_STATE_LOADING , { 1002 {"stages" ,d -> stages }, 1003 {"current" ,d -> stage }, 1004 {"value" ,progress }, 1005 }); 1006 } 1007return true; 1008 } 1009 1010// load the model and initialize llama_context 1011// this may also be called to resume from sleeping state 1012bool load_model (common_params & params ) { 1013load_progress_data load_progress_text (this ,"text_model" ); 1014load_progress_data load_progress_mmproj (this ,"mmproj_model" ); 1015load_progress_data load_progress_spec (this ,"spec_model" ); 1016 1017const bool is_resume = sleeping ; 1018 1019params_base = params ; 1020const auto output_limits = server_output_limits (params_base ); 1021params_base .n_outputs_max = output_limits .total ; 1022params_base .n_outputs_max_per_seq = output_limits .per_seq ; 1023 1024const bool has_mmproj = !params .mmproj .path .empty (); 1025const bool has_draft = params .speculative .has_dft (); 1026const bool spec_mtp = std::find (params_base .speculative .types .begin (), 1027params_base .speculative .types .end (), 1028COMMON_SPECULATIVE_TYPE_DRAFT_MTP )!= params_base .speculative .types .end (); 1029const bool has_spec = has_draft || spec_mtp ; 1030 1031if (callback_state ) { 1032 std::vector < std::string > stages = {"text_model" }; 1033if (has_spec ) { 1034stages .push_back ("spec_model" ); 1035 } 1036if (has_mmproj ) { 1037stages .push_back ("mmproj_model" ); 1038 } 1039load_progress_text .stages = stages ; 1040load_progress_mmproj .stages = stages ; 1041load_progress_spec .stages = stages ; 1042 1043// trigger 0% progress 1044load_progress_callback (0.0f ,& load_progress_text ); 1045 } 1046 1047 1048SRV_INF ("loading model '%s'\n" ,params .model .get_name ().c_str ()); 1049SRV_TRC ("local path '%s'\n" ,params .model .path .c_str ()); 1050 1051 std::string & mmproj_path = params_base .mmproj .path ; 1052mtmd_context_params mparams = mtmd_context_params_default (); 1053if (has_mmproj ) { 1054mparams .use_gpu = params_base .mmproj_use_gpu ; 1055mparams .device = params_base .mmproj_device ; 1056mparams .print_timings = false; 1057mparams .n_threads = params_base .cpuparams .n_threads ; 1058mparams .flash_attn_type = params_base .flash_attn_type ; 1059mparams .warmup = params_base .warmup ; 1060mparams .image_min_tokens = params_base .image_min_tokens ; 1061mparams .image_max_tokens = params_base .image_max_tokens ; 1062mparams .batch_max_tokens = params_base .mtmd_batch_max_tokens ; 1063mparams .media_marker = get_media_marker (); 1064// progress callback 1065mparams .progress_callback = load_progress_callback ; 1066mparams .progress_callback_user_data = & load_progress_mmproj ; 1067 } 1068 1069// optionally get the memory usage of mmproj 1070if (has_mmproj && params_base .fit_params ) { 1071int64_t t_start = ggml_time_us (); 1072auto mmproj_mem = mtmd_get_memory_usage (mmproj_path .c_str (),mparams ); 1073int64_t t_elapsed = ggml_time_us ()- t_start ; 1074if (!mmproj_mem .empty ()) { 1075size_t total = 0 ; 1076for (auto & [dev ,size ] :mmproj_mem ) { 1077total += size ; 1078 } 1079SRV_TRC ("[mtmd] estimated worst-case memory usage of mmproj is %.2f MiB (took %.2f ms)\n" ,total / (1024.0 * 1024.0 ),t_elapsed /1000.0 ); 1080GGML_ASSERT (!params_base .fit_params_target .empty ()); 1081for (auto & [dev ,size ] :mmproj_mem ) { 1082for (size_t i = 0 ;i < ggml_backend_dev_count ();i ++ ) { 1083if (ggml_backend_dev_get (i )== dev ) { 1084if (i < params_base .fit_params_target .size ()) { 1085SRV_DBG ("[mtmd] adding %.2f MiB to fit_params_target for device %s\n" ,size / (1024.0 * 1024.0 ),ggml_backend_dev_name (dev )); 1086params_base .fit_params_target [i ]+= size ; 1087 } 1088break ; 1089 } 1090 } 1091 } 1092 }else { 1093SRV_ERR ("%s" ,"[mtmd] failed to get memory usage of mmproj\n" ); 1094 } 1095 } 1096 1097// note: the draft / MTP context is fitted together with the target model, see common_fit_extra_model 1098 1099// attach a progress callback 1100 { 1101params_base .load_progress_callback = load_progress_callback ; 1102params_base .load_progress_callback_user_data = & load_progress_text ; 1103 } 1104 1105llama_init = common_init_from_params (params_base ); 1106 1107model_tgt = llama_init -> model (); 1108ctx_tgt = llama_init -> context (); 1109 1110if (model_tgt == nullptr ) { 1111SRV_ERR ("failed to load model, '%s'\n" ,params_base .model .path .c_str ()); 1112return false; 1113 } 1114 1115if (ctx_tgt == nullptr ) { 1116SRV_ERR ("failed to create_context with model '%s'\n" ,params_base .model .path .c_str ()); 1117return false; 1118 } 1119 1120vocab = llama_model_get_vocab (model_tgt ); 1121 1122n_ctx = llama_n_ctx (ctx_tgt ); 1123 1124add_bos_token = llama_vocab_get_add_bos (vocab ); 1125encoder_decoder = llama_model_has_encoder (model_tgt )&& llama_model_has_decoder (model_tgt ); 1126if (encoder_decoder ) { 1127if (has_spec || has_mmproj || !params_base .lora_adapters .empty ()) { 1128SRV_ERR ("%s" ,"encoder-decoder serving does not support speculative decoding, multimodal input or LoRA\n" ); 1129return false; 1130 } 1131llama_token decoder_start = llama_model_decoder_start_token (model_tgt ); 1132if (decoder_start == LLAMA_TOKEN_NULL ) { 1133decoder_start = llama_vocab_bos (vocab ); 1134 } 1135if (decoder_start < 0 || decoder_start >=llama_vocab_n_tokens (vocab )) { 1136SRV_ERR ("%s" ,"encoder-decoder model has no valid decoder start token\n" ); 1137return false; 1138 } 1139decoder_prefix = {decoder_start }; 1140char arch [64 ]= {}; 1141llama_model_meta_val_str (model_tgt ,"general.architecture" ,arch ,sizeof (arch )); 1142if (std::strcmp (arch ,"aliceai_t5_moe" )== 0 ) { 1143// AliceAI's infill sentinel is decoder input, not response text. 1144auto span = common_tokenize (vocab ,"<SPAN#0>" , false, true); 1145if (span .size ()!= 1 || common_token_to_piece (vocab ,span [0 ], true)!= "<SPAN#0>" ) { 1146SRV_ERR ("%s" ,"AliceAI requires a <SPAN#0> token\n" ); 1147return false; 1148 } 1149decoder_prefix .push_back (span [0 ]); 1150 } 1151// Encoder outputs belong to the context, not individual KV sequences. 1152params_base .n_parallel = 1 ; 1153params_base .ctx_shift = false; 1154params_base .n_cache_reuse = 0 ; 1155params_base .cache_ram_mib = 0 ; 1156params_base .cache_idle_slots = false; 1157params_base .n_ctx_checkpoints = 0 ; 1158params_base .slot_save_path .clear (); 1159prompt_cache .reset (); 1160SRV_WRN ("encoder-decoder mode: one slot, no prompt cache or context shift; encoder limit = %u tokens\n" ,llama_n_ubatch (ctx_tgt )); 1161 } 1162 1163if (has_spec ) { 1164// spec_mtp doesn't use load a model internally, so we report 0.0 and 1.0 manually 1165load_progress_callback (0.0f ,& load_progress_spec ); 1166load_progress_spec .t_last_load_progress_ms = 0 ;// reset so internal cbs aren't delayed 1167 1168 { 1169common_params params_dft = common_base_params_to_speculative (params_base ); 1170 1171// progress callback 1172params_dft .load_progress_callback = load_progress_callback ; 1173params_dft .load_progress_callback_user_data = & load_progress_spec ; 1174 1175spec_init = common_speculative_init_from_params (params_dft ,model_tgt ,ctx_tgt ); 1176model_dft = spec_init -> model (); 1177ctx_dft = spec_init -> context (); 1178 1179if (has_draft && model_dft == nullptr ) { 1180SRV_ERR ("failed to load draft model, '%s'\n" ,params_dft .model .path .c_str ()); 1181return false; 1182 } 1183 1184if (ctx_dft == nullptr ) { 1185SRV_ERR ("%s" ,"failed to create MTP context\n" ); 1186return false; 1187 } 1188 1189params_base .speculative .draft .ctx_tgt = ctx_tgt ; 1190params_base .speculative .draft .ctx_dft = ctx_dft ; 1191 } 1192 1193load_progress_callback (1.0f ,& load_progress_spec ); 1194 } 1195 1196if (has_mmproj ) { 1197if (callback_state ) { 1198callback_state (SERVER_STATE_LOADING , {{"stage" ,"mmproj_model" }}); 1199 } 1200 1201if (!is_resume ) { 1202mtmd_helper_log_set (common_log_default_callback ,nullptr ); 1203 } 1204 1205mctx = mtmd_init_from_file (mmproj_path .c_str (),model_tgt ,mparams ); 1206if (mctx == nullptr ) { 1207SRV_ERR ("failed to load multimodal model, '%s'\n" ,mmproj_path .c_str ()); 1208return false; 1209 } 1210SRV_INF ("loaded multimodal model, '%s'\n" ,mmproj_path .c_str ()); 1211 1212init_opt .video_params .fps_target = params_base .video_fps ; 1213init_opt .video_params .timestamp_interval_ms = params_base .video_timestamp_interval_ms ; 1214init_opt .video_params .ffmpeg_bin_dir = params_base .video_ffmpeg_bin_dir .empty () 1215 ?nullptr :params_base .video_ffmpeg_bin_dir .c_str (); 1216 1217if (params_base .ctx_shift ) { 1218params_base .ctx_shift = false; 1219SRV_WRN ("%s\n" ,"ctx_shift is not supported by multimodal, it will be disabled" ); 1220 } 1221 1222if (params_base .n_cache_reuse ) { 1223params_base .n_cache_reuse = 0 ; 1224SRV_WRN ("%s\n" ,"cache_reuse is not supported by multimodal, it will be disabled" ); 1225 } 1226 } 1227 1228if (!llama_memory_can_shift (llama_get_memory (ctx_tgt ))) { 1229if (params_base .ctx_shift ) { 1230params_base .ctx_shift = false; 1231SRV_WRN ("%s\n" ,"ctx_shift is not supported by this context, it will be disabled" ); 1232 } 1233 1234if (params_base .n_cache_reuse ) { 1235params_base .n_cache_reuse = 0 ; 1236SRV_WRN ("%s\n" ,"cache_reuse is not supported by this context, it will be disabled" ); 1237 } 1238 } 1239 1240if (llama_model_n_swa (model_tgt )== 0 ) { 1241if (params_base .swa_full ) { 1242params_base .swa_full = false; 1243SRV_WRN ("%s\n" ,"swa_full is not supported by this model, it will be disabled" ); 1244 } 1245 } 1246 1247n_swa = params_base .swa_full ?0 :llama_model_n_swa (model_tgt ); 1248 1249// Necessary similarity of prompt for slot selection 1250slot_prompt_similarity = params_base .slot_prompt_similarity ; 1251 1252const int n_ctx_train = llama_model_n_ctx_train (model_tgt ); 1253 1254 { 1255// note: the capping itself is done in n_ctx_slot(), here we only report it 1256const int n_ctx_seq = llama_n_ctx_seq (ctx_tgt ); 1257 1258if (params_base .kv_unified_per_slot > 0 ) { 1259if (n_ctx_seq > params_base .kv_unified_per_slot ) { 1260SRV_INF ("capping per-slot context (%d) to --kv-unified-per-slot (%d)\n" , 1261n_ctx_seq ,params_base .kv_unified_per_slot ); 1262 }else if (params_base .kv_unified_per_slot > n_ctx_seq ) { 1263// cap is above the per-slot pool capacity, so it can never bind 1264SRV_WRN ( 1265"--kv-unified-per-slot (%d) exceeds the per-slot pool capacity (%d) - cap has no effect, " 1266"slots are limited to %d (raise the KV pool with -c, or unset -c to size it to " 1267"n_parallel * kv_unified_per_slot)\n" , 1268params_base .kv_unified_per_slot ,n_ctx_seq ,n_ctx_seq ); 1269 } 1270 } 1271 1272const int n_ctx_capped = params_base .kv_unified_per_slot > 0 ? 1273 std::min (n_ctx_seq ,params_base .kv_unified_per_slot ) :n_ctx_seq ; 1274 1275if (n_ctx_capped > n_ctx_train ) { 1276SRV_WRN ("the slot context (%d) exceeds the training context of the model (%d) - capping\n" , 1277n_ctx_capped ,n_ctx_train ); 1278 } 1279 } 1280 1281slots .clear (); 1282 1283ctx_tgt_seq_rm_type = encoder_decoder ?COMMON_CONTEXT_SEQ_RM_TYPE_NO :common_context_can_seq_rm (ctx_tgt ); 1284if (ctx_tgt_seq_rm_type == COMMON_CONTEXT_SEQ_RM_TYPE_NO ) { 1285SRV_WRN ("%s" ,"speculative decoding not supported by this context\n" ); 1286 } 1287 1288if (ctx_tgt_seq_rm_type == COMMON_CONTEXT_SEQ_RM_TYPE_FULL ) { 1289SRV_TRC ("%s" ,"speculative decoding will use checkpoints\n" ); 1290 } 1291 1292// setup slots 1293SRV_INF ("initializing, n_slots = %d, n_ctx_slot = %d, kv_unified = '%s'\n" , 1294params_base .n_parallel ,n_ctx_slot (),params_base .kv_unified ?"true" :"false" ); 1295 1296// initialize slots 1297for (int i = 0 ;i < params_base .n_parallel ;i ++ ) { 1298slots .emplace_back (); 1299 } 1300 1301// try speculative decoding 1302if (ctx_tgt_seq_rm_type != COMMON_CONTEXT_SEQ_RM_TYPE_NO ) { 1303try { 1304spec .reset (common_speculative_init (params_base .speculative ,params_base .n_parallel )); 1305 }catch (const std::exception & e ) { 1306SRV_ERR ("failed to initialize speculative decoding context: %s\n" ,e .what ()); 1307if (params_base .speculative .has_synth ()) { 1308return false; 1309 } 1310 } 1311 } 1312 1313if (ctx_dft ) { 1314ctx_dft_seq_rm_type = common_context_can_seq_rm (ctx_dft ); 1315 } 1316 1317if (spec ) { 1318SRV_TRC ("%s" ,"speculative decoding context initialized\n" ); 1319 }else { 1320spec_init .reset (); 1321ctx_dft = nullptr ; 1322model_dft = nullptr ; 1323 } 1324 1325if (!spec && params_base .speculative .has_synth ()) { 1326SRV_ERR ("%s" ,"synthetic acceptance requires an initialized speculative decoding context\n" ); 1327return false; 1328 } 1329 1330for (int i = 0 ;i < params_base .n_parallel ;i ++ ) { 1331server_slot & slot = slots [i ]; 1332 1333slot .id = i ; 1334slot .ctx_tgt = ctx_tgt ; 1335slot .ctx_dft = ctx_dft ; 1336slot .mem .init (ctx_tgt ,ctx_dft ); 1337slot .spec = spec .get (); 1338slot .n_ctx = n_ctx_slot (); 1339 1340slot .mctx = mctx ; 1341slot .prompt .tokens .has_mtmd = mctx != nullptr ; 1342 1343SLT_TRC (slot ,"new slot, n_ctx = %d\n" ,slot .n_ctx ); 1344 1345slot .callback_on_release = [this ](int id_slot ) { 1346queue_tasks .pop_deferred_task (id_slot ); 1347 }; 1348 1349slot .callback_on_reset = [this ](const server_slot & slot ) { 1350// flush the generated token stats before reset() 1351if (slot .stats .n_gen > 0 ) { 1352metrics_on_prediction (slot ); 1353 } 1354 }; 1355 1356slot .reset (); 1357 } 1358 1359 { 1360const char * LLAMA_TRACE = getenv ("LLAMA_TRACE" ); 1361trace = LLAMA_TRACE ?atoi (LLAMA_TRACE ) :0 ; 1362 1363if (trace ) { 1364SRV_WRN ("LLAMA_TRACE = %d\n" ,trace ); 1365 } 1366 } 1367 1368 { 1369const char * LLAMA_SERVER_SLOTS_DEBUG = getenv ("LLAMA_SERVER_SLOTS_DEBUG" ); 1370slots_debug = LLAMA_SERVER_SLOTS_DEBUG ?atoi (LLAMA_SERVER_SLOTS_DEBUG ) :0 ; 1371 1372if (slots_debug ) { 1373SRV_WRN ("LLAMA_SERVER_SLOTS_DEBUG = %d\n" ,slots_debug ); 1374 } 1375 } 1376 1377 { 1378const char * LLAMA_SERVER_SLOTS_N_DIFF = getenv ("LLAMA_SERVER_SLOTS_N_DIFF" ); 1379slots_n_diff = LLAMA_SERVER_SLOTS_N_DIFF ?atoi (LLAMA_SERVER_SLOTS_N_DIFF ) :0 ; 1380 1381if (slots_n_diff ) { 1382SRV_WRN ("LLAMA_SERVER_SLOTS_N_DIFF = %d\n" ,slots_n_diff ); 1383 } 1384 } 1385 1386// the update_slots() logic will always submit a maximum of n_batch or n_parallel tokens 1387// note that n_batch can be > n_ctx (e.g. for non-causal attention models such as BERT where the KV cache is not used) 1388 { 1389const int32_t n_batch = llama_n_batch (ctx_tgt ); 1390const int32_t n_embd = llama_model_n_embd_inp (model_tgt ); 1391batch .init (std::max (n_batch ,params_base .n_parallel ),n_embd ); 1392 } 1393 1394if (params_base .cache_ram_mib != 0 ) { 1395if (params_base .cache_ram_mib < 0 ) { 1396SRV_TRC ("prompt cache is enabled, size limit: %s\n" ,"no limit" ); 1397 }else { 1398SRV_TRC ("prompt cache is enabled, size limit: %d MiB\n" ,params_base .cache_ram_mib ); 1399 } 1400SRV_TRC ("%s" ,"use `--cache-ram 0` to disable the prompt cache\n" ); 1401 1402prompt_cache = std::make_unique < server_prompt_cache > (params_base .cache_ram_mib ,n_ctx ); 1403 }else { 1404SRV_TRC ("%s" ,"prompt cache is disabled - use `--cache-ram N` to enable it\n" ); 1405 } 1406SRV_TRC ("%s" ,"for more info see https://github.com/ggml-org/llama.cpp/pull/16391\n" ); 1407 1408if (params_base .n_ctx_checkpoints > 0 ) { 1409SRV_TRC ("context checkpoints enabled, max = %d, min spacing = %d\n" , 1410params_base .n_ctx_checkpoints ,params_base .checkpoint_min_step ); 1411 }else { 1412SRV_TRC ("%s" ,"context checkpoints disabled\n" ); 1413 } 1414 1415if (!params_base .model_alias .empty ()) { 1416// backward compat: use first alias as model name 1417model_name = * params_base .model_alias .begin (); 1418 }else if (!params_base .model .get_name ().empty ()) { 1419model_name = params_base .model .get_name (); 1420 }else { 1421// fallback: derive model name from file name 1422auto model_path = std::filesystem::path (params_base .model .path ); 1423model_name = model_path .filename ().string (); 1424 } 1425 1426model_aliases = params_base .model_alias ; 1427model_tags = params_base .model_tags ; 1428 1429// propagate new defaults back to caller 1430params = params_base ; 1431 1432if (!is_resume ) { 1433return init (); 1434 } 1435 1436if (callback_state ) { 1437callback_state (SERVER_STATE_READY , {}); 1438 } 1439 1440return true; 1441 } 1442 1443// unlike load_model(), this is only called once during initialization 1444bool init () { 1445GGML_ASSERT (ctx_tgt != nullptr ); 1446GGML_ASSERT (model_tgt != nullptr ); 1447 1448GGML_ASSERT (!sleeping ); 1449 1450// wiring up server queues 1451queue_tasks .on_new_task ([this ](server_task && task ,bool is_yielding ) { 1452return process_single_task (std::move (task ),is_yielding ); 1453 }); 1454queue_tasks .on_update_slots ([this ]() { 1455update_slots (); 1456 }); 1457queue_tasks .on_sleeping_state ([this ](bool sleeping ) { 1458handle_sleeping_state (sleeping ); 1459 }); 1460 1461metrics .init (); 1462 1463if (params_base .cache_idle_slots ) { 1464if (params_base .cache_ram_mib == 0 ) { 1465SRV_WRN ("%s" ,"--cache-idle-slots requires --cache-ram, disabling\n" ); 1466params_base .cache_idle_slots = false; 1467 }else { 1468if (params_base .kv_unified ) { 1469SRV_TRC ("%s" ,"idle slots will be saved to prompt cache and cleared upon starting a new task\n" ); 1470 }else { 1471// without a unified KV cache, clearing a slot frees no reusable room, so we only 1472// publish a RAM-cache copy of idle slots (their KV stays in VRAM) [TAG_IDLE_SLOT_CLEAR] 1473SRV_TRC ("%s" ,"idle slots will be saved to prompt cache upon starting a new task\n" ); 1474 } 1475SRV_DBG ("%s" ,"__TEST_TAG_CACHE_IDLE_SLOTS_ENABLED__\n" ); 1476 } 1477 } 1478 1479 { 1480const std::string & cfg = params_base .ui_config_json ; 1481if (!cfg .empty ()) { 1482try { 1483json json_settings = json::parse (cfg ); 1484json_ui_settings = json_settings ; 1485 }catch (const std::exception & e ) { 1486SRV_ERR ("%s: failed to parse UI config: %s\n" ,__func__ ,e .what ()); 1487return false; 1488 } 1489 } 1490 } 1491 1492// populate chat template params 1493 { 1494common_chat_templates_ptr chat_templates ; 1495bool enable_thinking = false; 1496 1497try { 1498chat_templates = common_chat_templates_init (model_tgt ,params_base .chat_template ); 1499 1500SRV_TRC ("%s: chat template, example_format: '%s'\n" ,__func__ , 1501common_chat_format_example (chat_templates .get (),params_base .use_jinja ,params_base .default_template_kwargs ).c_str ()); 1502 1503// thinking is enabled if: 1504// 1. It's not explicitly disabled via --reasoning off 1505// 2. The chat template supports it 1506const bool template_supports_thinking = params_base .use_jinja && common_chat_templates_support_enable_thinking (chat_templates .get ()); 1507enable_thinking = params_base .enable_reasoning != 0 && template_supports_thinking ; 1508SRV_TRC ("%s: chat template, thinking = %d\n" ,__func__ ,enable_thinking ); 1509 }catch (const std::exception & e ) { 1510SRV_ERR ("%s: chat template parsing error: %s\n" ,__func__ ,e .what ()); 1511SRV_ERR ("%s: please consider disabling jinja via --no-jinja, or use a custom chat template via --chat-template\n" ,__func__ ); 1512SRV_ERR ("%s: for example: --no-jinja --chat-template chatml\n" ,__func__ ); 1513return false; 1514 } 1515 1516// IMPORTANT: chat_params is reused across sleeping / resuming states, 1517// never store llama_context/llama_model pointers in chat_params, 1518// as they may be invalidated after sleeping 1519chat_params = { 1520/* use_jinja */ params_base .use_jinja , 1521/* prefill_assistant */ params_base .prefill_assistant , 1522/* reasoning_format */ params_base .reasoning_format , 1523/* chat_template_kwargs */ params_base .default_template_kwargs , 1524/* tmpls */ std::move (chat_templates ), 1525/* allow_image */ mctx ?mtmd_support_vision (mctx ) : false, 1526/* allow_audio */ mctx ?mtmd_support_audio (mctx ) : false, 1527/* allow_video */ mctx ?mtmd_helper_support_video (mctx ) : false, 1528/* enable_thinking */ enable_thinking , 1529/* reasoning_budget */ params_base .sampling .reasoning_budget_tokens , 1530/* reasoning_budget_msg */ params_base .sampling .reasoning_budget_message , 1531/* media_path */ params_base .media_path , 1532/* force_pure_content */ params_base .force_pure_content_parser 1533 }; 1534 1535 { 1536auto caps = common_chat_templates_get_caps (chat_params .tmpls .get ()); 1537auto it = params_base .default_template_kwargs .find ("preserve_reasoning" ); 1538bool supported = caps .at ("supports_preserve_reasoning" ); 1539bool specified = params_base .preserve_reasoning_specified ; 1540// note: the kwarg is enabled by default if not specified explicitly, so check the value 1541bool enabled = it != params_base .default_template_kwargs .end ()&& it -> second == "true" ; 1542if (supported ) { 1543SRV_TRC ("preserve_reasoning kwarg: %s\n" , 1544it == params_base .default_template_kwargs .end () ?"unset (template default)" :it -> second .c_str ()); 1545 }else { 1546SRV_TRC ("%s" ,"preserve_reasoning kwarg: not supported by template\n" ); 1547 } 1548if (supported && !specified ) { 1549SRV_WRN ("%s" ,"chat template supports preserving reasoning, it is enabled by default (may use more tokens, disable via --no-reasoning-preserve)\n" ); 1550 } 1551if (supported && !enabled ) { 1552SRV_INF ("%s" ,"chat template supports preserving reasoning, consider enabling it via --reasoning-preserve\n" ); 1553 } 1554if (!supported && specified && enabled ) { 1555SRV_WRN ("%s" ,"chat template does NOT support preserving reasoning, --reasoning-preserve has no effect\n" ); 1556 } 1557 } 1558 } 1559 1560return true; 1561 } 1562 1563server_slot * get_slot_by_id (int id_slot ) { 1564// note: allow id_slot to be out of bounds (wrap around) 1565id_slot = id_slot %slots .size (); 1566 1567for (server_slot & slot :slots ) { 1568if (slot .id == id_slot ) { 1569return & slot ; 1570 } 1571 } 1572 1573return nullptr ; 1574 } 1575 1576server_slot * get_slot_by_cmpl_id (const std::string & cmpl_id ) { 1577if (cmpl_id .empty ()) { 1578return nullptr ; 1579 } 1580 1581for (server_slot & slot :slots ) { 1582if (slot .is_processing ()&& slot .task && slot .task -> params .oaicompat_cmpl_id == cmpl_id ) { 1583return & slot ; 1584 } 1585 } 1586 1587return nullptr ; 1588 } 1589 1590server_slot * get_available_slot (const server_task & task ) { 1591server_slot * ret = nullptr ; 1592 1593bool update_cache = false; 1594 1595// if a specific slot is requested, use it (still goes through cache update logic below) 1596if (task .id_slot != -1 ) { 1597ret = get_slot_by_id (task .id_slot ); 1598if (ret ) { 1599SLT_INF (* ret ,"selected slot by id (%d)\n" ,task .id_slot ); 1600 } 1601 } 1602 1603// find the slot that has at least n% prompt similarity 1604if (slot_prompt_similarity != 0.0f ) { 1605float f_sim_best = 0 ; 1606 1607for (server_slot & slot :slots ) { 1608if (task .id_slot != -1 && slot .id != task .id_slot ) { 1609continue ; 1610 } 1611 1612// skip the slot if it is not available 1613if (slot .is_processing ()) { 1614SLT_TRC (slot ," - skipping, is_processing = %d\n" ,slot .is_processing ()); 1615continue ; 1616 } 1617 1618const auto & tokens = slot .prompt .tokens ; 1619 1620// skip the slot if it does not contains cached tokens 1621if (tokens .empty ()) { 1622SLT_TRC (slot ,"%s" ," - skipping, slot is empty\n" ); 1623continue ; 1624 } 1625 1626// fraction of the Longest Common Prefix length with respect to the input prompt length 1627const size_t lcp_len = tokens .get_common_prefix (task .tokens ); 1628const float f_sim_cur = float (lcp_len ) /task .tokens .size (); 1629 1630SLT_TRC (slot ," - checking sim = %.3f (%zu/%zu) > %.3f\n" ,f_sim_cur ,lcp_len ,task .tokens .size (),slot_prompt_similarity ); 1631 1632// select the current slot if the criteria match 1633if (f_sim_cur > f_sim_best && f_sim_cur > slot_prompt_similarity ) { 1634f_sim_best = f_sim_cur ; 1635 1636ret = & slot ; 1637 } 1638 } 1639 1640if (ret != nullptr ) { 1641const float f_keep = (f_sim_best * task .tokens .size ()) /ret -> prompt .tokens .size (); 1642 1643if (task .id_slot == -1 ) { 1644SLT_INF (* ret ,"selected slot by LCP similarity, f_sim_best = %.3f (> %.3f thold), f_keep = %.3f\n" , 1645f_sim_best ,slot_prompt_similarity ,f_keep ); 1646 } 1647 1648// if we are about to lose a large portion of the existing context - save it in the prompt cache 1649if (f_keep < 0.5f ) { 1650update_cache = true; 1651 } 1652 } 1653 } 1654 1655// find the slot that has been least recently used 1656if (ret == nullptr ) { 1657int64_t t_last = -1 ; 1658 1659for (server_slot & slot :slots ) { 1660// skip the slot if it is not available 1661if (slot .is_processing ()) { 1662continue ; 1663 } 1664 1665// select the current slot if the criteria match 1666if (!ret || slot .t_last_used <=t_last ) { 1667t_last = slot .t_last_used ; 1668ret = & slot ; 1669 } 1670 } 1671 1672if (ret != nullptr ) { 1673SLT_INF (* ret ,"selected slot by LRU, t_last = %" PRId64 "\n" ,t_last ); 1674 1675update_cache = true; 1676 } 1677 } 1678 1679if (ret ) { 1680update_cache = update_cache && prompt_cache ; 1681 1682// cache prompts only for completion tasks 1683update_cache = update_cache && task .type == SERVER_TASK_TYPE_COMPLETION ; 1684 1685if (update_cache ) { 1686SRV_TRC ("%s" ,"updating prompt cache\n" ); 1687 1688const int64_t t_start = ggml_time_us (); 1689 1690ret -> prompt_save (* prompt_cache ); 1691 1692if (!ret -> prompt_load (* prompt_cache ,task .tokens )) { 1693ret -> prompt_clear (); 1694 } 1695 1696prompt_cache -> update (); 1697 1698SRV_TRC ("prompt cache update took %.2f ms\n" , (ggml_time_us ()- t_start ) /1000.0 ); 1699 } 1700 } 1701 1702return ret ; 1703 } 1704 1705// return true if at least one slot has been cleared 1706// TODO: improve logic 1707// - smarter decision which slot to clear (LRU or longest prompt?) 1708// - move slot to level 2 cache instead of removing? 1709// - instead of purging, try to store and resume later? 1710bool try_clear_idle_slots () { 1711bool res = false; 1712 1713if (!params_base .kv_unified ) { 1714return res ; 1715 } 1716 1717for (auto & slot :slots ) { 1718if (slot .is_processing ()) { 1719continue ; 1720 } 1721 1722if (slot .prompt .n_tokens ()> 0 ) { 1723SRV_WRN ("purging slot %d with %zu tokens\n" ,slot .id ,slot .prompt .tokens .size ()); 1724 1725slot .prompt_clear (); 1726 1727res = true; 1728 1729// clear slots one by one 1730break ; 1731 } 1732 } 1733 1734return res ; 1735 } 1736 1737 std::vector < common_adapter_lora_info > construct_lora_list (const std::map < int ,float > & config )const { 1738 std::vector < common_adapter_lora_info > output = params_base .lora_adapters ;// copy 1739for (size_t i = 0 ;i < output .size ();++ i ) { 1740auto it = config .find (i ); 1741if (it != config .end ()) { 1742output [i ].scale = it -> second ; 1743 }else { 1744output [i ].scale = 0.0f ; 1745 } 1746 } 1747return output ; 1748 } 1749 1750bool launch_slot_with_task (server_slot & slot ,server_task && task ) { 1751if (encoder_decoder ) { 1752if (!task .need_sampling ()|| task .is_parent ()|| task .is_child ()|| !task .params .lora .empty ()) { 1753send_error (task ,"Encoder-decoder serving supports text generation with one completion and no LoRA" ,ERROR_TYPE_NOT_SUPPORTED ); 1754return false; 1755 } 1756task .params .cache_prompt = false; 1757task .params .n_cache_reuse = 0 ; 1758task .params .sampling .backend_sampling = false; 1759 } 1760 1761// process per-request lora adapters 1762if (!task .params .lora .empty ()) { 1763auto task_loras = construct_lora_list (task .params .lora ); 1764if (!are_lora_equal (task_loras ,slot .lora )) { 1765// if lora has changed, check to see if the cache should be cleared 1766if (lora_should_clear_cache (slot .lora ,task_loras )) { 1767SLT_TRC (slot ,"clearing cache for lora change. %zu loras -> %zu loras\n" ,slot .lora .size (),task .params .lora .size ()); 1768slot .prompt .clear (); 1769 }else { 1770SLT_TRC (slot ,"keeping cache for alora. %zu target loras\n" ,task_loras .size ()); 1771 } 1772slot .lora = task_loras ; 1773 } 1774 }else { 1775slot .lora = params_base .lora_adapters ; 1776 } 1777 1778// if using alora, make sure it's only a single one requested and active 1779size_t alora_invocation_start = task .tokens .size (); 1780if (lora_all_alora (slot .lora )) { 1781const auto & enabled_ids = lora_get_enabled_ids (slot .lora ); 1782// TODO: This will error out if a user requests two aloras, but only 1783// provides the activation string for one. We could, instead search 1784// for all requested alora activation strings and then either keep 1785// only the last one, or reject if multiple are found. 1786if (enabled_ids .size ()!= 1 ) { 1787send_error (task ,"Cannot run multiple aLoRAs in a single request" ,ERROR_TYPE_INVALID_REQUEST ); 1788return false; 1789 } 1790const auto & lora = slot .lora [enabled_ids [0 ]].ptr ; 1791 1792// get the pointer and count for the invocation tokens 1793const uint64_t n_invocation_tokens = llama_adapter_get_alora_n_invocation_tokens (lora ); 1794const llama_token * invocation_tokens = llama_adapter_get_alora_invocation_tokens (lora ); 1795 1796// scan backwards through the prompt tokens to find the last 1797// occurrence of the invocation sequence 1798int match_idx = static_cast < int > (n_invocation_tokens )- 1 ; 1799for (int i = task .tokens .size ()- 1 ;i >=0 ;-- i ) { 1800// the token in this position matches the next token to find in 1801// the invocation sequence 1802if (task .tokens [i ]== invocation_tokens [match_idx ]) { 1803// if it's a full match, we've found the start 1804if (match_idx == 0 ) { 1805alora_invocation_start = i ; 1806break ; 1807 } 1808// otherwise, check the next token in the sequence 1809-- match_idx ; 1810 }else { 1811// no match in this position, so start looking over again 1812match_idx = static_cast < int > (n_invocation_tokens )- 1 ; 1813 } 1814 } 1815 1816// if the activation string is not found, disable the alora 1817if (alora_invocation_start == task .tokens .size ()) { 1818SLT_DBG (slot ,"alora %zu requested, but not found. deactivating\n" ,enabled_ids [0 ]); 1819slot .lora [enabled_ids [0 ]].scale = 0.0f ; 1820 }else { 1821SLT_DBG (slot ,"alora %zu activated starting at %zu\n" ,enabled_ids [0 ],alora_invocation_start ); 1822slot .alora_invocation_start = alora_invocation_start ; 1823 } 1824 } 1825 1826if (!task .tokens .validate (ctx_tgt )) { 1827send_error (task ,"Prompt contains invalid tokens" ,ERROR_TYPE_INVALID_REQUEST ); 1828return false; 1829 } 1830 1831SLT_DBG (slot ,"launching slot : %s\n" ,safe_json_to_str (slot .to_json ()).c_str ()); 1832 1833// initialize samplers 1834if (task .need_sampling ()) { 1835try { 1836slot .smpl .reset (common_sampler_init (model_tgt ,task .params .sampling )); 1837 }catch (std::exception & e ) { 1838 std::string err_msg = std::string ("Failed to initialize samplers: " )+ e .what (); 1839send_error (task ,err_msg ,ERROR_TYPE_INVALID_REQUEST ); 1840return false; 1841 } 1842 1843const bool need_pre_sample_logits = task .params .sampling .n_probs > 0 && !task .params .post_sampling_probs ; 1844 1845bool use_backend_sampling = task .params .sampling .backend_sampling ; 1846 1847// TODO: getting pre sampling logits is not yet supported with backend sampling 1848use_backend_sampling &= !need_pre_sample_logits ; 1849 1850// TODO: tmp until backend sampling is fully implemented 1851if (use_backend_sampling ) { 1852llama_set_sampler (ctx_tgt ,slot .id ,common_sampler_get (slot .smpl .get ())); 1853 }else { 1854llama_set_sampler (ctx_tgt ,slot .id ,nullptr ); 1855 } 1856 1857SLT_TRC (slot ,"sampler chain: %s\n" ,common_sampler_print (slot .smpl .get ()).c_str ()); 1858SLT_TRC (slot ,"sampler params: \n%s\n" ,task .params .sampling .c_str ()); 1859 1860if (spec && !common_speculative_get_synth_probs (spec .get ()).empty ()) { 1861const uint32_t seed = task .params .sampling .seed == LLAMA_DEFAULT_SEED 1862 ? std::random_device {}() 1863 :task .params .sampling .seed ; 1864slot .spec_synth_rng .seed (seed ); 1865 } 1866 }else { 1867slot .smpl .reset (); 1868 } 1869 1870// the per-request limit takes priority over the global one 1871slot .n_predict_max = task .params .n_predict != -1 ?task .params .n_predict :params_base .n_predict ; 1872 1873slot .task = std::make_unique < const server_task > (std::move (task )); 1874 1875slot .state = slot .task -> is_child () 1876 ?SLOT_STATE_WAIT_OTHER // wait for the parent to process prompt 1877 :SLOT_STATE_STARTED ; 1878 1879// reset server kill-switch counter 1880n_empty_consecutive = 0 ; 1881 1882SLT_INF (slot ,"processing task, is_child = %d\n" ,slot .task -> is_child ()); 1883return true; 1884 } 1885 1886bool process_token (completion_token_output & result ,server_slot & slot ) { 1887// remember which tokens were sampled - used for repetition penalties during sampling 1888const std::string token_str = result .text_to_send ; 1889slot .sampled = result .tok ; 1890 1891slot .generated_text += token_str ; 1892if (slot .task -> params .return_tokens ) { 1893slot .generated_tokens .push_back (result .tok ); 1894 } 1895slot .has_next_token = true; 1896 1897// check if there is incomplete UTF-8 character at the end 1898bool incomplete = validate_utf8 (slot .generated_text )< slot .generated_text .size (); 1899 1900// search stop word and delete it 1901if (!incomplete ) { 1902size_t pos = std::min (slot .n_sent_text ,slot .generated_text .size ()); 1903 1904const std::string str_test = slot .generated_text .substr (pos ); 1905bool send_text = true; 1906 1907size_t stop_pos = slot .find_stopping_strings (str_test ,token_str .size (), true); 1908if (stop_pos != std::string::npos ) { 1909slot .generated_text .erase ( 1910slot .generated_text .begin ()+ pos + stop_pos , 1911slot .generated_text .end ()); 1912pos = std::min (slot .n_sent_text ,slot .generated_text .size ()); 1913 }else if (slot .has_next_token && !llama_vocab_is_eog (vocab ,result .tok ) ) { 1914stop_pos = slot .find_stopping_strings (str_test ,token_str .size (), false); 1915send_text = stop_pos == std::string::npos ; 1916 } 1917 1918// check if there is any token to predict 1919if (send_text ) { 1920// no send the stop word in the response 1921result .text_to_send = slot .generated_text .substr (pos , std::string::npos ); 1922slot .n_sent_text += result .text_to_send .size (); 1923// add the token to slot queue and cache 1924 }else { 1925result .text_to_send = "" ; 1926 } 1927 1928slot .add_token (result ); 1929if (slot .task -> params .stream ) { 1930send_partial_response (slot ,result , false); 1931 } 1932 } 1933 1934if (incomplete ) { 1935slot .has_next_token = true; 1936 } 1937 1938// if context shifting is disabled, make sure that we don't run out of context 1939if (!params_base .ctx_shift && slot .prompt .n_tokens ()+ 1 >=slot .n_ctx ) { 1940slot .truncated = true; 1941slot .stop = STOP_TYPE_LIMIT ; 1942slot .has_next_token = false; 1943 1944SLT_DBG (slot ,"stopped due to running out of context capacity, prompt.n_tokens() = %d, task.n_tokens = %d, n_gen = %d, n_ctx = %d\n" , 1945slot .prompt .n_tokens (),slot .task -> n_tokens (), (int )slot .stats .n_gen ,slot .n_ctx ); 1946 } 1947 1948// check the limits 1949if (slot .stats .n_gen > 0 && slot .has_next_token && !slot .has_budget ()) { 1950slot .stop = STOP_TYPE_LIMIT ; 1951slot .has_next_token = false; 1952 1953SLT_DBG (slot ,"stopped by limit, n_gen = %d, n_predict = %d\n" , (int )slot .stats .n_gen ,slot .task -> params .n_predict ); 1954 } 1955 1956if (slot .has_new_line ) { 1957// require that each new line has a whitespace prefix (i.e. indentation) of at least slot.params.n_indent 1958if (slot .task -> params .n_indent > 0 ) { 1959// check the current indentation 1960// TODO: improve by not doing it more than once for each new line 1961if (slot .last_nl_pos > 0 ) { 1962size_t pos = slot .last_nl_pos ; 1963 1964int n_indent = 0 ; 1965while (pos < slot .generated_text .size ()&& (slot .generated_text [pos ]== ' ' || slot .generated_text [pos ]== '\t' )) { 1966n_indent ++ ; 1967pos ++ ; 1968 } 1969 1970if (pos < slot .generated_text .size ()&& n_indent < slot .task -> params .n_indent ) { 1971slot .stop = STOP_TYPE_LIMIT ; 1972slot .has_next_token = false; 1973 1974// cut the last line 1975slot .generated_text .erase (pos , std::string::npos ); 1976 1977SLT_DBG (slot ,"stopped by indentation limit, n_gen = %d, n_indent = %d\n" , (int )slot .stats .n_gen ,n_indent ); 1978 } 1979 } 1980 1981// find the next new line 1982 { 1983const size_t pos = slot .generated_text .find ('\n' ,slot .last_nl_pos ); 1984 1985if (pos != std::string::npos ) { 1986slot .last_nl_pos = pos + 1 ; 1987 } 1988 } 1989 } 1990 } 1991 1992// check if there is a new line in the generated text 1993if (result .text_to_send .find ('\n' )!= std::string::npos ) { 1994slot .has_new_line = true; 1995 1996// if we have seen a new line, we stop after a certain time limit, but only upon another new line 1997if (slot .task -> params .t_max_predict_ms > 0 && slot .stats .t_gen_ms ()> slot .task -> params .t_max_predict_ms ) { 1998slot .stop = STOP_TYPE_LIMIT ; 1999slot .has_next_token = false; 2000 2001SLT_DBG (slot ,"stopped by time limit, n_gen = %d, t_max_predict_ms = %d ms\n" , (int )slot .stats .n_gen , (int )slot .task -> params .t_max_predict_ms ); 2002 } 2003 } 2004 2005if (llama_vocab_is_eog (vocab ,result .tok )) { 2006slot .stop = STOP_TYPE_EOS ; 2007slot .has_next_token = false; 2008 2009SLT_DBG (slot ,"%s" ,"stopped by EOS\n" ); 2010 } 2011 2012SLT_DBG (slot ,"n_gen = %d, n_remaining = %d, next token: %5d '%s'\n" , (int )slot .stats .n_gen ,slot .n_remaining (),result .tok ,token_str .c_str ()); 2013 2014return slot .has_next_token ;// continue 2015 } 2016 2017void populate_token_probs (const server_slot & slot ,completion_token_output & result ,bool post_sampling ,bool special ,int idx )const { 2018const size_t n_probs_request = slot .task -> params .sampling .n_probs ; 2019 2020if (post_sampling ) { 2021const auto * cur_p = common_sampler_get_candidates (slot .smpl .get (), true); 2022const size_t max_probs = cur_p -> size ; 2023const size_t n_probs = std::min (max_probs ,n_probs_request ); 2024 2025// set probability for sampled token 2026for (size_t i = 0 ;i < max_probs ;i ++ ) { 2027if (cur_p -> data [i ].id == result .tok ) { 2028result .prob = cur_p -> data [i ].p ; 2029break ; 2030 } 2031 } 2032 2033// set probability for top n_probs tokens 2034result .probs .reserve (n_probs ); 2035for (size_t i = 0 ;i < n_probs ;i ++ ) { 2036// Some samplers do return 0.0 probabilities, others don't. 2037// Filter 0.0 probailities, to ensure the behavior is consistent. 2038if (cur_p -> data [i ].p == 0.0 ) { 2039break ; 2040 } 2041 2042result .probs .push_back ({ 2043cur_p -> data [i ].id , 2044common_token_to_piece (ctx_tgt ,cur_p -> data [i ].id ,special ), 2045cur_p -> data [i ].p 2046 }); 2047 } 2048 }else { 2049 std::vector < llama_token_data > cur = get_token_probabilities (ctx_tgt ,idx ,n_probs_request ); 2050const size_t max_probs = cur .size (); 2051const size_t n_probs = std::min (max_probs ,n_probs_request ); 2052 2053// set probability for sampled token 2054for (size_t i = 0 ;i < max_probs ;i ++ ) { 2055// set probability for sampled token 2056if (cur [i ].id == result .tok ) { 2057result .prob = cur [i ].p ; 2058break ; 2059 } 2060 } 2061 2062// set probability for top n_probs tokens 2063result .probs .reserve (n_probs ); 2064for (size_t i = 0 ;i < n_probs ;i ++ ) { 2065result .probs .push_back ({ 2066cur [i ].id , 2067common_token_to_piece (ctx_tgt ,cur [i ].id ,special ), 2068cur [i ].p 2069 }); 2070 } 2071 } 2072 } 2073 2074void send_error (const server_task & task ,const std::string & error ,const enum error_type type = ERROR_TYPE_SERVER ) { 2075send_error (task .id ,error ,type ); 2076 } 2077 2078void send_error (const server_slot & slot ,const std::string & error ,const enum error_type type = ERROR_TYPE_SERVER ) { 2079send_error (slot .task -> id ,error ,type ,slot .task -> n_tokens (),slot .n_ctx ); 2080 } 2081 2082void send_error (const int id_task ,const std::string & error ,const enum error_type type = ERROR_TYPE_SERVER ,const int32_t n_prompt_tokens = 0 ,const int32_t n_ctx = 0 ) { 2083SRV_ERR ("task id = %d, error: %s\n" ,id_task ,error .c_str ()); 2084 2085if (type == ERROR_TYPE_EXCEED_CONTEXT_SIZE ) { 2086GGML_ASSERT (n_ctx > 0 && n_prompt_tokens > 0 ); 2087 } 2088 2089auto res = std::make_unique < server_task_result_error > (); 2090res -> id = id_task ; 2091res -> err_type = type ; 2092res -> err_msg = error ; 2093res -> n_prompt_tokens = n_prompt_tokens ; 2094res -> n_ctx = n_ctx ; 2095 2096queue_results .send (std::move (res )); 2097 } 2098 2099void send_partial_response (server_slot & slot ,const completion_token_output & tkn ,bool is_progress ,bool is_begin = false) { 2100auto res = std::make_unique < server_task_result_cmpl_partial > (); 2101 2102res -> id = slot .task -> id ; 2103res -> index = slot .task -> index ; 2104 2105if (is_progress ) { 2106res -> is_progress = true; 2107res -> progress .total = slot .task -> n_tokens (); 2108res -> progress .cache = slot .stats .n_prompt_cached ; 2109res -> progress .processed = slot .prompt .tokens .size (); 2110res -> progress .time_ms = slot .stats .t_elapsed_us () /1000 ; 2111 } 2112if (is_begin ) { 2113res -> is_begin = true; 2114 }else { 2115res -> content = tkn .text_to_send ; 2116res -> tokens = {tkn .tok }; 2117 } 2118 2119res -> n_decoded = slot .stats .n_gen ; 2120res -> n_prompt_tokens = slot .task -> n_tokens (); 2121res -> n_prompt_tokens_cache = slot .stats .n_prompt_cached ; 2122res -> post_sampling_probs = slot .task -> params .post_sampling_probs ; 2123 2124res -> verbose = slot .task -> params .verbose ; 2125res -> res_type = slot .task -> params .res_type ; 2126res -> oaicompat_model = slot .task -> params .oaicompat_model ; 2127res -> oaicompat_cmpl_id = slot .task -> params .oaicompat_cmpl_id ; 2128 2129// populate res.probs_output 2130if (slot .task -> params .sampling .n_probs > 0 ) { 2131res -> prob_output = tkn ;// copy the token probs 2132 } 2133 2134// populate timings if this is final response or timings_per_token is enabled 2135if (slot .stop != STOP_TYPE_NONE || slot .task -> params .timings_per_token ) { 2136res -> stats = slot .stats ; 2137 } 2138 2139queue_results .send (std::move (res )); 2140 } 2141 2142void send_final_response (server_slot & slot ) { 2143auto res = std::make_unique < server_task_result_cmpl_final > (); 2144 2145res -> id = slot .task -> id ; 2146res -> id_slot = slot .id ; 2147 2148res -> index = slot .task -> index ; 2149 2150// keep copy of last generated text for debugging purposes 2151if (slots_debug ) { 2152slot .debug_generated_text = slot .generated_text ; 2153 } 2154 2155// in stream mode, content and tokens are already in last partial chunk 2156if (slot .task -> params .stream ) { 2157res -> content = "" ; 2158res -> tokens = llama_tokens {}; 2159 }else { 2160res -> content = std::move (slot .generated_text ); 2161res -> tokens = std::move (slot .generated_tokens ); 2162 } 2163res -> stats = slot .stats ; 2164res -> prompt = slot .task -> tokens .detokenize (ctx_tgt , true); 2165res -> response_fields = std::move (slot .task -> params .response_fields ); 2166 2167res -> truncated = slot .truncated ; 2168res -> n_decoded = slot .stats .n_gen ; 2169res -> n_prompt_tokens = slot .task -> n_tokens (); 2170res -> n_prompt_tokens_cache = slot .stats .n_prompt_cached ; 2171res -> n_tokens_cached = slot .prompt .n_tokens (); 2172res -> has_new_line = slot .has_new_line ; 2173res -> stopping_word = slot .stopping_word ; 2174res -> stop = slot .stop ; 2175res -> post_sampling_probs = slot .task -> params .post_sampling_probs ; 2176 2177res -> verbose = slot .task -> params .verbose ; 2178res -> stream = slot .task -> params .stream ; 2179res -> include_usage = slot .task -> params .include_usage ; 2180res -> res_type = slot .task -> params .res_type ; 2181res -> oaicompat_model = slot .task -> params .oaicompat_model ; 2182res -> oaicompat_cmpl_id = slot .task -> params .oaicompat_cmpl_id ; 2183 2184// populate res.probs_output 2185if (slot .task -> params .sampling .n_probs > 0 ) { 2186if (!slot .task -> params .stream && slot .stop == STOP_TYPE_WORD ) { 2187const llama_tokens stop_word_toks = common_tokenize (ctx_tgt ,slot .stopping_word , false); 2188 2189size_t safe_offset = std::min (slot .generated_token_probs .size (),stop_word_toks .size ()); 2190res -> probs_output = std::vector < completion_token_output > ( 2191slot .generated_token_probs .begin (), 2192slot .generated_token_probs .end ()- safe_offset ); 2193 }else { 2194res -> probs_output = std::vector < completion_token_output > ( 2195slot .generated_token_probs .begin (), 2196slot .generated_token_probs .end ()); 2197 } 2198 } 2199 2200res -> generation_params = slot .task -> params ;// copy the parameters 2201 2202queue_results .send (std::move (res )); 2203 } 2204 2205void send_embedding (const server_slot & slot ,const llama_batch & batch ) { 2206auto res = std::make_unique < server_task_result_embd > (); 2207res -> id = slot .task -> id ; 2208res -> index = slot .task -> index ; 2209res -> n_tokens = slot .task -> n_tokens (); 2210res -> res_type = slot .task -> params .res_type ; 2211 2212const int n_embd_out = llama_model_n_embd_out (model_tgt ); 2213 2214 std::vector < float > embd_res (n_embd_out ,0.0f ); 2215 2216for (int i = 0 ;i < batch .n_tokens ;++ i ) { 2217if (!batch .logits [i ]|| batch .seq_id [i ][0 ]!= slot .id ) { 2218continue ; 2219 } 2220 2221const float * embd = nullptr ; 2222if (llama_pooling_type (slot .ctx_tgt )== LLAMA_POOLING_TYPE_NONE ) { 2223embd = llama_get_embeddings_ith (slot .ctx_tgt ,i ); 2224 }else { 2225embd = llama_get_embeddings_seq (slot .ctx_tgt ,batch .seq_id [i ][0 ]); 2226 } 2227 2228if (embd == nullptr ) { 2229SLT_ERR (slot ,"failed to get embeddings, token = %d, seq_id = %d\n" ,batch .token [i ],batch .seq_id [i ][0 ]); 2230 2231res -> embedding .push_back (std::vector < float > (n_embd_out ,0.0f )); 2232continue ; 2233 } 2234 2235// normalize only when there is pooling 2236if (llama_pooling_type (slot .ctx_tgt )!= LLAMA_POOLING_TYPE_NONE ) { 2237common_embd_normalize (embd ,embd_res .data (),n_embd_out ,slot .task -> params .embd_normalize ); 2238res -> embedding .push_back (embd_res ); 2239break ; 2240 } 2241 2242res -> embedding .emplace_back (embd ,embd + n_embd_out ); 2243 } 2244 2245SLT_DBG (slot ,"%s" ,"sending embeddings\n" ); 2246 2247queue_results .send (std::move (res )); 2248 } 2249 2250void send_rerank (const server_slot & slot ,const llama_batch & batch ) { 2251auto res = std::make_unique < server_task_result_rerank > (); 2252res -> id = slot .task -> id ; 2253res -> index = slot .task -> index ; 2254res -> n_tokens = slot .task -> n_tokens (); 2255 2256for (int i = 0 ;i < batch .n_tokens ;++ i ) { 2257if (!batch .logits [i ]|| batch .seq_id [i ][0 ]!= slot .id ) { 2258continue ; 2259 } 2260 2261const float * embd = llama_get_embeddings_seq (ctx_tgt ,batch .seq_id [i ][0 ]); 2262if (embd == NULL ) { 2263embd = llama_get_embeddings_ith (ctx_tgt ,i ); 2264 } 2265 2266if (embd == NULL ) { 2267SLT_ERR (slot ,"failed to get embeddings, token = %d, seq_id = %d\n" ,batch .token [i ],batch .seq_id [i ][0 ]); 2268 2269res -> score = -1e6 ; 2270continue ; 2271 } 2272 2273res -> score = embd [0 ]; 2274 } 2275 2276SLT_DBG (slot ,"sending rerank result, res.score = %f\n" ,res -> score ); 2277 2278queue_results .send (std::move (res )); 2279 } 2280 2281// 2282// Functions to process the task 2283// 2284 2285// tokenize the input if it's set by CLI, return false on error 2286bool tokenize_cli_input (server_task & task ) { 2287try { 2288auto & prompt = task .cli_prompt ; 2289if (mctx != nullptr ) { 2290task .tokens = process_mtmd_prompt (mctx ,prompt ,task .cli_files ,init_opt ); 2291 }else { 2292task .tokens = std::move (tokenize_input_prompts (vocab ,mctx ,prompt , true, true,init_opt )[0 ]); 2293 } 2294task .cli_prompt .clear (); 2295task .cli_files .clear (); 2296 }catch (const std::exception & e ) { 2297send_error (task , std::string ("Failed to format input: " )+ e .what (),ERROR_TYPE_INVALID_REQUEST ); 2298return false; 2299 } 2300return true; 2301 } 2302 2303 std::vector < server_slot *> get_free_slots (size_t n_slots_needed ,int exclude_id_slot ) { 2304 std::vector < server_slot *> free_slots ; 2305for (auto & slot :slots ) { 2306if (!slot .is_processing ()&& slot .id != exclude_id_slot ) { 2307free_slots .push_back (& slot ); 2308 } 2309if (free_slots .size () >=n_slots_needed ) { 2310break ; 2311 } 2312 } 2313return free_slots ; 2314 } 2315 2316// launch multiple slots for parent + child tasks 2317bool launch_slots_with_parent_task (server_slot & parent_slot , std::vector < server_slot *> & child_slots ,server_task && parent_task ) { 2318GGML_ASSERT (!parent_slot .is_processing ()); 2319GGML_ASSERT (parent_task .is_parent ()); 2320GGML_ASSERT (child_slots .size ()== parent_task .child_tasks .size ()); 2321 2322int id_parent = parent_task .id ; 2323 2324SRV_TRC ("launching slots for parent task id_task = %d with %zu child tasks\n" ,id_parent ,parent_task .child_tasks .size ()); 2325 2326// to be called in case of failure to release all launched slots 2327auto release_slots = [this ,id_parent ]() { 2328for (auto & slot :slots ) { 2329if (slot .is_processing ()&& ( 2330slot .task -> id == id_parent || 2331slot .task -> id_parent == id_parent 2332 )) { 2333slot .release (); 2334 } 2335 } 2336 }; 2337 2338// launch all child tasks first 2339size_t idx = 0 ; 2340GGML_ASSERT (child_slots .size ()== parent_task .child_tasks .size ()); 2341for (auto * slot :child_slots ) { 2342int id_child = parent_task .child_tasks [idx ].id ; 2343if (!launch_slot_with_task (* slot , std::move (parent_task .child_tasks [idx ]))) { 2344SRV_ERR ("failed to launch slot with child task, id_task = %d\n" ,id_child ); 2345release_slots (); 2346return false; 2347 } 2348idx ++ ; 2349 } 2350 2351// finally, launch the parent task 2352if (!launch_slot_with_task (parent_slot , std::move (parent_task ))) { 2353SRV_ERR ("failed to launch slot with task, id_task = %d\n" ,id_parent ); 2354release_slots (); 2355return false; 2356 } 2357 2358return true; 2359 } 2360 2361// n_tokens_cur: the number of tokens added to the batch for the current slot 2362void create_checkpoint (server_slot & slot ,const int64_t n_tokens_cur ,llama_pos pos_min ,llama_pos pos_max ) { 2363const int id_task = slot .task -> id ; 2364 2365// evict checkpoints within min-step of a previous checkpoint, unless they were 2366// created by the current task 2367// only when the list is full, otherwise short prompts keep just the oldest checkpoint 2368int64_t last = -1 ; 2369for (auto it = slot .prompt .checkpoints .begin (); 2370slot .prompt .checkpoints .size ()+ 1 >= (size_t )params_base .n_ctx_checkpoints && 2371it != slot .prompt .checkpoints .end (); ) { 2372if (it -> id_task != id_task && last >=0 && it -> n_tokens <=last + params_base .checkpoint_min_step ) { 2373SLT_TRC (slot ,"erasing context checkpoint too close to an earlier one (pos_min = %d, pos_max = %d, n_tokens = %" PRId64 ", size = %.3f MiB)\n" , 2374it -> pos_min ,it -> pos_max ,it -> n_tokens , (float )it -> size () /1024 /1024 ); 2375 2376it = slot .prompt .checkpoints .erase (it ); 2377continue ; 2378 } 2379 2380last = it -> n_tokens ; 2381++ it ; 2382 } 2383 2384while (slot .prompt .checkpoints .size () >= (size_t )params_base .n_ctx_checkpoints ) { 2385// make room for the new checkpoint, if needed 2386const auto & cur = slot .prompt .checkpoints .front (); 2387 2388SLT_WRN (slot ,"erasing old context checkpoint (pos_min = %d, pos_max = %d, n_tokens = %" PRId64 ", size = %.3f MiB)\n" , 2389cur .pos_min ,cur .pos_max ,cur .n_tokens , (float )cur .size () /1024 /1024 ); 2390 2391slot .prompt .checkpoints .erase (slot .prompt .checkpoints .begin ()); 2392 } 2393 2394// replace an existing checkpoint at the same n_tokens instead of appending a duplicate 2395 { 2396const int64_t n_tokens_new = slot .prompt .n_tokens ()- n_tokens_cur ; 2397for (auto it = slot .prompt .checkpoints .begin ();it != slot .prompt .checkpoints .end (); ) { 2398if (it -> n_tokens == n_tokens_new ) { 2399SLT_TRC (slot ,"superseding context checkpoint at n_tokens = %" PRId64 "\n" ,it -> n_tokens ); 2400it = slot .prompt .checkpoints .erase (it ); 2401 }else { 2402++ it ; 2403 } 2404 } 2405 } 2406 2407auto & cur = slot .prompt .checkpoints .emplace_back (); 2408 2409cur .id_task = id_task ; 2410 2411// [TAG_CHECKPOINTS_FIX_POS_MIN] 2412// TODO: here we incorrectly deterimne that the saved checkpoint data covers the [pos_min, pos_max] range 2413// this is not true for SWA models: https://github.com/ggml-org/llama.cpp/pull/24411#issuecomment-4677983225 2414cur .update_pos (slot .prompt .n_tokens ()- n_tokens_cur ,pos_min ,pos_max ); 2415 2416cur .update_tgt (ctx_tgt ,slot .id ,LLAMA_STATE_SEQ_FLAGS_PARTIAL_ONLY ); 2417cur .update_dft (ctx_dft ,slot .id ,LLAMA_STATE_SEQ_FLAGS_PARTIAL_ONLY ); 2418// stash the draft's speculative state with the checkpoint 2419common_speculative_get_state (spec .get (),slot .id ,cur .data_spec ); 2420 2421SLT_TRC (slot , 2422"created context checkpoint %d of %d (pos_min = %d, pos_max = %d, n_tokens = %" PRId64 ", size = %.3f MiB)\n" , 2423 (int )slot .prompt .checkpoints .size (),params_base .n_ctx_checkpoints ,cur .pos_min , 2424cur .pos_max ,cur .n_tokens , (float )cur .size () /1024 /1024 ); 2425 } 2426 2427// returns false to decline the task, it is offered again after the decode is done 2428bool process_single_task (server_task && task ,bool is_yielding ) { 2429// while yielding, an encode / decode is running and only reading the server state is safe 2430if (is_yielding && task .type != SERVER_TASK_TYPE_METRICS && task .type != SERVER_TASK_TYPE_SLOT_GET ) { 2431SRV_DBG ("decoding, decline task, id_task = %d\n" ,task .id ); 2432return false; 2433 } 2434 2435switch (task .type ) { 2436case SERVER_TASK_TYPE_COMPLETION : 2437case SERVER_TASK_TYPE_INFILL : 2438case SERVER_TASK_TYPE_EMBEDDING : 2439case SERVER_TASK_TYPE_RERANK : 2440 { 2441// special case: if input is provided via CLI, tokenize it first 2442// otherwise, no need to tokenize as it's already done inside the HTTP thread 2443if (task .cli ) { 2444if (!tokenize_cli_input (task )) { 2445break ; 2446 } 2447 } 2448 2449const int id_task = task .id ; 2450 2451server_slot * slot = get_available_slot (task ); 2452 2453// 2454// slot scheduling logic 2455// 2456 2457if (slot == nullptr ) { 2458// if no slot is available, we defer this task for processing later 2459SRV_DBG ("no slot is available, defer task, id_task = %d\n" ,id_task ); 2460queue_tasks .defer (std::move (task )); 2461break ; 2462 } 2463 2464if (slot -> is_processing ()) { 2465// if requested slot is unavailable, we defer this task for processing later 2466SRV_DBG ("requested slot is unavailable, defer task, id_task = %d\n" ,id_task ); 2467queue_tasks .defer (std::move (task )); 2468break ; 2469 } 2470 2471if (task .is_parent ()) { 2472// try getting free slots for all child tasks 2473size_t n_child_tasks = task .child_tasks .size (); 2474 std::vector < server_slot *> child_slots = get_free_slots (n_child_tasks ,slot -> id ); 2475if (child_slots .size ()< n_child_tasks ) { 2476SRV_DBG ("not enough free slots for child tasks, n_free = %zu, n_children = %zu, defer task, id_task = %d\n" ,child_slots .size (),n_child_tasks ,id_task ); 2477queue_tasks .defer (std::move (task )); 2478break ; 2479 } 2480if (!launch_slots_with_parent_task (* slot ,child_slots , std::move (task ))) { 2481SRV_ERR ("failed to launch slot with parent task, id_task = %d\n" ,id_task ); 2482break ;// drop the task 2483 } 2484 }else if (!launch_slot_with_task (* slot , std::move (task ))) { 2485SRV_ERR ("failed to launch slot with task, id_task = %d\n" ,id_task ); 2486break ;// drop the task 2487 } 2488 2489if (params_base .cache_idle_slots ) { 2490for (auto & slot :slots ) { 2491if (!slot .is_processing ()) { 2492SLT_TRC (slot ,"%s" ,"saving idle slot to prompt cache\n" ); 2493 2494if (slot .prompt_save (* prompt_cache )) { 2495SLT_DBG (slot ,"%s" ,"__TEST_TAG_CACHE_IDLE_SLOT__\n" ); 2496prompt_cache -> update (); 2497 } 2498 2499if (params_base .kv_unified ) { 2500// [TAG_IDLE_SLOT_CLEAR] 2501slot .prompt_clear (); 2502 } 2503 } 2504 } 2505 } 2506 }break ; 2507case SERVER_TASK_TYPE_CANCEL : 2508 { 2509// release slot linked with the task id 2510for (auto & slot :slots ) { 2511if (slot .task && slot .task -> id == task .id_target ) { 2512slot .release (); 2513break ; 2514 } 2515 } 2516 }break ; 2517case SERVER_TASK_TYPE_CONTROL : 2518 { 2519auto res = std::make_unique < server_task_result_control > (); 2520res -> id = task .id ; 2521 2522server_slot * slot = get_slot_by_cmpl_id (task .params .control_cmpl_id ); 2523if (slot == nullptr ) { 2524SRV_WRN ("control %s on unknown completion id=%s, no live slot\n" , 2525task .params .control_action .c_str (),task .params .control_cmpl_id .c_str ()); 2526res -> success = false; 2527res -> message = "no active completion for this id" ; 2528queue_results .send (std::move (res )); 2529break ; 2530 } 2531 2532if (task .params .control_action == "reasoning_end" ) { 2533// the budget sampler only exists when reasoning control was armed 2534if (!slot -> task -> params .sampling .reasoning_control ) { 2535res -> success = false; 2536res -> message = "reasoning control not enabled for this completion" ; 2537queue_results .send (std::move (res )); 2538break ; 2539 } 2540// act on the live slot mid generation, never defer 2541common_sampler_reasoning_budget_force (slot -> smpl .get ()); 2542res -> success = true; 2543 }else { 2544res -> success = false; 2545res -> message = "unknown control action" ; 2546 } 2547 2548queue_results .send (std::move (res )); 2549 }break ; 2550case SERVER_TASK_TYPE_NEXT_RESPONSE : 2551 { 2552// do nothing 2553 }break ; 2554case SERVER_TASK_TYPE_METRICS : 2555 { 2556int n_processing_slots = 0 ; 2557 2558for (server_slot & slot :slots ) { 2559if (slot .is_processing ()) { 2560n_processing_slots ++ ; 2561 } 2562 } 2563SRV_DBG ("n_processing_slots = %d\n" ,n_processing_slots ); 2564 2565auto res = std::make_unique < server_task_result_metrics > (); 2566res -> id = task .id ; 2567res -> n_processing_slots = n_processing_slots ; 2568res -> n_tasks_deferred = queue_tasks .queue_tasks_deferred_size (); 2569res -> metrics = metrics ; 2570 2571if (task .metrics_reset_bucket ) { 2572metrics .reset_bucket (); 2573 } 2574queue_results .send (std::move (res )); 2575 }break ; 2576case SERVER_TASK_TYPE_SLOT_GET : 2577 { 2578json slots_data = json::array (); 2579 2580int n_idle_slots = 0 ; 2581 2582for (server_slot & slot :slots ) { 2583if (!slot .is_processing ()) { 2584n_idle_slots ++ ; 2585 } 2586 2587slots_data .push_back (slot .to_json (slots_debug == 0 )); 2588 } 2589SRV_DBG ("n_idle_slots = %d\n" ,n_idle_slots ); 2590 2591auto res = std::make_unique < server_task_result_slots > (); 2592res -> id = task .id ; 2593res -> slots_data = std::move (slots_data ); 2594res -> n_idle_slots = n_idle_slots ; 2595 2596queue_results .send (std::move (res )); 2597 }break ; 2598case SERVER_TASK_TYPE_SLOT_SAVE : 2599 { 2600const int id_slot = task .slot_action .id_slot ; 2601server_slot * slot = get_slot_by_id (id_slot ); 2602if (slot == nullptr ) { 2603send_error (task ,"Invalid slot ID" ,ERROR_TYPE_INVALID_REQUEST ); 2604break ; 2605 } 2606if (slot -> is_processing ()) { 2607// if requested slot is unavailable, we defer this task for processing later 2608SRV_DBG ("requested slot is unavailable, defer task, id_task = %d\n" ,task .id ); 2609queue_tasks .defer (std::move (task )); 2610break ; 2611 } 2612 2613const int64_t t_start = ggml_time_us (); 2614 2615 std::string filename = task .slot_action .filename ; 2616 std::string filepath = task .slot_action .filepath ; 2617 2618 std::vector < char > packed ; 2619try { 2620packed = slot -> prompt .tokens .serialize (); 2621 }catch (const std::exception & err ) { 2622send_error (task ,err .what (),ERROR_TYPE_NOT_SUPPORTED ); 2623break ; 2624 } 2625 2626GGML_ASSERT (packed .size () %sizeof (llama_token )== 0 ); 2627const size_t nwrite = llama_state_seq_save_file ( 2628ctx_tgt ,filepath .c_str (),slot -> id , 2629reinterpret_cast < const llama_token *> (packed .data ()),packed .size () /sizeof (llama_token )); 2630if (nwrite == 0 ) { 2631send_error (task ,"Unable to save slot" ,ERROR_TYPE_SERVER ); 2632break ; 2633 } 2634 2635const int64_t t_end = ggml_time_us (); 2636const double t_save_ms = (t_end - t_start ) /1000.0 ; 2637 2638auto res = std::make_unique < server_task_result_slot_save_load > (); 2639res -> id = task .id ; 2640res -> id_slot = id_slot ; 2641res -> filename = filename ; 2642res -> is_save = true; 2643res -> n_tokens = slot -> prompt .tokens .size (); 2644res -> n_bytes = nwrite ; 2645res -> t_ms = t_save_ms ; 2646queue_results .send (std::move (res )); 2647 }break ; 2648case SERVER_TASK_TYPE_SLOT_RESTORE : 2649 { 2650const int id_slot = task .slot_action .id_slot ; 2651server_slot * slot = get_slot_by_id (id_slot ); 2652if (slot == nullptr ) { 2653send_error (task ,"Invalid slot ID" ,ERROR_TYPE_INVALID_REQUEST ); 2654break ; 2655 } 2656if (slot -> is_processing ()) { 2657// if requested slot is unavailable, we defer this task for processing later 2658SRV_DBG ("requested slot is unavailable, defer task, id_task = %d\n" ,task .id ); 2659queue_tasks .defer (std::move (task )); 2660break ; 2661 } 2662 2663const int64_t t_start = ggml_time_us (); 2664 2665 std::string filename = task .slot_action .filename ; 2666 std::string filepath = task .slot_action .filepath ; 2667 2668size_t nread = 0 ; 2669try { 2670size_t n_packed = 0 ; 2671llama_tokens packed ; 2672nread = llama_state_seq_load_file (ctx_tgt ,filepath .c_str (),slot -> id ,nullptr ,0 ,& n_packed ); 2673if (nread != 0 ) { 2674packed .resize (std::max < size_t > (1 ,n_packed )); 2675nread = llama_state_seq_load_file (ctx_tgt ,filepath .c_str (),slot -> id ,packed .data (),packed .size (),& n_packed ); 2676 } 2677if (nread == 0 ) { 2678throw std::runtime_error ("No available space in KV cache or invalid slot save file" ); 2679 } 2680packed .resize (n_packed ); 2681 2682server_tokens restored = server_tokens::deserialize (packed ,mctx != nullptr ); 2683 2684if (restored .size ()> (size_t )slot -> n_ctx ) { 2685throw std::runtime_error ("Restored prompt does not fit in the slot context" ); 2686 } 2687 2688if (!restored .validate (ctx_tgt )) { 2689throw std::runtime_error ("Invalid tokens in slot save file" ); 2690 } 2691 2692slot -> prompt .clear (); 2693slot -> prompt .tokens = std::move (restored ); 2694 }catch (const std::exception & err ) { 2695slot -> prompt_clear (); 2696send_error (task , std::string ("Unable to restore slot: " )+ err .what (),ERROR_TYPE_INVALID_REQUEST ); 2697break ; 2698 } 2699 2700const int64_t t_end = ggml_time_us (); 2701const double t_restore_ms = (t_end - t_start ) /1000.0 ; 2702 2703auto res = std::make_unique < server_task_result_slot_save_load > (); 2704res -> id = task .id ; 2705res -> id_slot = id_slot ; 2706res -> filename = filename ; 2707res -> is_save = false; 2708res -> n_tokens = slot -> prompt .tokens .size (); 2709res -> n_bytes = nread ; 2710res -> t_ms = t_restore_ms ; 2711queue_results .send (std::move (res )); 2712 }break ; 2713case SERVER_TASK_TYPE_SLOT_ERASE : 2714 { 2715const int id_slot = task .slot_action .id_slot ; 2716server_slot * slot = get_slot_by_id (id_slot ); 2717if (slot == nullptr ) { 2718send_error (task ,"Invalid slot ID" ,ERROR_TYPE_INVALID_REQUEST ); 2719break ; 2720 } 2721if (slot -> is_processing ()) { 2722// if requested slot is unavailable, we defer this task for processing later 2723SRV_DBG ("requested slot is unavailable, defer task, id_task = %d\n" ,task .id ); 2724queue_tasks .defer (std::move (task )); 2725break ; 2726 } 2727 2728// Erase token cache 2729const size_t n_erased = slot -> prompt .tokens .size (); 2730 2731slot -> prompt_clear (); 2732 2733auto res = std::make_unique < server_task_result_slot_erase > (); 2734res -> id = task .id ; 2735res -> id_slot = id_slot ; 2736res -> n_erased = n_erased ; 2737queue_results .send (std::move (res )); 2738 }break ; 2739case SERVER_TASK_TYPE_GET_LORA : 2740 { 2741// TODO @ngxson : make lora_adapters a dedicated member of server_context 2742auto & loras = params_base .lora_adapters ; 2743auto res = std::make_unique < server_task_result_get_lora > (); 2744res -> id = task .id ; 2745for (size_t i = 0 ;i < loras .size ();++ i ) { 2746auto & lora = loras [i ]; 2747 std::string alora_invocation_string = "" ; 2748const uint64_t n_alora_tokens = llama_adapter_get_alora_n_invocation_tokens (lora .ptr ); 2749llama_tokens alora_invocation_tokens ; 2750if (n_alora_tokens ) { 2751const llama_token * alora_tokens = llama_adapter_get_alora_invocation_tokens (lora .ptr ); 2752for (uint64_t j = 0 ;j < n_alora_tokens ;++ j ) { 2753alora_invocation_string += common_token_to_piece (vocab ,alora_tokens [j ]); 2754alora_invocation_tokens .push_back (alora_tokens [j ]); 2755 } 2756 } 2757res -> loras .push_back (server_task_result_get_lora::lora { 2758lora , 2759alora_invocation_string , 2760alora_invocation_tokens , 2761 }); 2762 } 2763queue_results .send (std::move (res )); 2764 }break ; 2765case SERVER_TASK_TYPE_SET_LORA : 2766 { 2767auto new_loras = construct_lora_list (task .set_lora ); 2768// logging 2769for (size_t i = 0 ;i < new_loras .size ();++ i ) { 2770SRV_TRC ("set lora adapter idx=%zu scale=%f\n" ,i ,new_loras [i ].scale ); 2771 } 2772// TODO @ngxson : make lora_adapters a dedicated member of server_context 2773params_base .lora_adapters = new_loras ; 2774auto res = std::make_unique < server_task_result_apply_lora > (); 2775res -> id = task .id ; 2776queue_results .send (std::move (res )); 2777 }break ; 2778 } 2779 2780return true; 2781 } 2782 2783void iterate (std::vector < server_slot > & slots , std::function < void (server_slot & )> callback ) { 2784for (auto & slot :slots ) { 2785try { 2786callback (slot ); 2787 }catch (const std::exception & e ) { 2788SLT_ERR (slot ,"got exception: %s\n" ,e .what ()); 2789send_error (slot , std::string ("got exception: " )+ e .what (),ERROR_TYPE_SERVER ); 2790slot .release (); 2791 } 2792 } 2793 } 2794 2795void iterate (std::vector < server_slot *> & slots , std::function < void (server_slot & )> callback ) { 2796for (auto & slot :slots ) { 2797try { 2798callback (* slot ); 2799 }catch (const std::exception & e ) { 2800SLT_ERR (* slot ,"got exception: %s\n" ,e .what ()); 2801send_error (* slot , std::string ("got exception: " )+ e .what (),ERROR_TYPE_SERVER ); 2802slot -> release (); 2803 } 2804 } 2805 } 2806 2807void abort_all_slots (const std::string & reason ) { 2808for (auto & slot :slots ) { 2809if (slot .is_processing ()) { 2810send_error (slot ,reason ,ERROR_TYPE_SERVER ); 2811slot .release (); 2812 } 2813 } 2814 } 2815 2816// @ngxson : for debugging only 2817int64_t t_pre_decode = 0 ; 2818int64_t t_decode = 0 ; 2819int64_t t_post_decode = 0 ; 2820int64_t t_sampl = 0 ; 2821int64_t n_pre_decode = 0 ; 2822int64_t n_decode = 0 ; 2823int64_t n_post_decode = 0 ; 2824int64_t n_sampl = 0 ; 2825// #define DEBUG_TIMINGS 2826#ifdef DEBUG_TIMINGS 2827struct scoped_timer { 2828int64_t & t ; 2829int64_t & n ; 2830int64_t t_start ; 2831scoped_timer (int64_t & t_ ,int64_t & n_ ) :t (t_ ),n (n_ ) { 2832t_start = ggml_time_us (); 2833 } 2834 ~scoped_timer () { 2835t += ggml_time_us ()- t_start ; 2836n ++ ; 2837 } 2838 }; 2839#else 2840struct scoped_timer { 2841scoped_timer (int64_t & ,int64_t & ) {} 2842 ~scoped_timer () {} 2843 }; 2844#endif 2845 2846void update_slots () { 2847#ifdef DEBUG_TIMINGS 2848static int64_t t_prev = 0 ; 2849int64_t t_start = ggml_time_us (); 2850if (t_start - t_prev > 5 * 1000 * 1000 ) {// every 5 seconds 2851t_prev = t_start ; 2852SRV_INF ("n_pre_decode = %" PRId64 "\n" ,n_pre_decode ); 2853SRV_INF ("avg t_pre_decode = %f ms\n" , (double )t_pre_decode /n_pre_decode /1000.0 ); 2854SRV_INF ("avg t_decode = %f ms\n" , (double )t_decode /n_decode /1000.0 ); 2855SRV_INF ("avg t_post_decode = %f ms\n" , (double )t_post_decode /n_post_decode /1000.0 ); 2856SRV_INF ("avg t_sampl = %f ms\n" , (double )t_sampl /n_sampl /1000.0 ); 2857 } 2858#endif 2859 2860// check if all slots are idle 2861 { 2862bool all_idle = true; 2863 2864for (auto & slot :slots ) { 2865if (slot .is_processing ()) { 2866all_idle = false; 2867break ; 2868 } 2869 } 2870 2871if (all_idle ) { 2872SRV_TRC ("%s" ,"all slots are idle\n" ); 2873 2874metrics_flush_idle (); 2875 2876return ;// skip further processing 2877 2878 }else { 2879SRV_DBG ("%s" ,"posting NEXT_RESPONSE\n" ); 2880 2881server_task task (SERVER_TASK_TYPE_NEXT_RESPONSE ); 2882task .id = queue_tasks .get_new_id (); 2883queue_tasks .post (std::move (task )); 2884 } 2885 } 2886 2887try { 2888scoped_timer t (t_pre_decode ,n_pre_decode ); 2889pre_decode (); 2890batch .render (); 2891 }catch (const std::exception & e ) { 2892SRV_ERR ("pre_decode() failed: %s\n" ,e .what ()); 2893abort_all_slots ("pre_decode() failed: " + std::string (e .what ())); 2894 2895// the batch is half-built and not rendered, skip now to avoid UB 2896return ; 2897 } 2898 2899GGML_ASSERT (batch .slot_batched || batch .size ()== 0 ); 2900 2901if (batch .slot_batched ) { 2902auto & slot_batched = batch .slot_batched ; 2903auto & alora_scale = batch .alora_scale ; 2904auto & alora_disabled_id = batch .alora_disabled_id ; 2905 2906// TODO @ngxson : alora handling is too messy, need to refactor it to be more clear and maintainable 2907// apply lora, only need to do it once per batch 2908common_set_adapter_lora (ctx_tgt ,slot_batched -> lora ); 2909 2910// if the lora is temporarily disabled for an alora, re-enable it 2911// for next time 2912if (alora_scale > 0.0f ) { 2913SRV_DBG ("re-enabling alora with scale %f\n" ,alora_scale ); 2914slot_batched -> lora [alora_disabled_id ].scale = alora_scale ; 2915 } 2916 2917llama_set_embeddings (ctx_tgt ,slot_batched -> need_embd ()); 2918 } 2919 2920llama_batch batch_view ; 2921int32_t off_next = 0 ; 2922int32_t n_batch = llama_n_batch (ctx_tgt ); 2923for (int32_t off = 0 ;off < batch .size ();off = off_next ) { 2924const int32_t n_tokens = std::min (n_batch ,batch .size ()- off ); 2925try { 2926scoped_timer t (t_decode ,n_decode ); 2927// TODO @ngxson : maybe handle n_batch == 1 here instead of inside decode() 2928 2929batch_view = batch .get_view (off ,n_tokens ); 2930bool ok = decode (n_batch ,off ,batch_view ); 2931#ifdef DEBUG_TIMINGS 2932llama_synchronize (ctx_tgt ); 2933#endif 2934 2935if (ok ) { 2936// move the head of the batch forward with the number of tokens we just processed 2937off_next = off + n_tokens ; 2938 2939// on successful decode, restore the original batch size 2940n_batch = llama_n_batch (ctx_tgt ); 2941 }else { 2942// try again with the updated n_batch 2943continue ; 2944 } 2945 }catch (const std::exception & e ) { 2946SRV_ERR ("decode() failed: %s\n" ,e .what ()); 2947abort_all_slots ("decode() failed: " + std::string (e .what ())); 2948break ;// stop any further processing 2949 } 2950 2951try { 2952scoped_timer t (t_post_decode ,n_post_decode ); 2953post_decode (n_tokens ,off ,batch_view ); 2954 }catch (const std::exception & e ) { 2955SRV_ERR ("post_decode() failed: %s\n" ,e .what ()); 2956abort_all_slots ("post_decode() failed: " + std::string (e .what ())); 2957break ;// stop any further processing 2958 } 2959 } 2960 } 2961 2962void pre_decode () { 2963// apply context-shift if needed 2964// TODO: simplify and improve 2965iterate (slots , [& ](server_slot & slot ) { 2966if (slot .state == SLOT_STATE_GENERATING && slot .prompt .n_tokens ()+ 1 >=slot .n_ctx ) { 2967if (!params_base .ctx_shift ) { 2968// this check is redundant (for good) 2969// we should never get here, because generation should already stopped in process_token() 2970send_error (slot ,"context shift is disabled" ,ERROR_TYPE_SERVER ); 2971slot .release (); 2972return ; 2973 } 2974 2975if (mctx ) { 2976// we should never reach this because params_base.ctx_shift is automatically disabled if mmproj is loaded 2977// we don't support ctx_shift because an image chunk may contains multiple tokens 2978GGML_ABORT ("not supported by multimodal" ); 2979 } 2980 2981if (slot .task -> is_parent ()|| slot .task -> is_child ()) { 2982send_error (slot ,"context shift cannot be used for shared prompt" ,ERROR_TYPE_SERVER ); 2983slot .release (); 2984return ; 2985 } 2986 2987// Shift context 2988int n_keep = slot .task -> params .n_keep < 0 ?slot .task -> n_tokens () :slot .task -> params .n_keep ; 2989 2990if (add_bos_token ) { 2991n_keep += 1 ; 2992 } 2993 2994n_keep = std::min (slot .n_ctx - 4 ,n_keep ); 2995 2996const int n_left = slot .prompt .n_tokens ()- n_keep ; 2997int n_discard = slot .task -> params .n_discard ?slot .task -> params .n_discard : (n_left /2 ); 2998 2999// ref: https://github.com/ggml-org/llama.cpp/pull/24786 3000n_discard = std::clamp (n_discard ,0 , std::max (0 ,n_left - 1 )); 3001 3002SLT_WRN (slot ,"slot context shift, n_keep = %d, n_left = %d, n_discard = %d\n" ,n_keep ,n_left ,n_discard ); 3003 3004slot .mem .seq_rm (slot .id ,n_keep ,n_keep + n_discard ); 3005slot .mem .seq_add (slot .id ,n_keep + n_discard ,slot .prompt .tokens .pos_next (),- n_discard ); 3006 3007// add generated tokens to cache 3008// ref: https://github.com/ggml-org/llama.cpp/pull/16818#discussion_r2473269481 3009 { 3010GGML_ASSERT (!slot .prompt .tokens .has_mtmd ); 3011 3012llama_tokens new_tokens = slot .prompt .tokens .get_tokens ();// copy 3013for (size_t i = n_keep + n_discard ;i < new_tokens .size ();i ++ ) { 3014new_tokens [i - n_discard ]= new_tokens [i ]; 3015 } 3016 3017new_tokens .resize (slot .prompt .tokens .size ()- n_discard ); 3018 3019slot .prompt .clear (); 3020slot .prompt .tokens .insert (new_tokens ); 3021 } 3022 3023slot .truncated = true; 3024 } 3025 }); 3026 3027// start populating the batch for this iteration 3028batch .clear (); 3029 3030// track if given slot can be batched with slots already in the batch 3031auto & slot_batched = batch .slot_batched ; 3032 3033 std::vector < server_slot *> generating ; 3034 std::vector < server_slot *> drafting ; 3035 3036// determine which slots are generating and drafting 3037iterate (slots , [& ](server_slot & slot ) { 3038if (slot .state != SLOT_STATE_GENERATING ) { 3039return ; 3040 } 3041 3042// check if we can batch this slot with the previous one 3043if (!slot_batched ) { 3044slot_batched = & slot ; 3045 }else if (!slot_batched -> can_batch_with (slot )) { 3046return ; 3047 } 3048 3049generating .push_back (& slot ); 3050 3051if (spec ) { 3052common_speculative_get_draft_params (spec .get (),slot .id ).drafting = false; 3053 3054const bool use_ckpt_tgt = ctx_tgt_seq_rm_type == COMMON_CONTEXT_SEQ_RM_TYPE_FULL ; 3055const bool use_ckpt_dft = ctx_dft_seq_rm_type == COMMON_CONTEXT_SEQ_RM_TYPE_FULL ; 3056 3057const int n_draft_max = slot .get_n_draft_max (); 3058 3059if (n_draft_max > 0 ) { 3060GGML_ASSERT (slot .can_speculate ()); 3061 3062if (!slot .spec_draft .empty ()) { 3063// we have a previous (partial) draft to reuse 3064if (use_ckpt_tgt ) { 3065GGML_ASSERT (!slot .spec_ckpt .empty ()); 3066 } 3067 }else { 3068GGML_ASSERT (slot .spec_i_batch .empty ()); 3069 3070slot .spec_ckpt .update_pos ( 3071slot .prompt .n_tokens (), 3072llama_memory_seq_pos_min (llama_get_memory (ctx_tgt ),slot .id ), 3073llama_memory_seq_pos_max (llama_get_memory (ctx_tgt ),slot .id )); 3074 3075if (use_ckpt_dft ) { 3076slot .spec_ckpt .update_dft (ctx_dft ,slot .id ,LLAMA_STATE_SEQ_FLAGS_PARTIAL_ONLY ); 3077 } 3078 3079slot .spec_prompt = slot .prompt .tokens .get_text_tokens (); 3080 3081common_speculative_get_draft_params (spec .get (),slot .id )= { 3082/* .drafting = */ true, 3083/* .n_max = */ n_draft_max , 3084/* .pos0 = */ slot .prompt .tokens .pos_next (), 3085/* .id_last = */ slot .sampled , 3086/* .prompt = */ & slot .spec_prompt , 3087/* .result = */ & slot .spec_draft , 3088 }; 3089 3090drafting .push_back (& slot ); 3091 } 3092 } 3093 } 3094 }); 3095 3096// generate the actual drafts (if any) 3097if (!drafting .empty ()) { 3098queue_tasks .yield_to_queue ([& ]() { 3099common_speculative_draft (spec .get ()); 3100 }); 3101 } 3102 3103// make checkpoints if needed 3104iterate (drafting , [& ](server_slot & slot ) { 3105auto & draft = slot .spec_draft ; 3106auto & ckpt = slot .spec_ckpt ; 3107 3108slot .stats .n_draft_tokens += draft .size (); 3109 3110// TODO: avoid restoring the draft context and re-evaluating the drafted tokens when not needed [TAG_SPEC_AVOID_DRAFT_REEVAL] 3111const bool use_ckpt_dft = ctx_dft_seq_rm_type == COMMON_CONTEXT_SEQ_RM_TYPE_FULL ; 3112 3113if (ctx_dft ) { 3114if (use_ckpt_dft ) { 3115ckpt .load_dft (ctx_dft ,slot .id ,LLAMA_STATE_SEQ_FLAGS_PARTIAL_ONLY ); 3116 } 3117 3118if (!llama_memory_seq_rm (llama_get_memory (ctx_dft ),slot .id ,ckpt .pos_max + 1 ,-1 )) { 3119GGML_ABORT ("failed to remove sequence %d\n" ,slot .id ); 3120 } 3121 } 3122 3123if (!draft .empty ()) { 3124const bool use_ckpt_tgt = 3125ctx_tgt_seq_rm_type == COMMON_CONTEXT_SEQ_RM_TYPE_FULL || 3126 (ctx_tgt_seq_rm_type == COMMON_CONTEXT_SEQ_RM_TYPE_RS && draft .size ()> llama_n_rs_seq (ctx_tgt )); 3127 3128const bool use_ckpt_dft = 3129 (ctx_dft_seq_rm_type == COMMON_CONTEXT_SEQ_RM_TYPE_RS && draft .size ()> llama_n_rs_seq (ctx_dft )); 3130 3131if (use_ckpt_tgt ) { 3132//const int64_t t_start = ggml_time_us(); 3133 3134ckpt .update_tgt (ctx_tgt ,slot .id ,LLAMA_STATE_SEQ_FLAGS_PARTIAL_ONLY ); 3135 3136//const int64_t t_total = ggml_time_us() - t_start; 3137//printf("checkpoint total: %f ms\n", t_total / 1000.0); 3138 3139SLT_DBG (slot ,"created speculative checkpoint (pos_min = %d, pos_max = %d, n_tokens = %d, size = %.3f MiB, draft = %.3f MiB)\n" , 3140ckpt .pos_min ,ckpt .pos_max ,slot .prompt .n_tokens (), 3141 (float )ckpt .size () /1024 /1024 , 3142 (float )ckpt .data_dft .size () /1024 /1024 ); 3143 } 3144 3145if (use_ckpt_dft ) { 3146ckpt .update_dft (ctx_dft ,slot .id ,LLAMA_STATE_SEQ_FLAGS_PARTIAL_ONLY ); 3147 } 3148 } 3149 }); 3150 3151// update the batch with the sampled/drafted tokens 3152iterate (generating , [& ](server_slot & slot ) { 3153slot .handle_last_sampled_token (batch ); 3154 }); 3155 3156// process in chunks of params.n_batch 3157int32_t n_batch = llama_n_batch (ctx_tgt ); 3158int32_t n_ubatch = llama_n_ubatch (ctx_tgt ); 3159 3160auto & alora_scale = batch .alora_scale ; 3161auto & alora_disabled_id = batch .alora_disabled_id ; 3162 3163// next, batch any pending prompts without exceeding n_batch 3164if (params_base .cont_batching || batch .size ()== 0 ) { 3165bool add_ok = true;// false means the batch is full, skip remaining slots 3166 3167iterate (slots , [& ](server_slot & slot ) { 3168if (!add_ok || batch .size () >=n_batch ) { 3169return ;// batch is full, skip remaining slots 3170 } 3171 3172if (!slot .is_processing ()) { 3173return ; 3174 } 3175 3176// check if we can batch this slot with the previous one 3177if (slot_batched && !slot_batched -> can_batch_with (slot )) { 3178return ; 3179 } 3180 3181// check if this is a child slot 3182if (slot .state == SLOT_STATE_WAIT_OTHER ) { 3183SLT_DBG (slot ,"%s" ,"waiting for parent slot to complete\n" ); 3184return ; 3185 } 3186 3187// this slot still has a prompt to be processed 3188if (slot .state == SLOT_STATE_PROCESSING_PROMPT || slot .state == SLOT_STATE_STARTED ) { 3189const auto & input_tokens = slot .task -> tokens ; 3190 3191// used to determine the number of tokens added to the batch for the current slot 3192const auto n_tokens_prev = batch .size (); 3193 3194// TODO: maybe move branch to outside of this loop in the future 3195if (slot .state == SLOT_STATE_STARTED ) { 3196slot .stats .update_prompt_start (); 3197 3198slot .state = SLOT_STATE_PROCESSING_PROMPT ; 3199 3200SLT_TRC (slot ,"new prompt, n_ctx_slot = %d, n_keep = %d, task.n_tokens = %d\n" , 3201slot .n_ctx ,slot .task -> params .n_keep ,slot .task -> n_tokens ()); 3202 3203// print prompt tokens (for debugging) 3204/*if (1) { 3205// first 16 tokens (avoid flooding logs) 3206for (int i = 0; i < std::min<int>(16, input_tokens.size()); i++) { 3207SLT_DBG(slot, "prompt token %3d: %6d '%s'\n", i, input_tokens[i], common_token_to_piece(ctx_tgt, input_tokens[i]).c_str()); 3208} 3209} else { 3210// all 3211for (int i = 0; i < (int) input_tokens.size(); i++) { 3212SLT_DBG(slot, "prompt token %3d: %6d '%s'\n", i, input_tokens[i], common_token_to_piece(ctx_tgt, input_tokens[i]).c_str()); 3213} 3214}*/ 3215 3216// keep track how many tokens we can reuse from the previous state 3217int n_past = 0 ; 3218 3219// empty prompt passed -> release the slot and send empty response 3220if (input_tokens .empty ()) { 3221SLT_WRN (slot ,"%s" ,"empty prompt - releasing slot\n" ); 3222 3223slot .print_timings (); 3224send_final_response (slot ); 3225slot .release (); 3226 3227return ; 3228 } 3229 3230// TODO: support memory-less logits computation 3231if (slot .task -> need_logits ()&& !llama_get_memory (ctx_tgt )) { 3232send_error (slot ,"the current context does not logits computation. skipping" ,ERROR_TYPE_SERVER ); 3233slot .release (); 3234return ; 3235 } 3236 3237if (!slot .can_split ()) { 3238if (slot .task -> n_tokens ()> n_ubatch ) { 3239send_error (slot , 3240string_format ( 3241"input (%d tokens) is too large to process. increase the physical batch " 3242"size (current batch size: %d)" , 3243slot .task -> n_tokens (),n_ubatch ), 3244ERROR_TYPE_SERVER ); 3245slot .release (); 3246return ; 3247 } 3248 3249if (slot .task -> n_tokens ()> slot .n_ctx ) { 3250send_error ( 3251slot , 3252string_format ( 3253"input (%d tokens) is larger than the max context size (%d tokens). skipping" , 3254slot .task -> n_tokens (),slot .n_ctx ), 3255ERROR_TYPE_EXCEED_CONTEXT_SIZE ); 3256slot .release (); 3257return ; 3258 } 3259 }else { 3260if (slot .task -> n_tokens () >=slot .n_ctx ) { 3261send_error (slot , 3262string_format ("request (%d tokens) exceeds the available context size (%d " 3263"tokens), try increasing it" , 3264slot .task -> n_tokens (),slot .n_ctx ), 3265ERROR_TYPE_EXCEED_CONTEXT_SIZE ); 3266slot .release (); 3267return ; 3268 } 3269 3270if (slot .task -> params .cache_prompt ) { 3271// reuse any previously computed tokens that are common with the new prompt 3272n_past = slot .prompt .tokens .get_common_prefix (input_tokens ); 3273 3274// if there is an alora invoked, don't cache after the invocation start 3275if (slot .alora_invocation_start > 0 ) { 3276SLT_DBG (slot ,"only caching to alora invocation start (n_past = %d, alora_invocation_start = %d)\n" ,n_past ,slot .alora_invocation_start ); 3277n_past = std::min (n_past ,slot .alora_invocation_start - 1 ); 3278 } 3279 3280const auto n_cache_reuse = slot .task -> params .n_cache_reuse ; 3281 3282const bool can_cache_reuse = 3283llama_memory_can_shift (llama_get_memory (ctx_tgt ))&& 3284 !slot .prompt .tokens .has_mtmd ; 3285 3286if (!can_cache_reuse && n_cache_reuse > 0 ) { 3287SLT_WRN (slot ,"cache reuse is not supported - ignoring n_cache_reuse = %d\n" ,n_cache_reuse ); 3288 } 3289 3290// reuse chunks from the cached prompt by shifting their KV cache in the new position 3291if (can_cache_reuse && n_cache_reuse > 0 ) { 3292GGML_ASSERT (!slot .prompt .tokens .has_mtmd ); 3293 3294size_t head_c = n_past ;// cache 3295size_t head_p = n_past ;// current prompt 3296 3297if (mctx ) { 3298// we should never reach this 3299GGML_ABORT ("not supported by multimodal" ); 3300 } 3301 3302SLT_DBG (slot ,"trying to reuse chunks with size > %d, n_past = %d\n" ,n_cache_reuse ,n_past ); 3303 3304while (head_c < slot .prompt .tokens .size ()&& 3305head_p < input_tokens .size ()) { 3306 3307size_t n_match = 0 ; 3308while (head_c + n_match < slot .prompt .tokens .size ()&& 3309head_p + n_match < input_tokens .size ()&& 3310slot .prompt .tokens [head_c + n_match ]== input_tokens [head_p + n_match ]) { 3311n_match ++ ; 3312 } 3313 3314if (n_match >= (size_t )n_cache_reuse ) { 3315SLT_TRC (slot ,"reusing chunk with size %zu, shifting KV cache [%zu, %zu) -> [%zu, %zu)\n" ,n_match ,head_c ,head_c + n_match ,head_p ,head_p + n_match ); 3316//for (size_t i = head_p; i < head_p + n_match; i++) { 3317// SLT_DBG(slot, "cache token %3zu: %6d '%s'\n", i, prompt_tokens[i], common_token_to_piece(ctx_tgt, prompt_tokens[i]).c_str()); 3318//} 3319 3320const int64_t kv_shift = (int64_t )head_p - (int64_t )head_c ; 3321 3322slot .mem .seq_rm (slot .id ,head_p ,head_c ); 3323slot .mem .seq_add (slot .id ,head_c ,head_c + n_match ,kv_shift ); 3324 3325for (size_t i = 0 ;i < n_match ;i ++ ) { 3326slot .prompt .tokens .set_token (head_p + i ,slot .prompt .tokens [head_c + i ]); 3327n_past ++ ; 3328 } 3329 3330head_c += n_match ; 3331head_p += n_match ; 3332 }else { 3333head_c += 1 ; 3334 } 3335 } 3336 3337SLT_DBG (slot ,"after context reuse, new n_past = %d\n" ,n_past ); 3338 } 3339 }else { 3340// if we don't cache the prompt, we have to remove all previous tokens 3341n_past = 0 ; 3342 } 3343 3344llama_pos pos_next = slot .prompt .tokens .pos_next (n_past ); 3345 3346// ref: https://github.com/ggml-org/llama.cpp/pull/24110 3347const bool has_new_tokens = (n_past < slot .task -> n_tokens ()); 3348 3349// the largest pos_min required for a checkpoint to be useful 3350const auto pos_min_thold = std::max (0 ,pos_next - n_swa - (has_new_tokens ?0 :1 )); 3351 3352if (n_past > 0 && n_past <=slot .prompt .n_tokens ()) { 3353const auto pos_min = llama_memory_seq_pos_min (llama_get_memory (ctx_tgt ),slot .id ); 3354if (pos_min == -1 ) { 3355SLT_ERR (slot ,"n_past = %d, slot.prompt.tokens.size() = %d, seq_id = %d, pos_min = %d\n" ,n_past , (int )slot .prompt .tokens .size (),slot .id ,pos_min ); 3356GGML_ABORT ("pos_min == -1, but n_past > 0 - should not happen: https://github.com/ggml-org/llama.cpp/pull/13833#discussion_r2116181237" ); 3357 } 3358 3359// when the prompt prefix does not match, print the tokens around the mismatch 3360// this is useful for debugging prompt caching 3361if (slots_debug ) { 3362const int np0 = std::max < int > (n_past - slots_n_diff ,0 ); 3363const int np1 = std::min < int > (n_past + slots_n_diff + 2 , std::min (slot .prompt .tokens .size (),slot .task -> tokens .size ())); 3364 3365 std::stringstream ss0 ; 3366 std::stringstream ss1 ; 3367 3368 std::stringstream st0 ; 3369 std::stringstream st1 ; 3370 3371ss0 <<"old: ... " ; 3372ss1 <<"new: ... " ; 3373 3374for (int i = np0 ;i < np1 ;i ++ ) { 3375if (i == n_past ) { 3376ss0 <<" | " ; 3377ss1 <<" | " ; 3378 } 3379 3380 { 3381const auto token = slot .prompt .tokens [i ]; 3382const auto piece = token != LLAMA_TOKEN_NULL ?common_token_to_piece (ctx_tgt ,token ) :"[mtmd]" ; 3383ss0 <<piece ; 3384st0 << std::setw (8 ) <<token ; 3385 } 3386 3387 { 3388const auto token = slot .task -> tokens [i ]; 3389const auto piece = token != LLAMA_TOKEN_NULL ?common_token_to_piece (ctx_tgt ,token ) :"[mtmd]" ; 3390ss1 <<piece ; 3391st1 << std::setw (8 ) <<token ; 3392 } 3393 } 3394 3395SLT_WRN (slot ,"%s\n" ,ss0 .str ().c_str ()); 3396SLT_WRN (slot ,"%s\n" ,ss1 .str ().c_str ()); 3397 3398SLT_WRN (slot ,"%s\n" ,st0 .str ().c_str ()); 3399SLT_WRN (slot ,"%s\n" ,st1 .str ().c_str ()); 3400 } 3401 3402if (pos_min >=pos_min_thold ) { 3403// search for a context checkpoint 3404const auto it = std::find_if ( 3405slot .prompt .checkpoints .rbegin (), 3406slot .prompt .checkpoints .rend (), 3407 [& ](const auto & cur ) { 3408// guarantee that a checkpoint will result in at least one token being processed [TAG_PROMPT_LOGITS] 3409SLT_TRC (slot ,"checking checkpoint with [%d, %d] against %d...\n" ,cur .pos_min ,cur .pos_max ,pos_min_thold ); 3410// workaround for [TAG_CHECKPOINTS_FIX_POS_MIN] 3411if (cur .pos_max > pos_next ) { 3412return false; 3413 } 3414return cur .pos_min < pos_min_thold || cur .pos_min == 0 ; 3415 } 3416 ); 3417 3418bool do_reset = it == slot .prompt .checkpoints .rend (); 3419 3420if (!do_reset ) { 3421// restore the context checkpoint 3422it -> load_tgt (ctx_tgt ,slot .id ,LLAMA_STATE_SEQ_FLAGS_PARTIAL_ONLY ); 3423it -> load_dft (ctx_dft ,slot .id ,LLAMA_STATE_SEQ_FLAGS_PARTIAL_ONLY ); 3424// restore the draft's speculative state 3425common_speculative_set_state (spec .get (),slot .id ,it -> data_spec ); 3426 3427pos_next = std::min (pos_next , std::max (it -> pos_min + 1 ,it -> pos_max )); 3428n_past = std::min (slot .prompt .tokens .size_up_to_pos (pos_next ), (size_t )it -> n_tokens ); 3429SLT_TRC (slot ,"restored context checkpoint (pos_min = %d, pos_max = %d, n_tokens = %" PRId64 ", n_past = %d, size = %.3f MiB)\n" ,it -> pos_min ,it -> pos_max ,it -> n_tokens ,n_past , (float )it -> size () /1024 /1024 ); 3430 } 3431 3432if (do_reset ) { 3433SLT_TRC (slot ,"forcing full prompt re-processing due to lack of cache data (likely due to SWA or hybrid/recurrent memory, see %s)\n" , 3434"https://github.com/ggml-org/llama.cpp/pull/13194#issuecomment-2868343055" ); 3435pos_next = 0 ; 3436n_past = 0 ; 3437 } 3438 } 3439 } 3440 3441 { 3442// erase any checkpoints with pos_max > pos_next 3443for (auto it = slot .prompt .checkpoints .begin ();it != slot .prompt .checkpoints .end ();) { 3444const auto & cur = * it ; 3445if (cur .pos_max > pos_next ) { 3446SLT_TRC (slot ,"erased invalidated context checkpoint (pos_min = %d, pos_max = %d, n_tokens = %" PRId64 ", n_swa = %d, pos_next = %d, size = %.3f MiB)\n" ,cur .pos_min ,cur .pos_max ,cur .n_tokens ,n_swa ,pos_next , (float )cur .size () /1024 /1024 ); 3447it = slot .prompt .checkpoints .erase (it ); 3448 }else { 3449++ it ; 3450 } 3451 } 3452 } 3453 } 3454 3455// [TAG_PROMPT_LOGITS] 3456if (n_past == slot .task -> n_tokens ()&& n_past > 0 ) { 3457SLT_WRN (slot ,"need to evaluate at least 1 token for each active slot (n_past = %d, task.n_tokens() = %d)\n" ,n_past ,slot .task -> n_tokens ()); 3458n_past -- ; 3459SLT_WRN (slot ,"n_past was set to %d\n" ,n_past ); 3460 } 3461 3462slot .stats .n_prompt_cached = n_past ; 3463slot .stats .n_prompt_processed = 0 ; 3464 3465metrics .add_prompt_cached (n_past ); 3466 3467slot .prompt .tokens .keep_first (n_past ); 3468 3469// this is to signal the client that the request has started processing 3470if (slot .task -> params .stream ) { 3471if (slot .task -> params .return_progress ) { 3472// send initial 0% progress update if needed 3473send_partial_response (slot , {}, true); 3474 }else { 3475// otherwise, for streaming without progress, signal HTTP to send the headers (i.e. 200 status) 3476send_partial_response (slot , {}, false, true); 3477 } 3478 } 3479 }// end of SLOT_STATE_STARTED 3480 3481if (!slot .can_split ()) { 3482// cannot fit the prompt in the current batch - will try next iter 3483if (batch .size ()+ slot .task -> n_tokens ()> n_batch ) { 3484return ; 3485 } 3486 } 3487 3488// note: the prompt timing is advanced in post_decode(), so it does not cover 3489// the tokens added to the batch below 3490slot .print_timings_pp (); 3491 3492// truncate any tokens that are beyond n_past for this slot 3493const llama_pos p0 = slot .prompt .tokens .pos_next (); 3494 3495SLT_TRC (slot ,"cached n_tokens = %d, memory_seq_rm [%d, end)\n" ,slot .prompt .n_tokens (),p0 ); 3496 3497slot .mem .seq_rm (slot .id ,p0 ,-1 ); 3498 3499// If using an alora, there may be uncached tokens that come 3500// before the invocation sequence. When this happens, the 3501// tokens before the invocation sequence need to be 3502// processed without the adapter in a separate batch, then 3503// the adapter needs to be enabled for the remaining tokens. 3504if (lora_all_alora (slot .lora )&& slot .alora_invocation_start - 1 > slot .prompt .n_tokens ()) { 3505SLT_DBG (slot ,"processing pre-alora tokens without the adapter (n_tokens = %d, alora_invocation_start = %d)\n" ,slot .prompt .n_tokens (),slot .alora_invocation_start ); 3506const auto & enabled_loras = lora_get_enabled_ids (slot .lora ); 3507GGML_ASSERT (enabled_loras .size ()== 1 ); 3508alora_scale = slot .lora [enabled_loras [0 ]].scale ; 3509slot .lora [enabled_loras [0 ]].scale = 0.0f ; 3510alora_disabled_id = enabled_loras [0 ]; 3511 } 3512 3513bool do_checkpoint = params_base .n_ctx_checkpoints > 0 ; 3514 3515// make checkpoints only for completion tasks 3516do_checkpoint = do_checkpoint && slot .task -> type == SERVER_TASK_TYPE_COMPLETION ; 3517 3518// make a checkpoint of the parts of the memory that cannot be rolled back. 3519// checkpoints are created only if: 3520// - the model does not support partial sequence removal 3521// - the model uses SWA (and we are not using `swa_full`) 3522// - the model supports partial sequence removal but only up to a fixed bound 3523do_checkpoint = do_checkpoint && ( 3524ctx_tgt_seq_rm_type == COMMON_CONTEXT_SEQ_RM_TYPE_FULL || 3525ctx_tgt_seq_rm_type == COMMON_CONTEXT_SEQ_RM_TYPE_RS || 3526n_swa > 0 ); 3527 3528bool has_mtmd = false; 3529 3530// check if we should process the mtmd chunk 3531while (true) { 3532auto cur_token_idx = slot .prompt .n_tokens (); 3533if ( 3534cur_token_idx >=slot .task -> n_tokens ()|| 3535input_tokens [cur_token_idx ]!= LLAMA_TOKEN_NULL // encountered a text token 3536 ) { 3537break ; 3538 } 3539 3540// process the mtmd chunk 3541// note: it submits its own decode, potentially be async 3542// so the timing is queued and flushed on the next sync 3543metrics_pre_decode (); 3544 3545// encode on the worker thread, so we can still handle metrics tasks 3546size_t n_tokens_out = 0 ; 3547int32_t res = 0 ; 3548queue_tasks .yield_to_queue ([& ]() { 3549res = process_mtmd_chunk (slot ,slot .mbatch ,cur_token_idx ,n_tokens_out ); 3550 }); 3551 3552if (res != 0 ) { 3553SLT_ERR (slot ,"failed to process mtmd chunk, res = %d\n" ,res ); 3554send_error (slot ,"failed to process mtmd chunk" ,ERROR_TYPE_SERVER ); 3555slot .release (); 3556return ;// the slot is done, skip it entirely 3557 } 3558 3559metrics_queue_prompt (n_tokens_out ); 3560slot .stats .n_prompt_processed += n_tokens_out ; 3561slot .stats .update_prompt_last (); 3562 3563// add the mtmd chunk to cache 3564 { 3565const auto & chunk = input_tokens .find_chunk (cur_token_idx ); 3566// the chunk is already in the KV cache at this point, so we don't need to keep its data around 3567slot .prompt .tokens .push_back_placeholder (chunk .get ()); 3568 } 3569 3570has_mtmd = true; 3571 } 3572 3573const auto & spans = slot .task -> params .message_spans ; 3574const auto last_user_pos = spans .last_user_message_pos (); 3575 3576// add prompt tokens for processing in the current batch 3577while (slot .prompt .n_tokens ()< slot .task -> n_tokens ()&& batch .size ()< n_batch ) { 3578// get next token to process 3579llama_token cur_tok = input_tokens [slot .prompt .n_tokens ()]; 3580if (cur_tok == LLAMA_TOKEN_NULL ) { 3581break ;// end of text chunk 3582 } 3583 3584// if this is an alora request with pre-invocation 3585// tokens that are not cached, we need to stop filling 3586// this batch at those pre-invocation tokens. 3587if (alora_scale > 0 && slot .prompt .n_tokens ()== slot .alora_invocation_start - 1 ) { 3588SLT_DBG (slot ,"stop prompt batch filling at (n_tokens = %d, alora_invocation_start = %d)\n" ,slot .prompt .n_tokens (),slot .alora_invocation_start ); 3589break ; 3590 } 3591 3592// embedding requires all tokens in the batch to be output; 3593// MTP also wants logits at every prompt position so the 3594// streaming hook can mirror t_h_nextn into ctx_dft. 3595add_ok &=batch .add (slot .id , 3596cur_tok , 3597/* pos = */ slot .prompt .tokens .pos_next (), 3598/* output = */ slot .need_embd (), 3599/* is_prompt = */ true); 3600slot .prompt .tokens .push_back (cur_tok ); 3601 3602// break at the last user message, or at user messages at least min step past the last checkpoint 3603if (do_checkpoint && spans .is_user_start (slot .prompt .n_tokens ())) { 3604const auto pos = slot .prompt .n_tokens (); 3605const auto & checkpoints = slot .prompt .checkpoints ; 3606 3607if (pos == last_user_pos || checkpoints .empty ()|| pos > checkpoints .back ().n_tokens + params_base .checkpoint_min_step ) { 3608break ; 3609 } 3610 } 3611 3612// process the last few tokens of the prompt separately in order to allow for a checkpoint to be created. 3613// create checkpoints that many tokens before the end of the prompt: 3614// - 4 + n_ubatch 3615// - 4 3616// ref: https://github.com/ggml-org/llama.cpp/pull/20288 3617if (do_checkpoint ) { 3618static const int checkpoint_offsets []= {4 + n_ubatch ,4 }; 3619 3620bool should_break = false; 3621for (int offset :checkpoint_offsets ) { 3622const int n_last = std::min (n_batch ,offset ); 3623if (slot .task -> n_tokens ()== slot .prompt .n_tokens ()+ n_last ) { 3624should_break = true; 3625break ; 3626 } 3627 } 3628if (should_break ) { 3629break ; 3630 } 3631 } 3632 } 3633 3634// the number of tokens added to the batch for the current slot 3635const auto n_tokens_cur = batch .size ()- n_tokens_prev ; 3636 3637const auto n_tokens_start = slot .prompt .n_tokens ()- n_tokens_cur ; 3638 3639const bool near_prompt_end = slot .task -> n_tokens ()< slot .prompt .n_tokens ()+ n_ubatch ; 3640 3641const bool is_user_start = spans .is_user_start (n_tokens_start ); 3642const bool is_last_user_message = n_tokens_start == last_user_pos ; 3643 3644// entire prompt has been processed 3645if (slot .prompt .n_tokens ()== slot .task -> n_tokens ()) { 3646slot .state = SLOT_STATE_DONE_PROMPT ; 3647 3648GGML_ASSERT (batch .size ()> 0 ); 3649 3650// extract the logits only for the last token 3651batch .set_output (batch .size ()- 1 , true); 3652 3653slot .stats .n_gen = 0 ; 3654slot .i_batch = batch .size ()- 1 ; 3655 3656if (!encoder_decoder ) { 3657slot .init_sampler (); 3658 } 3659 }else { 3660// skip ordinary mid-prompt checkpoints, unless the batch starts a user 3661// message or we are near the end of the prompt 3662if (!is_user_start && !near_prompt_end ) { 3663do_checkpoint = false; 3664 } 3665 } 3666 3667const auto pos_min = llama_memory_seq_pos_min (llama_get_memory (ctx_tgt ),slot .id ); 3668const auto pos_max = llama_memory_seq_pos_max (llama_get_memory (ctx_tgt ),slot .id ); 3669 3670// nothing to checkpoint yet 3671// TODO: is this check needed? 3672if (do_checkpoint && pos_min < 0 ) { 3673do_checkpoint = false; 3674 } 3675 3676// do not checkpoint after mtmd chunks 3677do_checkpoint = do_checkpoint && !has_mtmd ; 3678 3679// no need to create checkpoints that are too close together, unless it's the last user message 3680do_checkpoint = do_checkpoint && ( 3681slot .prompt .checkpoints .empty ()|| 3682is_last_user_message || near_prompt_end || 3683n_tokens_start > slot .prompt .checkpoints .back ().n_tokens + params_base .checkpoint_min_step ); 3684SLT_DBG (slot ,"main/do_checkpoint = %s, pos_min = %d, pos_max = %d\n" ,do_checkpoint ?"yes" :"no" ,pos_min ,pos_max ); 3685 3686// note: we create the checkpoint before calling llama_decode(), so the current batch is not 3687// yet processed and therefore it is not part of the checkpoint. 3688if (do_checkpoint ) { 3689create_checkpoint (slot ,n_tokens_cur ,pos_min ,pos_max ); 3690 } 3691 } 3692 3693if (!slot_batched ) { 3694slot_batched = & slot ; 3695 } 3696 }); 3697 } 3698 } 3699 3700// returns true = success ; false = retry with smaller batch size 3701// throw std::runtime_error on fatal error 3702bool decode (int32_t & n_batch ,int32_t off ,llama_batch & batch_view ) { 3703SRV_DBG ("n_batch (effective) = %d, off = %d\n" ,n_batch ,off ); 3704 3705metrics_pre_decode (); 3706 3707if (batch .size ()== 0 ) { 3708SRV_WRN ("%s" ,"no tokens to decode\n" ); 3709 3710if (++ n_empty_consecutive > 3 ) { 3711GGML_ABORT ("fatal error - please provide logs and repro in %s\n" ,"https://github.com/ggml-org/llama.cpp/pull/20277" ); 3712 } 3713 3714return true;// nothing to decode 3715 }else { 3716n_empty_consecutive = 0 ; 3717 } 3718 3719// TODO @ngxson : dft model may have different n_embd than the tgt model, so we check & reject if that's the case 3720// this case is not currently used by any models, but may need to be supported in the future 3721if (spec && batch .has_embd ) { 3722if (llama_model_n_embd_inp (model_dft )!= llama_model_n_embd_inp (model_tgt )) { 3723SRV_ERR ("%s" ,"unsupported batch.has_embd + spec case\n" ); 3724throw std::runtime_error ("unsupported batch.has_embd + spec case" ); 3725 } 3726 } 3727 3728bool has_output = false; 3729for (int i = off ;i < off + batch_view .n_tokens ;++ i ) { 3730has_output |=batch .tokens [i ].output ; 3731 } 3732 3733// yield to the queue, so we can still handle metrics tasks while decoding 3734// note: the sync is done here too, so that the wait is also covered by the yield 3735int ret = 0 ; 3736queue_tasks .yield_to_queue ([& ]() { 3737if (encoder_decoder && slots [0 ].state == SLOT_STATE_DONE_PROMPT ) { 3738auto & slot = slots [0 ]; 3739auto encoder_batch = batch_view ; 3740encoder_batch .logits = nullptr ; 3741ret = llama_encode (ctx_tgt ,encoder_batch ); 3742if (ret != 0 ) { 3743throw std::runtime_error ("failed to encode prompt" ); 3744 } 3745 3746// The encoder prompt is not part of the decoder's KV history. 3747slot .prompt .clear (); 3748slot .prompt .tokens .insert (decoder_prefix ); 3749slot .init_sampler (); 3750// Quantized MoE routing is sensitive to batching; keep prefix evaluation incremental. 3751for (auto & token :decoder_prefix ) { 3752auto decoder_batch = llama_batch_get_one (& token ,1 ); 3753ret = llama_decode (ctx_tgt ,decoder_batch ); 3754if (ret != 0 ) { 3755break ; 3756 } 3757 } 3758slot .i_batch = off ; 3759 }else { 3760ret = llama_decode (ctx_tgt ,batch_view ); 3761 } 3762if (ret == 0 && has_output ) { 3763llama_synchronize (ctx_tgt ); 3764 } 3765 }); 3766 3767if (ret != 0 ) { 3768 { 3769 std::string err ; 3770 3771if (n_batch == 1 && ret == 1 ) { 3772// TODO: try to terminate only the largest active slot/sequence and continue with the rest 3773// need to remove the tokens from the current batch too 3774err = "Context size has been exceeded." ; 3775 } 3776 3777if (ret == -1 ) { 3778err = "Invalid input batch." ; 3779 } 3780 3781if (ret < -1 ) { 3782// TODO: update slot state based on llama_memory_seq_pos_min() and llama_memory_seq_pos_max() 3783err = "Compute error." ; 3784 } 3785 3786// TODO: handle ret == 2 (abort) when we start aborting 3787 3788if (!err .empty ()) { 3789SRV_ERR ("%s off = %d, n_batch = %d, ret = %d\n" ,err .c_str (),off ,n_batch ,ret ); 3790 3791for (auto & slot :slots ) { 3792if (slot .is_processing ()) { 3793send_error (slot ,err ); 3794slot .release (); 3795 3796// note: it's complicated to keep track of how much of the current batch has been 3797// processed before the error occurred, so we simply clear the entire context 3798slot .prompt_clear (); 3799 } 3800 } 3801 3802// stop, do not retry with smaller batch size 3803throw std::runtime_error (err ); 3804 } 3805 } 3806 3807// retry with half the batch size to try to find a free slot in the KV cache 3808if (!try_clear_idle_slots ()) { 3809n_batch /=2 ; 3810 } 3811 3812SRV_WRN ("failed to find free space in the KV cache, retrying with smaller batch size, off = %d, n_batch = %d, ret = %d\n" ,off ,n_batch ,ret ); 3813 3814return false;// retry with the updated n_batch 3815 }else { 3816// success, apply batch metrics 3817metrics_post_decode (off ,batch_view .n_tokens ,has_output ); 3818 } 3819 3820// TODO: avoid restoring the draft context and re-evaluating the drafted tokens when not needed [TAG_SPEC_AVOID_DRAFT_REEVAL] 3821// for now, always re-evaluate for simplicity 3822// ref: https://github.com/ggml-org/llama.cpp/pull/22728#issuecomment-4400925384 3823if (spec ) { 3824bool ok = true; 3825queue_tasks .yield_to_queue ([& ]() { 3826ok = common_speculative_process (spec .get (),batch_view ); 3827 }); 3828 3829if (!ok ) { 3830SRV_ERR ("%s" ,"failed to process speculative batch\n" ); 3831 3832// TODO: handle error 3833throw std::runtime_error ("failed to process speculative batch" ); 3834 } 3835 } 3836 3837// handle `n_cmpl > 1` tasks - when the main prompt is processed, activate all child tasks too 3838for (auto & slot :slots ) { 3839if (slot .state == SLOT_STATE_DONE_PROMPT && slot .task -> is_parent ()) { 3840 std::vector < server_slot *> children ; 3841for (auto & other :slots ) { 3842if (other .state == SLOT_STATE_WAIT_OTHER && slot .task -> id == other .task -> id_parent ) { 3843children .push_back (& other ); 3844 } 3845 } 3846 3847// all children slots should already launched by launch_slots_with_parent_task() 3848// copy state to the child slots 3849for (auto & child :children ) { 3850SLT_TRC (slot ," - copying state to child %d\n" ,child -> id ); 3851 3852GGML_ASSERT (child -> state == SLOT_STATE_WAIT_OTHER ); 3853 3854slot .copy_state_to (* child ); 3855child -> state = SLOT_STATE_DONE_PROMPT ; 3856 } 3857 } 3858 } 3859 3860return true; 3861 } 3862 3863void post_decode (int32_t n_batch_tokens ,int32_t off ,llama_batch & batch_view ) { 3864// for checking if a given batch index is inside batch_view 3865auto is_inside_view = [& ](int32_t idx ) { 3866return idx >=off && idx < off + n_batch_tokens ; 3867 }; 3868 3869// TODO @ngxson : it's tricky to make sub-batch compatible with common_sampler_sample_and_accept_n, 3870// so for now we will throw an error in this case: https://github.com/ggml-org/llama.cpp/issues/24840 3871iterate (slots , [& ](server_slot & slot ) { 3872for (auto & i :slot .spec_i_batch ) { 3873if (!is_inside_view (i )) { 3874throw std::runtime_error (string_format ("speculative batch index %d is not inside the current sub-batch [%d, %d)" ,i ,off ,off + n_batch_tokens )); 3875 } 3876 } 3877 }); 3878 3879auto accept_special_token = [& ](server_slot & slot ,llama_token token ) { 3880return params_base .special || 3881slot .task -> params .sampling .preserved_tokens .find (token )!= slot .task -> params .sampling .preserved_tokens .end (); 3882 }; 3883 3884iterate (slots , [& ](server_slot & slot ) { 3885// optionally send prompt processing progress 3886if (slot .state == SLOT_STATE_PROCESSING_PROMPT || slot .state == SLOT_STATE_DONE_PROMPT ) { 3887if (slot .task -> params .stream && slot .task -> params .return_progress ) { 3888send_partial_response (slot , {}, true); 3889 } 3890 } 3891 3892if (!is_inside_view (slot .i_batch )) { 3893// the required token not in this sub-batch, skip 3894return ; 3895 } 3896 3897if (slot .state == SLOT_STATE_DONE_PROMPT ) { 3898if (slot .task -> type == SERVER_TASK_TYPE_EMBEDDING ) { 3899// prompt evaluated for embedding 3900send_embedding (slot ,batch_view ); 3901slot .release (); 3902slot .i_batch = -1 ; 3903return ; 3904 } 3905 3906if (slot .task -> type == SERVER_TASK_TYPE_RERANK ) { 3907send_rerank (slot ,batch_view ); 3908slot .release (); 3909slot .i_batch = -1 ; 3910return ; 3911 } 3912 3913GGML_ASSERT (slot .task -> need_sampling ()); 3914 3915// prompt evaluated for next-token prediction 3916slot .state = SLOT_STATE_GENERATING ; 3917 3918if (slot .can_speculate ()) { 3919common_speculative_begin (spec .get (),slot .id ,slot .prompt .tokens .get_text_tokens ()); 3920 } 3921 }else if (slot .state != SLOT_STATE_GENERATING ) { 3922return ; 3923 } 3924 3925if (slot .can_speculate ()&& !slot .spec_draft .empty ()) { 3926return ;// sample using speculative decoding 3927 } 3928 3929// shifted according to the current sub-batch 3930const int tok_idx = slot .i_batch - off ; 3931 3932llama_token id ; 3933 { 3934scoped_timer timer (t_sampl ,n_sampl ); 3935id = common_sampler_sample (slot .smpl .get (),slot .ctx_tgt ,tok_idx ); 3936 } 3937 3938slot .i_batch = -1 ; 3939 3940common_sampler_accept (slot .smpl .get (),id , true); 3941 3942// here we have synchronized the llama_context (due to the sampling above), so we can do time measurement 3943const int64_t t_now = ggml_time_us (); 3944 3945slot .stats .n_gen += 1 ; 3946 3947if (slot .stats .n_gen == 1 ) { 3948slot .stats .update_prompt_last (); 3949slot .t_print_last = t_now ; 3950slot .n_gen_last = 0 ; 3951 } 3952 3953slot .stats .update_gen_last (); 3954 3955completion_token_output result ; 3956result .tok = id ; 3957result .text_to_send = common_token_to_piece (slot .ctx_tgt ,result .tok ,accept_special_token (slot ,result .tok )); 3958result .prob = 1.0f ;// TODO: set it here instead of doing inside populate_token_probs 3959 3960if (slot .task -> params .sampling .n_probs > 0 ) { 3961populate_token_probs (slot ,result ,slot .task -> params .post_sampling_probs ,params_base .special ,tok_idx ); 3962 } 3963 3964if (!process_token (result ,slot )) { 3965// release slot because of stop condition 3966slot .print_timings (); 3967send_final_response (slot ); 3968slot .release (); 3969 3970return ; 3971 } 3972 3973slot .print_timings_tg (); 3974 }); 3975 3976// speculative decoding - main model sample and accept 3977iterate (slots , [& ](server_slot & slot ) { 3978if (slot .state != SLOT_STATE_GENERATING || !slot .can_speculate ()|| 3979slot .spec_draft .empty ()|| slot .spec_i_batch .empty ()) { 3980return ; 3981 } 3982 3983// save the original draft size 3984const size_t n_draft = slot .spec_draft .size (); 3985 3986GGML_ASSERT (n_draft > 0 ); 3987 3988// verify and try to accept the draft 3989 { 3990common_sampler_ptr smpl_save (common_sampler_clone (slot .smpl .get ())); 3991 3992GGML_ASSERT (slot .spec_i_batch .size ()== n_draft + 1 ); 3993const auto & synth_probs = common_speculative_get_synth_probs (spec .get ()); 3994auto accepted = synth_probs .empty () 3995 ?common_sampler_sample_and_accept_n (slot .smpl .get (),slot .ctx_tgt ,slot .spec_i_batch ,slot .spec_draft ) 3996 :server_sample_and_accept_synth ( 3997slot .smpl .get (),slot .ctx_tgt ,slot .spec_i_batch ,slot .spec_draft , 3998synth_probs ,slot .spec_synth_rng ,slot .spec_is_replay ); 3999slot .spec_i_batch .clear (); 4000 4001GGML_ASSERT (accepted .size () >=1 ); 4002 4003const uint32_t n_rollback = slot .spec_draft .size ()+ 1 - accepted .size (); 4004 4005const bool use_ckpt_tgt = 4006ctx_tgt_seq_rm_type == COMMON_CONTEXT_SEQ_RM_TYPE_FULL || 4007 (ctx_tgt_seq_rm_type == COMMON_CONTEXT_SEQ_RM_TYPE_RS && n_rollback > llama_n_rs_seq (ctx_tgt )); 4008 4009// check for partial draft acceptance 4010if (n_rollback > 0 ) { 4011if (use_ckpt_tgt ) { 4012if (trace > 0 ) { 4013SLT_INF (slot ,"accepted %2zu/%2zu draft tokens (restore checkpoint)\n" ,accepted .size ()- 1 ,slot .spec_draft .size ()); 4014 } 4015 4016// partial acceptance is not supported by the context -> truncate the draft and restore the state 4017slot .spec_is_replay = true; 4018slot .spec_draft = std::move (accepted ); 4019 4020const auto & ckpt = slot .spec_ckpt ; 4021 4022SLT_DBG (slot ,"restoring speculative checkpoint (pos_min = %d, pos_max = %d, size = %zu)\n" ,ckpt .pos_min ,ckpt .pos_max ,ckpt .size ()); 4023 4024ckpt .load_tgt (slot .ctx_tgt ,slot .id ,LLAMA_STATE_SEQ_FLAGS_PARTIAL_ONLY ); 4025 4026if (slot .ctx_dft ) { 4027ckpt .load_dft (slot .ctx_dft ,slot .id ,LLAMA_STATE_SEQ_FLAGS_PARTIAL_ONLY ); 4028 } 4029 4030slot .mem .seq_rm (slot .id ,ckpt .pos_max + 1 ,-1 ); 4031 4032slot .prompt .tokens .keep_first (ckpt .n_tokens ); 4033common_sampler_copy (smpl_save .get (),slot .smpl .get ()); 4034 4035return ; 4036 } 4037 } 4038 4039if (trace > 0 ) { 4040SLT_INF (slot ,"accepted %2zu/%2zu draft tokens\n" ,accepted .size ()- 1 ,n_draft ); 4041 } 4042 4043common_speculative_accept (spec .get (),slot .id ,accepted .size ()- 1 ); 4044 4045slot .spec_draft = std::move (accepted ); 4046 } 4047 4048const auto ids = std::move (slot .spec_draft ); 4049 4050size_t n_accepted = ids .size ()- 1 ; 4051if (slot .spec_is_replay && n_accepted > 0 ) { 4052n_accepted -- ; 4053 } 4054slot .spec_is_replay = false; 4055 4056slot .stats .update_gen_last (); 4057 4058// update how many tokens out of those tested were accepted 4059slot .stats .n_draft_accepted += n_accepted ; 4060slot .stats .n_draft_verif_steps += 1 ; 4061 4062auto & n_accepted_per_pos = slot .n_accepted_per_pos ; 4063if (n_accepted_per_pos .empty ()) { 4064n_accepted_per_pos .resize (common_speculative_n_max (spec .get ()),0 ); 4065 } 4066for (size_t i = 0 ;i < n_accepted && i < n_accepted_per_pos .size ();++ i ) { 4067n_accepted_per_pos [i ]++ ; 4068 } 4069 4070// add accepted tokens to the prompt 4071slot .prompt .tokens .keep_first (slot .prompt .n_tokens ()- n_draft ); 4072slot .prompt .tokens .insert ({ids .begin (),ids .end ()- 1 }); 4073 4074slot .sampled = ids .back ();// last accepted token 4075SLT_DBG (slot ,"add accepted tokens: sampled=%d, ids.size=%zu, n_draft=%zu\n" ,slot .sampled ,ids .size (),n_draft ); 4076 4077slot .mem .seq_rm (slot .id ,slot .prompt .tokens .pos_next (),-1 ); 4078 4079for (size_t i = 0 ;i < ids .size ();++ i ) { 4080completion_token_output result ; 4081 4082result .tok = ids [i ]; 4083result .text_to_send = common_token_to_piece (slot .ctx_tgt ,result .tok ,accept_special_token (slot ,result .tok )); 4084result .prob = 1.0f ;// set later 4085 4086// TODO: set result.probs 4087 4088slot .stats .n_gen += 1 ; 4089 4090if (!process_token (result ,slot )) { 4091slot .print_timings (); 4092send_final_response (slot ); 4093slot .release (); 4094 4095return ; 4096 } 4097 } 4098 4099slot .print_timings_tg (); 4100 4101SLT_DBG (slot ,"accepted %d/%d draft tokens, new n_tokens = %d\n" , (int )n_accepted , (int )n_draft ,slot .prompt .n_tokens ()); 4102 }); 4103 } 4104 4105// context size of a single slot, capped by --kv-unified-per-slot and by the training context of the model 4106int n_ctx_slot ()const { 4107int res = llama_n_ctx_seq (ctx_tgt ); 4108 4109if (params_base .kv_unified_per_slot > 0 ) { 4110res = std::min (res ,params_base .kv_unified_per_slot ); 4111 } 4112 4113return std::min (res ,llama_model_n_ctx_train (model_tgt )); 4114 } 4115 4116server_response_reader get_response_reader () { 4117return server_response_reader (queue_tasks ,queue_results ,HTTP_POLLING_SECONDS ); 4118 } 4119 4120// 4121// metrics helpers 4122// 4123 4124// call before submitting a decode, so that the queued prompt stats can be timed 4125void metrics_pre_decode () { 4126t_decode_start = ggml_time_us (); 4127 } 4128 4129// the batch is submitted, but its compute may not be done yet 4130void metrics_queue_prompt (uint64_t n_tokens ) { 4131if (n_tokens == 0 ) { 4132return ; 4133 } 4134if (n_prompt_queued == 0 ) { 4135t_prompt_start = t_decode_start ; 4136 } 4137n_prompt_queued += n_tokens ; 4138 } 4139 4140// call only after the context is synchronized, otherwise the time is meaningless 4141void metrics_flush_prompt () { 4142if (n_prompt_queued == 0 ) { 4143return ; 4144 } 4145metrics .add_prompt (n_prompt_queued ,ggml_time_us ()- t_prompt_start ); 4146n_prompt_queued = 0 ; 4147 } 4148 4149// has_output is computed by the caller, which also already synchronized the context if it is set 4150void metrics_post_decode (int32_t off ,int32_t n_tokens ,bool has_output ) { 4151metrics .n_decode ++ ; 4152for (const auto & slot :slots ) { 4153if (slot .is_processing ()) { 4154metrics .n_busy_slots ++ ; 4155 } 4156metrics .n_tokens_max = std::max (metrics .n_tokens_max , (uint64_t )slot .prompt .n_tokens ()); 4157 } 4158 4159// apply enqueued prompt tokens stats 4160// note: a slot can be released before we get here, which clears its stats 4161// the tokens were still computed, counted in the global metrics, not in slot 4162uint64_t n_prompt_tokens = 0 ; 4163 4164for (int i = off ;i < off + n_tokens ;++ i ) { 4165const auto & t = batch .tokens [i ]; 4166 4167if (!t .is_prompt ) { 4168continue ;// generated tokens are handled after sampling 4169 } 4170 4171n_prompt_tokens ++ ; 4172 4173auto & slot = slots [t .id_slot ]; 4174if (slot .stats .is_set ()) { 4175slot .stats .n_prompt_processed ++ ; 4176 } 4177 } 4178 4179metrics_queue_prompt (n_prompt_tokens ); 4180 4181if (has_output ) { 4182// the context is already synchronized, so the timings are correct 4183metrics_flush_prompt (); 4184 } 4185 4186// advance the prompt timing of the slots that had tokens in this batch 4187// note: a second pass, it must run after the sync to reflect the compute 4188const int64_t t_now = ggml_time_us (); 4189for (int i = off ;i < off + n_tokens ;++ i ) { 4190const auto & t = batch .tokens [i ]; 4191auto & slot = slots [t .id_slot ]; 4192if (t .is_prompt && slot .stats .is_set ()) { 4193slot .stats .set_prompt_last (t_now ); 4194 } 4195 } 4196 } 4197 4198// flush any queued prompt metrics if all slots are now idle 4199void metrics_flush_idle () { 4200if (n_prompt_queued == 0 ) { 4201return ; 4202 } 4203 4204llama_synchronize (ctx_tgt ); 4205metrics_flush_prompt (); 4206 } 4207 4208void metrics_on_prediction (const server_slot & slot ) { 4209const uint64_t t_us = slot .stats .t_gen_us (); 4210const uint64_t n = slot .stats .n_gen ; 4211const uint64_t n_steps = slot .stats .n_gen_steps (); 4212 4213metrics .predict .add (n ,n_steps ,t_us ); 4214metrics .predict_bucket .add (n ,n_steps ,t_us ); 4215 4216metrics .n_draft_tokens += slot .stats .n_draft_tokens ; 4217metrics .n_draft_accepted += slot .stats .n_draft_accepted ; 4218metrics .n_draft_verif_steps += slot .stats .n_draft_verif_steps ; 4219 4220auto & dst = metrics .n_accepted_per_pos ; 4221const auto & src = slot .n_accepted_per_pos ; 4222 4223if (dst .size ()< src .size ()) { 4224dst .resize (src .size (),0 ); 4225 } 4226for (size_t i = 0 ;i < src .size ();i ++ ) { 4227dst [i ]+= src [i ]; 4228 } 4229 } 4230}; 4231 4232// 4233// server_context (public API) 4234// 4235 4236server_context::server_context () :impl (new server_context_impl ()) {} 4237server_context::~server_context ()= default ; 4238 4239bool server_context::load_model (common_params & params ) { 4240return impl -> load_model (params ); 4241} 4242 4243void server_context::start_loop () { 4244auto & params = impl -> params_base ; 4245impl -> queue_tasks .start_loop (params .sleep_idle_seconds * 1000 ); 4246} 4247 4248void server_context::terminate () { 4249impl -> queue_tasks .terminate (); 4250} 4251 4252llama_context * server_context::get_llama_context ()const { 4253return impl -> ctx_tgt ; 4254} 4255 4256server_response_reader server_context::get_response_reader () { 4257return impl -> get_response_reader (); 4258} 4259 4260server_context_meta server_context::get_meta ()const { 4261auto bos_id = llama_vocab_bos (impl -> vocab ); 4262auto eos_id = llama_vocab_eos (impl -> vocab ); 4263auto bos_token_str = bos_id != LLAMA_TOKEN_NULL ?common_token_to_piece (impl -> ctx_tgt ,bos_id , true) :"" ; 4264auto eos_token_str = eos_id != LLAMA_TOKEN_NULL ?common_token_to_piece (impl -> ctx_tgt ,eos_id , true) :"" ; 4265 4266const char * ftype_name = llama_ftype_name (llama_model_ftype (impl -> model_tgt )); 4267 4268return server_context_meta { 4269/* build_info */ std::string (llama_build_info ()), 4270/* model_name */ impl -> model_name , 4271/* model_aliases */ impl -> model_aliases , 4272/* model_tags */ impl -> model_tags , 4273/* model_path */ impl -> params_base .model .path , 4274/* has_mtmd */ impl -> mctx != nullptr , 4275/* has_inp_image */ impl -> chat_params .allow_image , 4276/* has_inp_audio */ impl -> chat_params .allow_audio , 4277/* has_inp_video */ impl -> chat_params .allow_video , 4278/* json_ui_settings */ impl -> json_ui_settings , 4279/* slot_n_ctx */ impl -> n_ctx_slot (), 4280/* pooling_type */ llama_pooling_type (impl -> ctx_tgt ), 4281 4282/* chat_params */ impl -> chat_params , 4283/* chat_template_caps */ common_chat_templates_get_caps (impl -> chat_params .tmpls .get ()), 4284 4285/* bos_token_str */ bos_token_str , 4286/* eos_token_str */ eos_token_str , 4287/* fim_pre_token */ llama_vocab_fim_pre (impl -> vocab ), 4288/* fim_sub_token */ llama_vocab_fim_suf (impl -> vocab ), 4289/* fim_mid_token */ llama_vocab_fim_mid (impl -> vocab ), 4290/* fim_pad_token */ llama_vocab_fim_pad (impl -> vocab ), 4291/* fim_rep_token */ llama_vocab_fim_rep (impl -> vocab ), 4292/* fim_sep_token */ llama_vocab_fim_sep (impl -> vocab ), 4293 4294/* logit_bias_eog */ impl -> params_base .sampling .logit_bias_eog , 4295 4296/* model_vocab_type */ llama_vocab_type (impl -> vocab ), 4297/* model_vocab_n_tokens */ llama_vocab_n_tokens (impl -> vocab ), 4298/* model_n_ctx_train */ llama_model_n_ctx_train (impl -> model_tgt ), 4299/* model_n_embd_inp */ llama_model_n_embd (impl -> model_tgt ), 4300/* model_n_params */ llama_model_n_params (impl -> model_tgt ), 4301/* model_size */ llama_model_size (impl -> model_tgt ), 4302/* model_ftype */ ftype_name , 4303 }; 4304} 4305 4306// generator-like API for HTTP response generation 4307// may have bypass_sleep = true if the task does not use ctx_server 4308struct server_res_generator :server_res_spipe { 4309server_response_reader rd ; 4310server_res_generator (server_queue & queue_tasks ,server_response & queue_results ,int sleep_idle_seconds ,bool bypass_sleep = false) 4311 :rd (queue_tasks ,queue_results ,HTTP_POLLING_SECONDS ) { 4312// fast path in case sleeping is disabled 4313bypass_sleep |=sleep_idle_seconds < 0 ; 4314if (!bypass_sleep ) { 4315queue_tasks .wait_until_no_sleep (); 4316 } 4317 } 4318void ok (const json & response_data ) { 4319status = 200 ; 4320data = safe_json_to_str (response_data ); 4321 } 4322void error (const json & error_data ) { 4323status = json_value (error_data ,"code" ,500 ); 4324data = safe_json_to_str ({{"error" ,error_data }}); 4325 } 4326}; 4327 4328void server_context::set_state_callback (server_state_callback_t callback ) { 4329impl -> callback_state = std::move (callback ); 4330} 4331 4332// 4333// server_routes 4334// 4335 4336std::unique_ptr < server_res_generator > server_routes::handle_completions_impl ( 4337const server_http_req & req , 4338server_task_type type , 4339const json & data , 4340const std::vector < raw_buffer > & files , 4341task_response_type res_type ) { 4342GGML_ASSERT (type == SERVER_TASK_TYPE_COMPLETION || type == SERVER_TASK_TYPE_INFILL ); 4343 4344auto res = create_response (); 4345auto completion_id = gen_chatcmplid (); 4346auto & rd = res -> rd ; 4347auto & params = this -> params ; 4348 4349res -> set_req (& req );// will also set spipe if needed 4350 4351int32_t sse_ping_interval = params .sse_ping_interval ; 4352 4353try { 4354 std::vector < server_task > tasks ; 4355 4356const auto & prompt = data .at ("prompt" ); 4357// TODO: this log can become very long, put it behind a flag or think about a more compact format 4358//SRV_DBG("Prompt: %s\n", prompt.is_string() ? prompt.get<std::string>().c_str() : prompt.dump(2).c_str()); 4359 4360if (!params .path_prompts_log_dir .empty ()) { 4361const auto file_path = std::filesystem::path (params .path_prompts_log_dir ) /string_format ("%012" PRId64 ".txt" ,ggml_time_ms ()); 4362 std::ofstream f (file_path ); 4363if (f ) { 4364f << (prompt .is_string () ?prompt .get < std::string > ().c_str () :prompt .dump (2 ).c_str ()); 4365 }else { 4366SRV_ERR ("failed to create %s\n" ,file_path .string ().c_str ()); 4367 } 4368 } 4369 4370// process prompt 4371 std::vector < server_tokens > inputs ; 4372 4373if (res_type != TASK_RESPONSE_TYPE_NONE && ctx_server .mctx != nullptr ) { 4374// This is the case used by OAI compatible chat path with MTMD. TODO It can be moved to the path below. 4375inputs .push_back (process_mtmd_prompt (ctx_server .mctx ,prompt .get < std::string > (),files ,ctx_server .init_opt )); 4376 }else { 4377// Everything else, including multimodal completions. 4378inputs = tokenize_input_prompts (ctx_server .vocab ,ctx_server .mctx ,prompt , true, true,ctx_server .init_opt ); 4379 } 4380 4381// tasks.reserve(inputs.size()); // TODO: this is inaccurate due to child tasks 4382 4383// message delimiters for checkpointing 4384json delims = json_value (data ,"message_delimiters" , json::array ()); 4385auto delimiters = common_chat_msg_delimiters_parse (delims ); 4386delimiters .tokenize (ctx_server .vocab ); 4387 4388for (size_t i = 0 ;i < inputs .size ();i ++ ) { 4389server_task task = server_task (type ); 4390 4391task .id = rd .get_new_id (); 4392 4393task .tokens = std::move (inputs [i ]); 4394task .params = server_schema::eval_llama_cmpl_schema ( 4395ctx_server .vocab , 4396params , 4397meta -> logit_bias_eog , 4398data ); 4399 4400task .params .message_spans = task .tokens .find_message_spans (delimiters ); 4401 4402task .id_slot = json_value (data ,"id_slot" ,-1 ); 4403sse_ping_interval = task .params .sse_ping_interval ; 4404 4405// OAI-compat 4406task .params .res_type = res_type ; 4407task .params .oaicompat_cmpl_id = completion_id ; 4408task .params .oaicompat_model = meta -> model_name ; 4409 4410// prepare child tasks 4411if (task .params .n_cmpl > 1 ) { 4412int n_children = task .params .n_cmpl - 1 ; 4413for (int j = 0 ;j < n_children ;j ++ ) { 4414task .add_child (task .id ,rd .get_new_id ()); 4415 } 4416 } 4417 4418tasks .push_back (std::move (task )); 4419 } 4420 4421rd .post_tasks (std::move (tasks )); 4422 }catch (const std::exception & e ) { 4423res -> error (format_error_response (e .what (),ERROR_TYPE_INVALID_REQUEST )); 4424return res ; 4425 } 4426 4427bool stream = json_value (data ,"stream" , false); 4428 4429if (!stream ) { 4430// non-stream, wait for the results 4431auto all_results = rd .wait_for_all (req .should_stop ); 4432if (all_results .is_terminated ) { 4433return res ;// connection is closed 4434 }else if (all_results .error ) { 4435res -> error (all_results .error -> to_json ()); 4436return res ; 4437 }else { 4438json arr = json::array (); 4439for (auto & res :all_results .results ) { 4440GGML_ASSERT (dynamic_cast < server_task_result_cmpl_final *> (res .get ())!= nullptr ); 4441arr .push_back (res -> to_json ()); 4442 } 4443GGML_ASSERT (!arr .empty ()&& "empty results" ); 4444if (arr .size ()== 1 ) { 4445// if single request, return single object instead of array 4446res -> ok (arr [0 ]); 4447 }else if (res_type == TASK_RESPONSE_TYPE_OAI_CHAT || res_type == TASK_RESPONSE_TYPE_OAI_CMPL ) { 4448// if multiple results in OAI format, we need to re-format them 4449json & choices = arr [0 ]["choices" ]; 4450for (size_t i = 1 ;i < arr .size ();i ++ ) { 4451choices .push_back (std::move (arr [i ]["choices" ][0 ])); 4452 } 4453res -> ok (arr [0 ]); 4454 }else { 4455// multi-results, non-OAI compat 4456res -> ok (arr ); 4457 } 4458 } 4459 }else { 4460// in streaming mode, the first error must be treated as non-stream response 4461// this is to match the OAI API behavior 4462// ref: https://github.com/ggml-org/llama.cpp/pull/16486#discussion_r2419657309 4463auto first_result = rd .next (req .should_stop ); 4464if (first_result == nullptr ) { 4465GGML_ASSERT (req .should_stop ()); 4466return res ;// connection is closed 4467 } 4468 4469if (first_result -> is_error ()) { 4470res -> error (first_result -> to_json ()); 4471return res ; 4472 } 4473 4474GGML_ASSERT ( 4475dynamic_cast < server_task_result_cmpl_partial *> (first_result .get ())!= nullptr || 4476dynamic_cast < server_task_result_cmpl_final *> (first_result .get ())!= nullptr 4477 ); 4478 4479// next responses are streamed 4480// to be sent immediately 4481json first_result_json = first_result -> to_json (); 4482if (first_result_json == nullptr ) { 4483res -> data = "" ;// simply send HTTP headers and status code 4484 }else if (res_type == TASK_RESPONSE_TYPE_ANTHROPIC ) { 4485res -> data = format_anthropic_sse (first_result_json ); 4486 }else if (res_type == TASK_RESPONSE_TYPE_OAI_RESP ) { 4487res -> data = format_oai_resp_sse (first_result_json ); 4488 }else { 4489res -> data = format_oai_sse (first_result_json ); 4490 } 4491res -> status = 200 ; 4492res -> content_type = "text/event-stream" ; 4493res -> set_next ([res_this = res .get (),res_type ,sse_ping_interval ](std::string & output )-> bool { 4494static auto format_error = [](task_response_type res_type ,const json & res_json ) { 4495if (res_type == TASK_RESPONSE_TYPE_ANTHROPIC ) { 4496return format_anthropic_sse ({ 4497 {"event" ,"error" }, 4498 {"data" ,res_json }, 4499 }); 4500 }else { 4501return format_oai_sse (json {{"error" ,res_json }}); 4502 } 4503 }; 4504 4505auto effective_should_stop = [& res_this ]() { 4506return res_this -> should_stop (); 4507 }; 4508 4509try { 4510if (effective_should_stop ()) { 4511SRV_DBG ("%s" ,"stopping streaming due to should_stop condition\n" ); 4512return false;// should_stop condition met 4513 } 4514 4515if (!res_this -> data .empty ()) { 4516// flush the first chunk 4517output = std::move (res_this -> data ); 4518res_this -> data .clear (); 4519return true; 4520 } 4521 4522server_response_reader & rd = res_this -> rd ; 4523 4524// check if there is more data 4525if (!rd .has_next ()) { 4526switch (res_type ) { 4527case TASK_RESPONSE_TYPE_NONE : 4528case TASK_RESPONSE_TYPE_OAI_RESP : 4529case TASK_RESPONSE_TYPE_ANTHROPIC : 4530output = "" ; 4531break ; 4532 4533default : 4534output = "data: [DONE]\n\n" ; 4535break ; 4536 } 4537SRV_DBG ("%s" ,"all results received, terminating stream\n" ); 4538return false;// no more data, terminate 4539 } 4540 4541// receive subsequent results 4542bool timeout = false; 4543int64_t start_time = ggml_time_ms (); 4544auto result = rd .next ([& timeout ,& start_time ,sse_ping_interval ,& effective_should_stop ]() { 4545if (effective_should_stop ()) { 4546return true;// should_stop condition met 4547 }else if (sse_ping_interval > 0 && ggml_time_ms ()- start_time > (int64_t )sse_ping_interval * 1000 ) { 4548timeout = true; 4549return true;// timeout 4550 } 4551return false; 4552 }); 4553 4554if (timeout ) { 4555// some clients may time out (e.g. undici) will time out if no data is received for a while, so we need to send a ping to keep the connection alive 4556SRV_DBG ("%s" ,"sending SSE ping\n" ); 4557output = ":\n\n" ; 4558return true; 4559 } 4560 4561if (result == nullptr ) { 4562SRV_DBG ("%s" ,"stopping streaming due to should_stop condition\n" ); 4563GGML_ASSERT (effective_should_stop ()); 4564return false;// should_stop condition met 4565 } 4566 4567// send the results 4568if (result -> is_error ()) { 4569json res_json = result -> to_json (); 4570output = format_error (res_type ,res_json ); 4571SRV_DBG ("%s" ,"error received during streaming, terminating stream\n" ); 4572return false;// terminate on error 4573 }else { 4574GGML_ASSERT ( 4575dynamic_cast < server_task_result_cmpl_partial *> (result .get ())!= nullptr 4576|| dynamic_cast < server_task_result_cmpl_final *> (result .get ())!= nullptr 4577 ); 4578json res_json = result -> to_json (); 4579if (res_type == TASK_RESPONSE_TYPE_ANTHROPIC ) { 4580output = format_anthropic_sse (res_json ); 4581 }else if (res_type == TASK_RESPONSE_TYPE_OAI_RESP ) { 4582output = format_oai_resp_sse (res_json ); 4583 }else { 4584output = format_oai_sse (res_json ); 4585 } 4586 } 4587 4588// has next data, continue 4589return true; 4590 4591 }catch (const std::exception & e ) { 4592json error_json = format_error_response (e .what (),ERROR_TYPE_SERVER ); 4593output = format_error (res_type ,error_json ); 4594 4595// terminate on exception 4596return false; 4597 } 4598 }); 4599 } 4600 4601return res ; 4602} 4603 4604std::unique_ptr < server_res_generator > server_routes::create_response (bool bypass_sleep ) { 4605return std::make_unique < server_res_generator > (queue_tasks ,queue_results ,params .sleep_idle_seconds ,bypass_sleep ); 4606} 4607 4608server_routes::server_routes (const common_params & params ,server_context & ctx_server ) 4609 :params (params ), 4610ctx_server (* ctx_server .impl ), 4611queue_tasks (ctx_server .impl -> queue_tasks ), 4612queue_results (ctx_server .impl -> queue_results ) { 4613init_routes (); 4614 4615// note: this must be registered before load_model() 4616// so that on sleep phase, the callback is called before ctx is destroyed 4617queue_tasks .on_sleeping_state ([this ](bool is_sleeping ) { 4618update_cached_responses (is_sleeping ); 4619 }); 4620} 4621 4622static json get_res_model_info (const server_context_meta & meta ) { 4623// note: do NOT use ctx_server here, otherwise it's not possible to use this during sleep 4624 4625return { 4626 {"id" ,meta .model_name }, 4627 {"aliases" ,meta .model_aliases }, 4628 {"tags" ,meta .model_tags }, 4629 {"object" ,"model" }, 4630 {"created" , std::time (0 )}, 4631 {"owned_by" ,"llamacpp" }, 4632 {"meta" , { 4633 {"vocab_type" ,meta .model_vocab_type }, 4634 {"n_vocab" ,meta .model_vocab_n_tokens }, 4635 {"n_ctx" ,meta .slot_n_ctx }, 4636 {"n_ctx_train" ,meta .model_n_ctx_train }, 4637 {"n_embd" ,meta .model_n_embd_inp }, 4638 {"n_params" ,meta .model_n_params }, 4639 {"size" ,meta .model_size }, 4640 {"ftype" ,meta .model_ftype }, 4641 }}, 4642 }; 4643} 4644 4645static json get_res_models (const server_context_meta & meta ) { 4646// note: do NOT use ctx_server here, otherwise it's not possible to use this during sleep 4647 4648return json { 4649 {"models" , json::array ({ 4650 { 4651 {"name" ,meta .model_name }, 4652 {"model" ,meta .model_name }, 4653 {"modified_at" ,"" }, 4654 {"size" ,"" }, 4655 {"digest" ,"" },// dummy value, llama.cpp does not support managing model file's hash 4656 {"type" ,"model" }, 4657 {"description" ,"" }, 4658 {"tags" , json::array ({"" })}, 4659 {"capabilities" ,meta .has_mtmd ? json::array ({"completion" ,"multimodal" }) : json::array ({"completion" })}, 4660 {"parameters" ,"" }, 4661 {"details" , { 4662 {"parent_model" ,"" }, 4663 {"format" ,"gguf" }, 4664 {"family" ,"" }, 4665 {"families" , json::array ({"" })}, 4666 {"parameter_size" ,"" }, 4667 {"quantization_level" ,"" } 4668 }} 4669 } 4670 })}, 4671 {"object" ,"list" }, 4672 {"data" , json::array ({ 4673get_res_model_info (meta ), 4674 })} 4675 }; 4676} 4677 4678static json get_res_props (const server_context_meta & meta ,const common_params & params ,bool is_sleeping ) { 4679// note: do NOT use ctx_server here, otherwise it's not possible to use this during sleep 4680 4681task_params tparams ; 4682tparams .sampling = params .sampling ; 4683json default_generation_settings_for_props = json { 4684 {"params" ,tparams .to_json (true) }, 4685 {"n_ctx" ,meta .slot_n_ctx }, 4686 }; 4687 4688 std::string tmpl_default = common_chat_templates_source (meta .chat_params .tmpls .get (),"" ); 4689 std::string tmpl_tools = common_chat_templates_source (meta .chat_params .tmpls .get (),"tool_use" ); 4690 4691json props = { 4692 {"default_generation_settings" ,default_generation_settings_for_props }, 4693 {"total_slots" ,params .n_parallel }, 4694 {"model_alias" ,meta .model_name }, 4695 {"model_ftype" ,meta .model_ftype }, 4696 {"model_path" ,meta .model_path }, 4697 {"modalities" ,json { 4698 {"vision" ,meta .has_inp_image }, 4699 {"video" ,meta .has_inp_video }, 4700 {"audio" ,meta .has_inp_audio }, 4701 } }, 4702 {"media_marker" ,get_media_marker () }, 4703 {"endpoint_slots" ,params .endpoint_slots }, 4704 {"endpoint_props" ,params .endpoint_props }, 4705 {"endpoint_metrics" ,params .endpoint_metrics }, 4706 {"ui" ,params .ui }, 4707 {"ui_settings" ,meta .json_ui_settings }, 4708 {"chat_template" ,tmpl_default }, 4709 {"chat_template_caps" ,meta .chat_template_caps }, 4710 {"bos_token" ,meta .bos_token_str }, 4711 {"eos_token" ,meta .eos_token_str }, 4712 {"build_info" ,meta .build_info }, 4713 {"is_sleeping" ,is_sleeping }, 4714 {"cors_proxy_enabled" ,params .ui_mcp_proxy }, 4715 }; 4716if (params .use_jinja ) { 4717if (!tmpl_tools .empty ()) { 4718props ["chat_template_tool_use" ]= tmpl_tools ; 4719 } 4720 } 4721 4722return props ; 4723} 4724 4725json server_routes::get_model_info ()const { 4726return get_res_model_info (* meta ); 4727} 4728 4729void server_routes::init_routes () { 4730// IMPORTANT: all lambda functions must start with create_response() 4731// this is to ensure that the server_res_generator can handle sleeping case correctly 4732 4733this -> get_health = [this ](const server_http_req & ) { 4734// error and loading states are handled by middleware 4735auto res = create_response (true); 4736 4737// this endpoint can be accessed during sleeping 4738// the next LOC is to avoid someone accidentally use ctx_server 4739bool ctx_server ;// do NOT delete this line 4740GGML_UNUSED (ctx_server ); 4741 4742res -> ok ({{"status" ,"ok" }}); 4743return res ; 4744 }; 4745 4746this -> get_metrics = [this ](const server_http_req & req ) { 4747auto res = create_response (true); 4748if (!params .endpoint_metrics ) { 4749res -> error (format_error_response ("This server does not support metrics endpoint. Start it with `--metrics`" ,ERROR_TYPE_NOT_SUPPORTED )); 4750return res ; 4751 } 4752 4753// render response using cached_metrics 4754auto use_cached_metrics = [& ]() { 4755 std::unique_lock < std::mutex > lock (mutex_cache ); 4756res -> headers ["Process-Start-Time-Unix" ]= std::to_string (cached_metrics .t_start ); 4757server_task_result_metrics tmp ; 4758tmp .metrics = cached_metrics ; 4759res -> content_type = "text/plain; version=0.0.4" ; 4760res -> status = 200 ; 4761res -> data = tmp .to_metrics (); 4762// the gauges are averaged over the window between two scrapes 4763cached_metrics .reset_bucket (); 4764should_reset_buckets = true; 4765 }; 4766 4767if (queue_tasks .is_sleeping ()) { 4768use_cached_metrics (); 4769 4770 }else { 4771// request slots data using task queue 4772 { 4773server_task task (SERVER_TASK_TYPE_METRICS ); 4774task .id = res -> rd .get_new_id (); 4775// the gauges are averaged over the window between two scrapes 4776task .metrics_reset_bucket = true; 4777res -> rd .post_task (std::move (task ), true);// high-priority task 4778 } 4779 4780// a task posted right before sleeping is never processed, do not wait for it 4781auto result = res -> rd .next ([& ]{ 4782return req .should_stop ()|| queue_tasks .is_sleeping (); 4783 }); 4784if (!result ) { 4785if (!req .should_stop ()) { 4786use_cached_metrics (); 4787 } 4788return res ; 4789 } 4790 4791if (result -> is_error ()) { 4792res -> error (result -> to_json ()); 4793return res ; 4794 } 4795 4796auto res_task = dynamic_cast < server_task_result_metrics *> (result .get ()); 4797GGML_ASSERT (res_task != nullptr ); 4798 4799res -> headers ["Process-Start-Time-Unix" ]= std::to_string (res_task -> metrics .t_start ); 4800res -> content_type = "text/plain; version=0.0.4" ; 4801res -> status = 200 ; 4802res -> data = res_task -> to_metrics (); 4803 } 4804 4805return res ; 4806 }; 4807 4808this -> get_slots = [this ](const server_http_req & req ) { 4809auto res = create_response (); 4810if (!params .endpoint_slots ) { 4811res -> error (format_error_response ("This server does not support slots endpoint. Start it with `--slots`" ,ERROR_TYPE_NOT_SUPPORTED )); 4812return res ; 4813 } 4814 4815// request slots data using task queue 4816 { 4817server_task task (SERVER_TASK_TYPE_SLOT_GET ); 4818task .id = res -> rd .get_new_id (); 4819res -> rd .post_task (std::move (task ), true);// high-priority task 4820 } 4821 4822// get the result 4823auto result = res -> rd .next (req .should_stop ); 4824if (!result ) { 4825// connection was closed 4826GGML_ASSERT (req .should_stop ()); 4827return res ; 4828 } 4829 4830if (result -> is_error ()) { 4831res -> error (result -> to_json ()); 4832return res ; 4833 } 4834 4835auto * res_task = dynamic_cast < server_task_result_slots *> (result .get ()); 4836GGML_ASSERT (res_task != nullptr ); 4837 4838// optionally return "fail_on_no_slot" error 4839if (!req .get_param ("fail_on_no_slot" ).empty ()) { 4840if (res_task -> n_idle_slots == 0 ) { 4841res -> error (format_error_response ("no slot available" ,ERROR_TYPE_UNAVAILABLE )); 4842return res ; 4843 } 4844 } 4845 4846res -> ok (res_task -> to_json ()); 4847return res ; 4848 }; 4849 4850this -> post_slots = [this ](const server_http_req & req ) { 4851auto res = create_response (); 4852if (params .slot_save_path .empty ()) { 4853res -> error (format_error_response ("This server does not support slots action. Start it with `--slot-save-path`" ,ERROR_TYPE_NOT_SUPPORTED )); 4854return res ; 4855 } 4856 4857 std::string id_slot_str = req .get_param ("id_slot" ); 4858 4859int id_slot ; 4860try { 4861id_slot = std::stoi (id_slot_str ); 4862 }catch (const std::exception & ) { 4863res -> error (format_error_response ("Invalid slot ID" ,ERROR_TYPE_INVALID_REQUEST )); 4864return res ; 4865 } 4866 4867 std::string action = req .get_param ("action" ); 4868 4869if (action == "save" ) { 4870return handle_slots_save (req ,id_slot ); 4871 } 4872if (action == "restore" ) { 4873return handle_slots_restore (req ,id_slot ); 4874 } 4875if (action == "erase" ) { 4876return handle_slots_erase (req ,id_slot ); 4877 } 4878 4879res -> error (format_error_response ("Invalid action" ,ERROR_TYPE_INVALID_REQUEST )); 4880return res ; 4881 }; 4882 4883this -> get_props = [this ](const server_http_req & ) { 4884auto res = create_response (true); 4885// note: do NOT use ctx_server here, this endpoint must be accessible during sleep 4886if (queue_tasks .is_sleeping ()) { 4887 std::unique_lock < std::mutex > lock (mutex_cache ); 4888res -> ok (cached_props ); 4889 }else { 4890res -> ok (get_res_props (* meta ,params , false)); 4891 } 4892return res ; 4893 }; 4894 4895this -> post_props = [this ](const server_http_req & ) { 4896auto res = create_response (); 4897if (!params .endpoint_props ) { 4898res -> error (format_error_response ("This server does not support changing global properties. Start it with `--props`" ,ERROR_TYPE_NOT_SUPPORTED )); 4899return res ; 4900 } 4901// update any props here 4902 4903res -> ok ({{"success" , true }}); 4904return res ; 4905 }; 4906 4907this -> post_infill = [this ](const server_http_req & req ) { 4908auto res = create_response (); 4909// check model compatibility 4910 std::string err ; 4911if (llama_vocab_fim_pre (ctx_server .vocab )== LLAMA_TOKEN_NULL ) { 4912err += "prefix token is missing. " ; 4913 } 4914if (llama_vocab_fim_suf (ctx_server .vocab )== LLAMA_TOKEN_NULL ) { 4915err += "suffix token is missing. " ; 4916 } 4917if (llama_vocab_fim_mid (ctx_server .vocab )== LLAMA_TOKEN_NULL ) { 4918err += "middle token is missing. " ; 4919 } 4920if (!err .empty ()) { 4921res -> error (format_error_response (string_format ("Infill is not supported by this model: %s" ,err .c_str ()),ERROR_TYPE_NOT_SUPPORTED )); 4922return res ; 4923 } 4924 4925// validate input 4926json data = json::parse (req .body ); 4927if (data .contains ("prompt" )&& !data .at ("prompt" ).is_string ()) { 4928// prompt is optional 4929res -> error (format_error_response ("\"prompt\" must be a string" ,ERROR_TYPE_INVALID_REQUEST )); 4930 } 4931 4932if (!data .contains ("input_prefix" )) { 4933res -> error (format_error_response ("\"input_prefix\" is required" ,ERROR_TYPE_INVALID_REQUEST )); 4934 } 4935 4936if (!data .contains ("input_suffix" )) { 4937res -> error (format_error_response ("\"input_suffix\" is required" ,ERROR_TYPE_INVALID_REQUEST )); 4938 } 4939 4940if (data .contains ("input_extra" )&& !data .at ("input_extra" ).is_array ()) { 4941// input_extra is optional 4942res -> error (format_error_response ("\"input_extra\" must be an array of {\"filename\": string, \"text\": string}" ,ERROR_TYPE_INVALID_REQUEST )); 4943return res ; 4944 } 4945 4946json input_extra = json_value (data ,"input_extra" , json::array ()); 4947for (const auto & chunk :input_extra ) { 4948// { "text": string, "filename": string } 4949if (!chunk .contains ("text" )|| !chunk .at ("text" ).is_string ()) { 4950res -> error (format_error_response ("extra_context chunk must contain a \"text\" field with a string value" ,ERROR_TYPE_INVALID_REQUEST )); 4951return res ; 4952 } 4953// filename is optional 4954if (chunk .contains ("filename" )&& !chunk .at ("filename" ).is_string ()) { 4955res -> error (format_error_response ("extra_context chunk's \"filename\" field must be a string" ,ERROR_TYPE_INVALID_REQUEST )); 4956return res ; 4957 } 4958 } 4959data ["input_extra" ]= input_extra ;// default to empty array if it's not exist 4960 4961 std::string prompt = json_value (data ,"prompt" , std::string ()); 4962 std::vector < server_tokens > tokenized_prompts = tokenize_input_prompts (ctx_server .vocab ,ctx_server .mctx ,prompt , false, true,ctx_server .init_opt ); 4963SRV_DBG ("creating infill tasks, n_prompts = %d\n" , (int )tokenized_prompts .size ()); 4964data ["prompt" ]= format_prompt_infill ( 4965ctx_server .vocab , 4966data .at ("input_prefix" ), 4967data .at ("input_suffix" ), 4968data .at ("input_extra" ), 4969params .n_batch , 4970params .n_predict , 4971meta -> slot_n_ctx , 4972params .spm_infill , 4973tokenized_prompts [0 ].get_tokens ()// TODO: this could maybe be multimodal. 4974 ); 4975 4976 std::vector < raw_buffer > files ;// dummy 4977return handle_completions_impl ( 4978req , 4979SERVER_TASK_TYPE_INFILL , 4980data , 4981files , 4982TASK_RESPONSE_TYPE_NONE );// infill is not OAI compatible 4983 }; 4984 4985this -> post_completions = [this ](const server_http_req & req ) { 4986auto res = create_response (); 4987 std::vector < raw_buffer > files ;// dummy 4988const json body = json::parse (req .body ); 4989return handle_completions_impl ( 4990req , 4991SERVER_TASK_TYPE_COMPLETION , 4992body , 4993files , 4994TASK_RESPONSE_TYPE_NONE ); 4995 }; 4996 4997this -> post_completions_oai = [this ](const server_http_req & req ) { 4998auto res = create_response (); 4999 std::vector < raw_buffer > files ;// dummy 5000const json body = json::parse (req .body ); 5001return handle_completions_impl ( 5002req , 5003SERVER_TASK_TYPE_COMPLETION , 5004body , 5005files , 5006TASK_RESPONSE_TYPE_OAI_CMPL ); 5007 }; 5008 5009this -> post_chat_completions = [this ](const server_http_req & req ) { 5010auto res = create_response (); 5011 std::vector < raw_buffer > files ; 5012json body = json::parse (req .body ); 5013json body_parsed = oaicompat_chat_params_parse ( 5014body , 5015meta -> chat_params , 5016files ); 5017return handle_completions_impl ( 5018req , 5019SERVER_TASK_TYPE_COMPLETION , 5020body_parsed , 5021files , 5022TASK_RESPONSE_TYPE_OAI_CHAT ); 5023 }; 5024 5025this -> post_chat_completions_tok = [this ](const server_http_req & req ) { 5026return handle_count_tokens (ctx_server .vocab ,ctx_server .mctx ,ctx_server .init_opt ,req ,TASK_RESPONSE_TYPE_OAI_CHAT ); 5027 }; 5028 5029this -> post_control = [this ](const server_http_req & req ) { 5030auto res = create_response (); 5031const json body = json::parse (req .body ); 5032 5033const std::string cmpl_id = json_value (body ,"id" , std::string ()); 5034const std::string action = json_value (body ,"action" , std::string ()); 5035if (cmpl_id .empty ()) { 5036res -> error (format_error_response ("missing completion id" ,ERROR_TYPE_INVALID_REQUEST )); 5037return res ; 5038 } 5039if (action != "reasoning_end" ) { 5040res -> error (format_error_response ("unknown control action" ,ERROR_TYPE_INVALID_REQUEST )); 5041return res ; 5042 } 5043 5044auto & rd = res -> rd ; 5045 { 5046server_task task (SERVER_TASK_TYPE_CONTROL ); 5047task .id = rd .get_new_id (); 5048task .params .control_cmpl_id = cmpl_id ; 5049task .params .control_action = action ; 5050rd .post_task (std::move (task )); 5051 } 5052 5053auto result = rd .next (req .should_stop ); 5054if (!result ) { 5055GGML_ASSERT (req .should_stop ()); 5056return res ; 5057 } 5058if (result -> is_error ()) { 5059res -> error (result -> to_json ()); 5060return res ; 5061 } 5062res -> ok (result -> to_json ()); 5063return res ; 5064 }; 5065 5066this -> post_responses_oai = [this ](const server_http_req & req ) { 5067auto res = create_response (); 5068 std::vector < raw_buffer > files ; 5069json body = server_chat_convert_responses_to_chatcmpl (json::parse (req .body )); 5070SRV_DBG ("%s\n" ,"Request converted: OpenAI Responses -> OpenAI Chat Completions" ); 5071SRV_DBG ("converted request: %s\n" ,body .dump ().c_str ()); 5072json body_parsed = oaicompat_chat_params_parse ( 5073body , 5074meta -> chat_params , 5075files ); 5076return handle_completions_impl ( 5077req , 5078SERVER_TASK_TYPE_COMPLETION , 5079body_parsed , 5080files , 5081TASK_RESPONSE_TYPE_OAI_RESP ); 5082 }; 5083 5084this -> post_responses_tok_oai = [this ](const server_http_req & req ) { 5085return handle_count_tokens (ctx_server .vocab ,ctx_server .mctx ,ctx_server .init_opt ,req ,TASK_RESPONSE_TYPE_OAI_RESP ); 5086 }; 5087 5088this -> post_transcriptions_oai = [this ](const server_http_req & req ) { 5089auto res = create_response (); 5090 5091if (!meta -> has_mtmd || !meta -> chat_params .allow_audio ) { 5092res -> error (format_error_response ("The current model does not support audio input." ,ERROR_TYPE_NOT_SUPPORTED )); 5093return res ; 5094 } 5095 5096 std::vector < raw_buffer > files ; 5097json body = convert_transcriptions_to_chatcmpl ( 5098 json::parse (req .body ), 5099meta -> chat_params .tmpls .get (), 5100req .files , 5101files ); 5102SRV_DBG ("%s\n" ,"Request converted: OpenAI Transcriptions -> OpenAI Chat Completions" ); 5103SRV_DBG ("converted request: %s\n" ,body .dump ().c_str ()); 5104json body_parsed = oaicompat_chat_params_parse ( 5105body , 5106meta -> chat_params , 5107files ); 5108return handle_completions_impl ( 5109req , 5110SERVER_TASK_TYPE_COMPLETION , 5111body_parsed , 5112files , 5113TASK_RESPONSE_TYPE_OAI_ASR ); 5114 }; 5115 5116this -> post_anthropic_messages = [this ](const server_http_req & req ) { 5117auto res = create_response (); 5118 std::vector < raw_buffer > files ; 5119json body = server_chat_convert_anthropic_to_oai (json::parse (req .body )); 5120SRV_DBG ("%s\n" ,"Request converted: Anthropic -> OpenAI Chat Completions" ); 5121SRV_DBG ("converted request: %s\n" ,body .dump ().c_str ()); 5122json body_parsed = oaicompat_chat_params_parse ( 5123body , 5124meta -> chat_params , 5125files ); 5126return handle_completions_impl ( 5127req , 5128SERVER_TASK_TYPE_COMPLETION , 5129body_parsed , 5130files , 5131TASK_RESPONSE_TYPE_ANTHROPIC ); 5132 }; 5133 5134this -> post_anthropic_count_tokens = [this ](const server_http_req & req ) { 5135return handle_count_tokens (ctx_server .vocab ,ctx_server .mctx ,ctx_server .init_opt ,req ,TASK_RESPONSE_TYPE_ANTHROPIC ); 5136 }; 5137 5138// same with handle_chat_completions, but without inference part 5139this -> post_apply_template = [this ](const server_http_req & req ) { 5140auto res = create_response (); 5141 std::vector < raw_buffer > files ;// dummy, unused 5142json body = json::parse (req .body ); 5143json data = oaicompat_chat_params_parse ( 5144body , 5145meta -> chat_params , 5146files ); 5147res -> ok ({{"prompt" , std::move (data .at ("prompt" )) }}); 5148return res ; 5149 }; 5150 5151this -> get_models = [this ](const server_http_req & ) { 5152auto res = create_response (true); 5153// note: do NOT use ctx_server here, this endpoint must be accessible during sleep 5154if (queue_tasks .is_sleeping ()) { 5155 std::unique_lock < std::mutex > lock (mutex_cache ); 5156res -> ok (cached_models ); 5157 }else { 5158res -> ok (get_res_models (* meta )); 5159 } 5160return res ; 5161 }; 5162 5163this -> post_tokenize = [this ](const server_http_req & req ) { 5164auto res = create_response (); 5165const json body = json::parse (req .body ); 5166json tokens_response = json::array (); 5167if (body .count ("content" )!= 0 ) { 5168const bool add_special = json_value (body ,"add_special" , false); 5169const bool parse_special = json_value (body ,"parse_special" , true); 5170const bool with_pieces = json_value (body ,"with_pieces" , false); 5171 5172llama_tokens tokens = tokenize_mixed (ctx_server .vocab ,body .at ("content" ),add_special ,parse_special ); 5173 5174if (with_pieces ) { 5175for (const auto & token :tokens ) { 5176 std::string piece = common_token_to_piece (ctx_server .vocab ,token ); 5177json piece_json ; 5178 5179// Check if the piece is valid UTF-8 5180if (is_valid_utf8 (piece )) { 5181piece_json = piece ; 5182 }else { 5183// If not valid UTF-8, store as array of byte values 5184piece_json = json::array (); 5185for (unsigned char c :piece ) { 5186piece_json .push_back (static_cast < int > (c )); 5187 } 5188 } 5189 5190tokens_response .push_back ({ 5191 {"id" ,token }, 5192 {"piece" ,piece_json } 5193 }); 5194 } 5195 }else { 5196tokens_response = tokens ; 5197 } 5198 } 5199 5200res -> ok (json {{"tokens" , std::move (tokens_response )}}); 5201return res ; 5202 }; 5203 5204this -> post_detokenize = [this ](const server_http_req & req ) { 5205auto res = create_response (); 5206const json body = json::parse (req .body ); 5207 5208 std::string content ; 5209if (body .count ("tokens" )!= 0 ) { 5210const llama_tokens tokens = body .at ("tokens" ).get < llama_tokens > (); 5211content = tokens_to_str (ctx_server .vocab ,tokens ); 5212 } 5213 5214res -> ok (json {{"content" , std::move (content )}}); 5215return res ; 5216 }; 5217 5218this -> post_embeddings = [this ](const server_http_req & req ) { 5219return handle_embeddings_impl (req ,TASK_RESPONSE_TYPE_NONE ); 5220 }; 5221 5222this -> post_embeddings_oai = [this ](const server_http_req & req ) { 5223return handle_embeddings_impl (req ,TASK_RESPONSE_TYPE_OAI_EMBD ); 5224 }; 5225 5226this -> post_rerank = [this ](const server_http_req & req ) { 5227auto res = create_response (); 5228if (!params .embedding || params .pooling_type != LLAMA_POOLING_TYPE_RANK ) { 5229res -> error (format_error_response ("This server does not support reranking. Start it with `--reranking`" ,ERROR_TYPE_NOT_SUPPORTED )); 5230return res ; 5231 } 5232 5233const json body = json::parse (req .body ); 5234 5235// if true, use TEI API format, otherwise use Jina API format 5236// Jina: https://jina.ai/reranker/ 5237// TEI: https://huggingface.github.io/text-embeddings-inference/#/Text%20Embeddings%20Inference/rerank 5238bool is_tei_format = body .contains ("texts" ); 5239 5240json query ; 5241if (body .count ("query" )== 1 ) { 5242query = body .at ("query" ); 5243if (!query .is_string ()) { 5244res -> error (format_error_response ("\"query\" must be a string" ,ERROR_TYPE_INVALID_REQUEST )); 5245return res ; 5246 } 5247 }else { 5248res -> error (format_error_response ("\"query\" must be provided" ,ERROR_TYPE_INVALID_REQUEST )); 5249return res ; 5250 } 5251 5252 std::vector < std::string > documents = json_value (body ,"documents" , 5253json_value (body ,"texts" , std::vector < std::string > ())); 5254if (documents .empty ()) { 5255res -> error (format_error_response ("\"documents\" must be a non-empty string array" ,ERROR_TYPE_INVALID_REQUEST )); 5256return res ; 5257 } 5258 5259int top_n = json_value (body ,"top_n" , (int )documents .size ()); 5260 5261// create and queue the task 5262json responses = json::array (); 5263auto & rd = res -> rd ; 5264 { 5265 std::vector < server_task > tasks ; 5266tasks .reserve (documents .size ()); 5267for (size_t i = 0 ;i < documents .size ();i ++ ) { 5268auto tmp = format_prompt_rerank (ctx_server .model_tgt ,ctx_server .vocab ,ctx_server .mctx ,query ,documents [i ],ctx_server .init_opt ); 5269server_task task = server_task (SERVER_TASK_TYPE_RERANK ); 5270task .id = rd .get_new_id (); 5271task .tokens = std::move (tmp ); 5272tasks .push_back (std::move (task )); 5273 } 5274rd .post_tasks (std::move (tasks )); 5275 } 5276 5277// wait for the results 5278auto all_results = rd .wait_for_all (req .should_stop ); 5279 5280// collect results 5281if (all_results .is_terminated ) { 5282return res ;// connection is closed 5283 }else if (all_results .error ) { 5284res -> error (all_results .error -> to_json ()); 5285return res ; 5286 }else { 5287for (auto & res :all_results .results ) { 5288GGML_ASSERT (dynamic_cast < server_task_result_rerank *> (res .get ())!= nullptr ); 5289responses .push_back (res -> to_json ()); 5290 } 5291 } 5292 5293// write JSON response 5294json root = format_response_rerank ( 5295body , 5296meta -> model_name , 5297responses , 5298is_tei_format , 5299documents , 5300top_n ); 5301 5302res -> ok (root ); 5303return res ; 5304 }; 5305 5306this -> get_lora_adapters = [this ](const server_http_req & req ) { 5307auto res = create_response (); 5308 5309auto & rd = res -> rd ; 5310 { 5311server_task task (SERVER_TASK_TYPE_GET_LORA ); 5312task .id = rd .get_new_id (); 5313rd .post_task (std::move (task )); 5314 } 5315 5316// get the result 5317auto result = rd .next (req .should_stop ); 5318if (!result ) { 5319// connection was closed 5320GGML_ASSERT (req .should_stop ()); 5321return res ; 5322 } 5323 5324if (result -> is_error ()) { 5325res -> error (result -> to_json ()); 5326return res ; 5327 } 5328 5329GGML_ASSERT (dynamic_cast < server_task_result_get_lora *> (result .get ())!= nullptr ); 5330res -> ok (result -> to_json ()); 5331return res ; 5332 }; 5333 5334this -> post_lora_adapters = [this ](const server_http_req & req ) { 5335auto res = create_response (); 5336const json body = json::parse (req .body ); 5337if (!body .is_array ()) { 5338res -> error (format_error_response ("Request body must be an array" ,ERROR_TYPE_INVALID_REQUEST )); 5339return res ; 5340 } 5341 5342auto & rd = res -> rd ; 5343 { 5344server_task task (SERVER_TASK_TYPE_SET_LORA ); 5345task .id = rd .get_new_id (); 5346task .set_lora = parse_lora_request (body ); 5347rd .post_task (std::move (task )); 5348 } 5349 5350// get the result 5351auto result = rd .next (req .should_stop ); 5352if (!result ) { 5353// connection was closed 5354GGML_ASSERT (req .should_stop ()); 5355return res ; 5356 } 5357 5358if (result -> is_error ()) { 5359res -> error (result -> to_json ()); 5360return res ; 5361 } 5362 5363GGML_ASSERT (dynamic_cast < server_task_result_apply_lora *> (result .get ())!= nullptr ); 5364res -> ok (result -> to_json ()); 5365return res ; 5366 }; 5367} 5368 5369std::unique_ptr < server_res_generator > server_routes::handle_slots_save (const server_http_req & req ,int id_slot ) { 5370auto res = create_response (); 5371const json request_data = json::parse (req .body ); 5372 std::string filename = request_data .at ("filename" ); 5373if (!fs_validate_filename (filename )) { 5374res -> error (format_error_response ("Invalid filename" ,ERROR_TYPE_INVALID_REQUEST )); 5375return res ; 5376 } 5377 std::string filepath = params .slot_save_path + filename ; 5378 5379auto & rd = res -> rd ; 5380 { 5381server_task task (SERVER_TASK_TYPE_SLOT_SAVE ); 5382task .id = rd .get_new_id (); 5383task .slot_action .id_slot = id_slot ; 5384task .slot_action .filename = filename ; 5385task .slot_action .filepath = filepath ; 5386rd .post_task (std::move (task )); 5387 } 5388 5389auto result = rd .next (req .should_stop ); 5390if (!result ) { 5391// connection was closed 5392GGML_ASSERT (req .should_stop ()); 5393return res ; 5394 } 5395 5396if (result -> is_error ()) { 5397res -> error (result -> to_json ()); 5398return res ; 5399 } 5400 5401res -> ok (result -> to_json ()); 5402return res ; 5403} 5404 5405std::unique_ptr < server_res_generator > server_routes::handle_slots_restore (const server_http_req & req ,int id_slot ) { 5406auto res = create_response (); 5407const json request_data = json::parse (req .body ); 5408 std::string filename = request_data .at ("filename" ); 5409if (!fs_validate_filename (filename )) { 5410res -> error (format_error_response ("Invalid filename" ,ERROR_TYPE_INVALID_REQUEST )); 5411return res ; 5412 } 5413 std::string filepath = params .slot_save_path + filename ; 5414 5415auto & rd = res -> rd ; 5416 { 5417server_task task (SERVER_TASK_TYPE_SLOT_RESTORE ); 5418task .id = rd .get_new_id (); 5419task .slot_action .id_slot = id_slot ; 5420task .slot_action .filename = filename ; 5421task .slot_action .filepath = filepath ; 5422rd .post_task (std::move (task )); 5423 } 5424 5425auto result = rd .next (req .should_stop ); 5426if (!result ) { 5427// connection was closed 5428GGML_ASSERT (req .should_stop ()); 5429return res ; 5430 } 5431 5432if (result -> is_error ()) { 5433res -> error (result -> to_json ()); 5434return res ; 5435 } 5436 5437GGML_ASSERT (dynamic_cast < server_task_result_slot_save_load *> (result .get ())!= nullptr ); 5438res -> ok (result -> to_json ()); 5439return res ; 5440} 5441 5442std::unique_ptr < server_res_generator > server_routes::handle_slots_erase (const server_http_req & req ,int id_slot ) { 5443auto res = create_response (); 5444auto & rd = res -> rd ; 5445 { 5446server_task task (SERVER_TASK_TYPE_SLOT_ERASE ); 5447task .id = rd .get_new_id (); 5448task .slot_action .id_slot = id_slot ; 5449rd .post_task (std::move (task )); 5450 } 5451 5452auto result = rd .next (req .should_stop ); 5453if (!result ) { 5454// connection was closed 5455GGML_ASSERT (req .should_stop ()); 5456return res ; 5457 } 5458 5459if (result -> is_error ()) { 5460res -> error (result -> to_json ()); 5461return res ; 5462 } 5463 5464GGML_ASSERT (dynamic_cast < server_task_result_slot_erase *> (result .get ())!= nullptr ); 5465res -> ok (result -> to_json ()); 5466return res ; 5467} 5468 5469std::unique_ptr < server_res_generator > server_routes::handle_embeddings_impl (const server_http_req & req ,task_response_type res_type ) { 5470auto res = create_response (); 5471if (!params .embedding ) { 5472res -> error (format_error_response ("This server does not support embeddings. Start it with `--embeddings`" ,ERROR_TYPE_NOT_SUPPORTED )); 5473return res ; 5474 } 5475 5476if (res_type != TASK_RESPONSE_TYPE_NONE && meta -> pooling_type == LLAMA_POOLING_TYPE_NONE ) { 5477res -> error (format_error_response ("Pooling type 'none' is not OAI compatible. Please use a different pooling type" ,ERROR_TYPE_INVALID_REQUEST )); 5478return res ; 5479 } 5480 5481const json body = json::parse (req .body ); 5482 5483// for the shape of input/content, see tokenize_input_prompts() 5484json prompt ; 5485if (body .count ("input" )!= 0 ) { 5486prompt = body .at ("input" ); 5487 }else if (body .contains ("content" )) { 5488res_type = TASK_RESPONSE_TYPE_NONE ;// "content" field is not OAI compatible 5489prompt = body .at ("content" ); 5490 }else { 5491res -> error (format_error_response ("\"input\" or \"content\" must be provided" ,ERROR_TYPE_INVALID_REQUEST )); 5492return res ; 5493 } 5494 5495bool use_base64 = false; 5496if (body .count ("encoding_format" )!= 0 ) { 5497const std::string & format = body .at ("encoding_format" ); 5498if (format == "base64" ) { 5499use_base64 = true; 5500 }else if (format != "float" ) { 5501res -> error (format_error_response ("The format to return the embeddings in. Can be either float or base64" ,ERROR_TYPE_INVALID_REQUEST )); 5502return res ; 5503 } 5504 } 5505 5506auto tokenized_prompts = tokenize_input_prompts (ctx_server .vocab ,ctx_server .mctx ,prompt , true, true,ctx_server .init_opt ); 5507for (const auto & tokens :tokenized_prompts ) { 5508// this check is necessary for models that do not add BOS token to the input 5509if (tokens .empty ()) { 5510res -> error (format_error_response ("Input content cannot be empty" ,ERROR_TYPE_INVALID_REQUEST )); 5511return res ; 5512 } 5513 } 5514 5515int embd_normalize = params .embd_normalize ; 5516if (body .count ("embd_normalize" )!= 0 ) { 5517embd_normalize = body .at ("embd_normalize" ).get < int > (); 5518if (meta -> pooling_type == LLAMA_POOLING_TYPE_NONE ) { 5519SRV_DBG ("embd_normalize is not supported by pooling type %d, ignoring it\n" ,meta -> pooling_type ); 5520 } 5521 } 5522 5523// create and queue the task 5524json responses = json::array (); 5525auto & rd = res -> rd ; 5526 { 5527 std::vector < server_task > tasks ; 5528for (size_t i = 0 ;i < tokenized_prompts .size ();i ++ ) { 5529server_task task = server_task (SERVER_TASK_TYPE_EMBEDDING ); 5530 5531task .id = rd .get_new_id (); 5532task .tokens = std::move (tokenized_prompts [i ]); 5533 5534// OAI-compat 5535task .params .res_type = res_type ; 5536task .params .embd_normalize = embd_normalize ; 5537 5538tasks .push_back (std::move (task )); 5539 } 5540rd .post_tasks (std::move (tasks )); 5541 } 5542 5543// wait for the results 5544auto all_results = rd .wait_for_all (req .should_stop ); 5545 5546// collect results 5547if (all_results .is_terminated ) { 5548return res ;// connection is closed 5549 }else if (all_results .error ) { 5550res -> error (all_results .error -> to_json ()); 5551return res ; 5552 }else { 5553for (auto & res :all_results .results ) { 5554GGML_ASSERT (dynamic_cast < server_task_result_embd *> (res .get ())!= nullptr ); 5555responses .push_back (res -> to_json ()); 5556 } 5557 } 5558 5559// write JSON response 5560json root = res_type == TASK_RESPONSE_TYPE_OAI_EMBD 5561 ?format_embeddings_response_oaicompat (body ,meta -> model_name ,responses ,use_base64 ) 5562 :json (responses ); 5563res -> ok (root ); 5564return res ; 5565} 5566 5567std::unique_ptr < server_res_generator > server_routes::handle_count_tokens (const llama_vocab * vocab ,mtmd_context * mctx ,const mtmd_helper_init_opt & init_opt ,const server_http_req & req ,task_response_type res_type ) { 5568auto res = create_response (); 5569 std::vector < raw_buffer > files ; 5570json body = json::parse (req .body ); 5571bool is_oai = false; 5572 5573switch (res_type ) { 5574case TASK_RESPONSE_TYPE_OAI_CHAT : 5575 { 5576is_oai = true; 5577 }break ; 5578case TASK_RESPONSE_TYPE_OAI_RESP : 5579 { 5580is_oai = true; 5581body = server_chat_convert_responses_to_chatcmpl (body ); 5582 }break ; 5583case TASK_RESPONSE_TYPE_ANTHROPIC : 5584 { 5585body = server_chat_convert_anthropic_to_oai (body ); 5586 }break ; 5587default : 5588res -> error (format_error_response ("invalid res_type" ,ERROR_TYPE_INVALID_REQUEST )); 5589return res ; 5590 } 5591 5592json body_parsed = oaicompat_chat_params_parse ( 5593body , 5594meta -> chat_params , 5595files ); 5596json prompt = body_parsed .at ("prompt" ); 5597// SRV_DBG("prompt = %s\n", prompt.dump().c_str()); 5598 5599// TODO @ngxson : refactor this code block, move this to server-common and reuse it in other places 5600size_t n_tokens ; 5601if (mctx != nullptr ) { 5602if (!prompt .is_string ()) { 5603throw std::runtime_error ("for mtmd, input prompt must be a string." ); 5604 } 5605n_tokens = process_mtmd_prompt (mctx ,prompt .get < std::string > (),files ,init_opt , true).size (); 5606 }else { 5607n_tokens = tokenize_mixed (vocab ,prompt , true, true).size (); 5608 } 5609 5610json response = {{"input_tokens" ,static_cast < int64_t > (n_tokens )}}; 5611if (is_oai ) { 5612response ["object" ]= "response.input_tokens" ; 5613 } 5614res -> ok (response ); 5615return res ; 5616} 5617 5618void server_routes::update_cached_responses (bool is_sleeping ) { 5619// caller is task_queue, so ctx_server can be accessed without holding locks 5620 std::unique_lock < std::mutex > lock (mutex_cache ); 5621 5622if (is_sleeping ) { 5623cached_models = get_res_models (* meta ); 5624cached_props = get_res_props (* meta ,params , true); 5625cached_metrics = ctx_server .get_metrics (); 5626 5627should_reset_buckets = false; 5628 5629SRV_DBG ("%s\n" ,"cached responses updated" ); 5630 5631 }else if (should_reset_buckets ) { 5632// a scrape during sleep already reported these buckets 5633ctx_server .reset_metrics_bucket (); 5634 5635should_reset_buckets = false; 5636 } 5637}