mirror of
https://github.com/ggml-org/llama.cpp.git
synced 2026-07-13 16:05:54 +02:00
Compare commits
12 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 99f3dc3229 | |||
| 34558825a2 | |||
| 8014d2cf97 | |||
| 4114ba18b2 | |||
| 0c4fa7a989 | |||
| 6b4dc2116a | |||
| e3546c7948 | |||
| d72bfa38f7 | |||
| 3cec3bcd16 | |||
| 13f2b28b09 | |||
| c92e806d1c | |||
| ea1f7bbb5d |
@@ -1081,6 +1081,9 @@ enum ggml_opt_optimizer_type common_opt_get_optimizer(const char *);
|
||||
struct common_prompt_checkpoint {
|
||||
int64_t n_tokens;
|
||||
|
||||
// (optional) id of the task that created the checkpoint
|
||||
int id_task = -1;
|
||||
|
||||
llama_pos pos_min;
|
||||
llama_pos pos_max;
|
||||
|
||||
|
||||
@@ -4493,7 +4493,14 @@ static bool ggml_backend_cuda_get_available_uma_memory(long * available_memory_k
|
||||
static void ggml_backend_cuda_device_get_memory(ggml_backend_dev_t dev, size_t * free, size_t * total) {
|
||||
ggml_backend_cuda_device_context * ctx = (ggml_backend_cuda_device_context *)dev->context;
|
||||
ggml_cuda_set_device(ctx->device);
|
||||
CUDA_CHECK(cudaMemGetInfo(free, total));
|
||||
cudaError_t err = cudaMemGetInfo(free, total);
|
||||
if (err != cudaSuccess) {
|
||||
(void)cudaGetLastError();
|
||||
GGML_LOG_WARN("%s: cudaMemGetInfo failed (%s), returning 0/0\n", __func__, cudaGetErrorString(err));
|
||||
*free = 0;
|
||||
*total = 0;
|
||||
return;
|
||||
}
|
||||
|
||||
// ref: https://github.com/ggml-org/llama.cpp/pull/17368
|
||||
#if defined(__linux__)
|
||||
|
||||
@@ -557,6 +557,10 @@ static struct gguf_context * gguf_init_from_reader(const struct gguf_reader & gr
|
||||
GGML_LOG_ERROR("%s: encountered bad_alloc error while reading key %" PRIi64 "\n", __func__, i);
|
||||
ok = false;
|
||||
}
|
||||
if (ok && key.empty()) {
|
||||
GGML_LOG_ERROR("%s: key %" PRIi64 " is empty\n", __func__, i);
|
||||
ok = false;
|
||||
}
|
||||
for (size_t j = 0; ok && j < ctx->kv.size(); ++j) {
|
||||
if (key == ctx->kv[j].key) {
|
||||
GGML_LOG_ERROR("%s: duplicate key '%s' for tensors %zu and %" PRIi64 " \n", __func__, key.c_str(), j, i);
|
||||
|
||||
@@ -5,7 +5,7 @@ import os
|
||||
import sys
|
||||
import subprocess
|
||||
|
||||
HTTPLIB_VERSION = "refs/tags/v0.49.0"
|
||||
HTTPLIB_VERSION = "refs/tags/v0.50.1"
|
||||
|
||||
vendor = {
|
||||
"https://github.com/nlohmann/json/releases/latest/download/json.hpp": "vendor/nlohmann/json.hpp",
|
||||
|
||||
+59
-15
@@ -29,6 +29,15 @@ static uint32_t dsv4_comp_size(uint32_t kv_size, uint32_t ratio) {
|
||||
return std::max<uint32_t>(1, (kv_size + ratio - 1)/ratio);
|
||||
}
|
||||
|
||||
static void dsv4_clear_tensor_stream(ggml_tensor * tensor, uint32_t stream) {
|
||||
GGML_ASSERT(ggml_is_contiguous(tensor));
|
||||
GGML_ASSERT(tensor->ne[3] == 1);
|
||||
GGML_ASSERT(stream < (uint32_t) tensor->ne[2]);
|
||||
|
||||
const size_t stream_size = tensor->nb[2];
|
||||
ggml_backend_tensor_memset(tensor, 0, stream*stream_size, stream_size);
|
||||
}
|
||||
|
||||
static int64_t dsv4_stream_offset(uint32_t n_stream, llama_seq_id seq_id, uint32_t size) {
|
||||
if (n_stream <= 1) {
|
||||
return 0;
|
||||
@@ -781,11 +790,20 @@ llama_dsv4_comp_state::llama_dsv4_comp_state(
|
||||
__func__, name, ratio, state_size, n_embd_state, n_stream, layers.size(), total_size()/1024.0/1024.0);
|
||||
}
|
||||
|
||||
void llama_dsv4_comp_state::clear(bool data) {
|
||||
void llama_dsv4_comp_state::clear(llama_seq_id seq_id, bool data) {
|
||||
if (!data) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (seq_id >= 0) {
|
||||
GGML_ASSERT((uint32_t) seq_id < n_stream);
|
||||
for (const auto & layer : layers) {
|
||||
dsv4_clear_tensor_stream(layer.kv, (uint32_t) seq_id);
|
||||
dsv4_clear_tensor_stream(layer.score, (uint32_t) seq_id);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
for (auto & [_, buf] : ctxs_bufs) {
|
||||
ggml_backend_buffer_clear(buf.get(), 0);
|
||||
}
|
||||
@@ -1034,7 +1052,7 @@ llama_kv_cache_dsv4::llama_kv_cache_dsv4(
|
||||
// graph does not necessarily overwrite; uninitialized buffer contents would
|
||||
// otherwise leak in (instance-specific garbage) and corrupt recall. Zero all
|
||||
// compressed buffers up front so reads of un-written rows are deterministic.
|
||||
clear_compressed(true);
|
||||
clear_compressed(-1, true);
|
||||
}
|
||||
|
||||
llama_memory_context_ptr llama_kv_cache_dsv4::init_batch(
|
||||
@@ -1147,7 +1165,7 @@ bool llama_kv_cache_dsv4::get_can_shift() const {
|
||||
|
||||
void llama_kv_cache_dsv4::clear(bool data) {
|
||||
kv_raw->clear(data);
|
||||
clear_compressed(true); // DSV4 compressed buffers must never expose stale/uninit rows
|
||||
clear_compressed(-1, true); // DSV4 compressed buffers must never expose stale/uninit rows
|
||||
}
|
||||
|
||||
bool llama_kv_cache_dsv4::seq_rm(llama_seq_id seq_id, llama_pos p0, llama_pos p1) {
|
||||
@@ -1169,7 +1187,7 @@ bool llama_kv_cache_dsv4::seq_rm(llama_seq_id seq_id, llama_pos p0, llama_pos p1
|
||||
const bool res = kv_raw->seq_rm(seq_id, p0, p1);
|
||||
|
||||
if (res) {
|
||||
clear_compressed(true);
|
||||
clear_compressed(seq_id, true);
|
||||
}
|
||||
|
||||
return res;
|
||||
@@ -1177,22 +1195,29 @@ bool llama_kv_cache_dsv4::seq_rm(llama_seq_id seq_id, llama_pos p0, llama_pos p1
|
||||
|
||||
void llama_kv_cache_dsv4::seq_cp(llama_seq_id seq_id_src, llama_seq_id seq_id_dst, llama_pos p0, llama_pos p1) {
|
||||
kv_raw->seq_cp(seq_id_src, seq_id_dst, p0, p1);
|
||||
clear_compressed(true);
|
||||
}
|
||||
|
||||
void llama_kv_cache_dsv4::seq_keep(llama_seq_id seq_id) {
|
||||
GGML_ASSERT(seq_id >= 0 && (uint32_t) seq_id < n_seq_max);
|
||||
|
||||
kv_raw->seq_keep(seq_id);
|
||||
clear_compressed(true);
|
||||
|
||||
for (llama_seq_id id = 0; id < (llama_seq_id) n_seq_max; ++id) {
|
||||
if (id == seq_id) {
|
||||
continue;
|
||||
}
|
||||
|
||||
kv_raw->seq_rm(id, -1, -1);
|
||||
clear_compressed(id, true);
|
||||
}
|
||||
}
|
||||
|
||||
void llama_kv_cache_dsv4::seq_add(llama_seq_id seq_id, llama_pos p0, llama_pos p1, llama_pos shift) {
|
||||
kv_raw->seq_add(seq_id, p0, p1, shift);
|
||||
clear_compressed(true);
|
||||
}
|
||||
|
||||
void llama_kv_cache_dsv4::seq_div(llama_seq_id seq_id, llama_pos p0, llama_pos p1, int d) {
|
||||
kv_raw->seq_div(seq_id, p0, p1, d);
|
||||
clear_compressed(true);
|
||||
}
|
||||
|
||||
llama_pos llama_kv_cache_dsv4::seq_pos_min(llama_seq_id seq_id) const {
|
||||
@@ -1328,13 +1353,32 @@ llama_dsv4_comp_state * llama_kv_cache_dsv4::get_lid_state() const {
|
||||
return lid_state.get();
|
||||
}
|
||||
|
||||
void llama_kv_cache_dsv4::clear_compressed(bool data) {
|
||||
kv_csa->clear(data);
|
||||
kv_hca->clear(data);
|
||||
kv_lid->clear(data);
|
||||
csa_state->clear(data);
|
||||
hca_state->clear(data);
|
||||
lid_state->clear(data);
|
||||
void llama_kv_cache_dsv4::clear_compressed(llama_seq_id seq_id, bool data) {
|
||||
if (seq_id < 0) {
|
||||
kv_csa->clear(data);
|
||||
kv_hca->clear(data);
|
||||
kv_lid->clear(data);
|
||||
} else {
|
||||
GGML_ASSERT((uint32_t) seq_id < n_seq_max);
|
||||
|
||||
const auto clear_seq = [seq_id, data](llama_kv_cache * kv) {
|
||||
kv->seq_rm(seq_id, -1, -1);
|
||||
|
||||
if (data) {
|
||||
for (uint32_t il : kv->get_layer_ids()) {
|
||||
dsv4_clear_tensor_stream(kv->get_k_storage(il), (uint32_t) seq_id);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
clear_seq(kv_csa.get());
|
||||
clear_seq(kv_hca.get());
|
||||
clear_seq(kv_lid.get());
|
||||
}
|
||||
|
||||
csa_state->clear(seq_id, data);
|
||||
hca_state->clear(seq_id, data);
|
||||
lid_state->clear(seq_id, data);
|
||||
}
|
||||
|
||||
//
|
||||
|
||||
@@ -21,7 +21,7 @@ public:
|
||||
const char * name,
|
||||
const llama_memory_i::layer_filter_cb & filter);
|
||||
|
||||
void clear(bool data);
|
||||
void clear(llama_seq_id seq_id, bool data);
|
||||
|
||||
uint32_t get_ratio() const;
|
||||
uint32_t get_state_size() const;
|
||||
@@ -67,6 +67,8 @@ private:
|
||||
// DSV4 uses a normal raw/SWA token cache plus compressed K-only block caches.
|
||||
// The compressed caches are storage only; DSV4-specific visibility and block
|
||||
// planning are handled by llama_kv_cache_dsv4_context / llm_graph_input_dsv4.
|
||||
// FIXME: currently the cache only supports non-unified mode even if unified flag is passed
|
||||
// FIXME: we currently conflate token_pos and buffer contents. See https://github.com/ggml-org/llama.cpp/pull/25521#discussion_r3558173819
|
||||
|
||||
class llama_kv_cache_dsv4 : public llama_memory_i {
|
||||
public:
|
||||
@@ -146,7 +148,7 @@ private:
|
||||
std::unique_ptr<llama_dsv4_comp_state> hca_state;
|
||||
std::unique_ptr<llama_dsv4_comp_state> lid_state;
|
||||
|
||||
void clear_compressed(bool data);
|
||||
void clear_compressed(llama_seq_id seq_id, bool data);
|
||||
};
|
||||
|
||||
// DSV4 raw attention only uses the SWA half of kv_raw. The base half is kept
|
||||
|
||||
+1
-2
@@ -313,8 +313,7 @@ llama_model * llama_model_create(llm_arch arch, const llama_model_params & param
|
||||
|
||||
if (model != nullptr) {
|
||||
model->arch = arch;
|
||||
auto & devices = model->devices;
|
||||
if (!devices.empty() && devices[0].is_meta && !llm_arch_supports_sm_tensor(arch)) {
|
||||
if (params.split_mode == LLAMA_SPLIT_MODE_TENSOR && !llm_arch_supports_sm_tensor(arch)) {
|
||||
throw std::runtime_error(std::string("LLAMA_SPLIT_MODE_TENSOR not implemented for architecture '") + llm_arch_name(arch) + "'");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5911,6 +5911,71 @@ static void test_developer_role_to_system_workaround() {
|
||||
}
|
||||
}
|
||||
|
||||
static void test_reasoning_budget_tokens_per_request() {
|
||||
LOG_DBG("%s\n", __func__);
|
||||
// Use Qwen3 template which has <think>...</think> reasoning markers.
|
||||
// The autoparser detects them and sets thinking_start/end_tag, which enables
|
||||
// the reasoning-budget code path in oaicompat_chat_params_parse.
|
||||
auto tmpls = read_templates("models/templates/Qwen-Qwen3-0.6B.jinja");
|
||||
|
||||
server_chat_params opt;
|
||||
opt.tmpls = std::move(tmpls);
|
||||
opt.use_jinja = true;
|
||||
opt.enable_thinking = true;
|
||||
opt.reasoning_budget = -1;
|
||||
opt.reasoning_format = COMMON_REASONING_FORMAT_NONE;
|
||||
|
||||
// Body with per-request reasoning_budget_tokens=0 (suppress thinking).
|
||||
json body = {
|
||||
{"messages", json::array({json{{"role", "user"}, {"content", "hello"}}})},
|
||||
{"reasoning_budget_tokens", 0},
|
||||
};
|
||||
std::vector<raw_buffer> out_files;
|
||||
auto llama_params = oaicompat_chat_params_parse(body, opt, out_files);
|
||||
|
||||
// The per-request value must win over the server default (-1).
|
||||
if (!llama_params.contains("reasoning_budget_tokens")) {
|
||||
throw std::runtime_error("reasoning_budget_tokens missing from llama_params (thinking_end_tag may be empty for this template)");
|
||||
}
|
||||
int got = llama_params["reasoning_budget_tokens"].get<int>();
|
||||
if (got != 0) {
|
||||
throw std::runtime_error(std::string("Expected reasoning_budget_tokens=0, got ") + std::to_string(got));
|
||||
}
|
||||
}
|
||||
|
||||
static void test_reasoning_budget_message_per_request() {
|
||||
LOG_DBG("%s\n", __func__);
|
||||
// Same code path as test_reasoning_budget_tokens_per_request: the Qwen3 template's
|
||||
// <think>...</think> markers enable the reasoning-budget block in oaicompat_chat_params_parse.
|
||||
auto tmpls = read_templates("models/templates/Qwen-Qwen3-0.6B.jinja");
|
||||
|
||||
server_chat_params opt;
|
||||
opt.tmpls = std::move(tmpls);
|
||||
opt.use_jinja = true;
|
||||
opt.enable_thinking = true;
|
||||
opt.reasoning_budget = -1;
|
||||
opt.reasoning_format = COMMON_REASONING_FORMAT_NONE;
|
||||
opt.reasoning_budget_message = "server default";
|
||||
|
||||
// Body with a per-request reasoning_budget_message override.
|
||||
const std::string per_request_message = "per-request message";
|
||||
json body = {
|
||||
{"messages", json::array({json{{"role", "user"}, {"content", "hello"}}})},
|
||||
{"reasoning_budget_message", per_request_message},
|
||||
};
|
||||
std::vector<raw_buffer> out_files;
|
||||
auto llama_params = oaicompat_chat_params_parse(body, opt, out_files);
|
||||
|
||||
// The per-request value must win over the server default.
|
||||
if (!llama_params.contains("reasoning_budget_message")) {
|
||||
throw std::runtime_error("reasoning_budget_message missing from llama_params (thinking_end_tag may be empty for this template)");
|
||||
}
|
||||
std::string got = llama_params["reasoning_budget_message"].get<std::string>();
|
||||
if (got != per_request_message) {
|
||||
throw std::runtime_error("Expected reasoning_budget_message='" + per_request_message + "', got '" + got + "'");
|
||||
}
|
||||
}
|
||||
|
||||
static void test_msg_diffs_compute() {
|
||||
LOG_DBG("%s\n", __func__);
|
||||
{
|
||||
@@ -6068,6 +6133,8 @@ int main(int argc, char ** argv) {
|
||||
test_convert_responses_to_chatcmpl();
|
||||
test_developer_role_to_system_workaround();
|
||||
test_template_generation_prompt();
|
||||
test_reasoning_budget_tokens_per_request();
|
||||
test_reasoning_budget_message_per_request();
|
||||
test_template_output_peg_parsers(detailed_debug);
|
||||
std::cout << "\n[chat] All tests passed!" << '\n';
|
||||
}
|
||||
|
||||
+6
-1
@@ -26,6 +26,7 @@ enum handcrafted_file_type {
|
||||
HANDCRAFTED_HEADER_EMPTY = 800,
|
||||
|
||||
HANDCRAFTED_KV_BAD_KEY_SIZE = 10 + offset_has_kv,
|
||||
HANDCRAFTED_KV_EMPTY_KEY = 15 + offset_has_kv,
|
||||
HANDCRAFTED_KV_BAD_TYPE = 20 + offset_has_kv,
|
||||
// HANDCRAFTED_KV_BAD_VALUE_SIZE = 30 + offset_has_kv, // removed because it can result in allocations > 1 TB (default sanitizer limit)
|
||||
HANDCRAFTED_KV_DUPLICATE_KEY = 40 + offset_has_kv,
|
||||
@@ -64,6 +65,7 @@ static std::string handcrafted_file_type_name(const enum handcrafted_file_type h
|
||||
case HANDCRAFTED_HEADER_EMPTY: return "HEADER_EMPTY";
|
||||
|
||||
case HANDCRAFTED_KV_BAD_KEY_SIZE: return "KV_BAD_KEY_SIZE";
|
||||
case HANDCRAFTED_KV_EMPTY_KEY: return "KV_EMPTY_KEY";
|
||||
case HANDCRAFTED_KV_BAD_TYPE: return "KV_BAD_TYPE";
|
||||
case HANDCRAFTED_KV_DUPLICATE_KEY: return "KV_DUPLICATE_KEY";
|
||||
case HANDCRAFTED_KV_BAD_ALIGN: return "KV_BAD_ALIGN";
|
||||
@@ -284,7 +286,9 @@ static FILE * get_handcrafted_file(const unsigned int seed, const enum handcraft
|
||||
const enum gguf_type type = gguf_type(hft == HANDCRAFTED_KV_BAD_TYPE ? GGUF_TYPE_COUNT : kv_types[i].first);
|
||||
const enum gguf_type type_arr = gguf_type(hft == HANDCRAFTED_KV_BAD_TYPE ? GGUF_TYPE_COUNT : kv_types[i].second);
|
||||
|
||||
const std::string key = "my_key_" + std::to_string((hft == HANDCRAFTED_KV_DUPLICATE_KEY ? i/2 : i));
|
||||
const std::string key = hft == HANDCRAFTED_KV_EMPTY_KEY
|
||||
? ""
|
||||
: "my_key_" + std::to_string((hft == HANDCRAFTED_KV_DUPLICATE_KEY ? i/2 : i));
|
||||
|
||||
if (hft == HANDCRAFTED_KV_BAD_KEY_SIZE) {
|
||||
const uint64_t n = -1;
|
||||
@@ -732,6 +736,7 @@ static std::pair<int, int> test_handcrafted_file(const unsigned int seed) {
|
||||
HANDCRAFTED_HEADER_EMPTY,
|
||||
|
||||
HANDCRAFTED_KV_BAD_KEY_SIZE,
|
||||
HANDCRAFTED_KV_EMPTY_KEY,
|
||||
HANDCRAFTED_KV_BAD_TYPE,
|
||||
HANDCRAFTED_KV_DUPLICATE_KEY,
|
||||
HANDCRAFTED_KV_BAD_ALIGN,
|
||||
|
||||
@@ -78,7 +78,84 @@ static llama_tokens test_baseline(struct llama_model * model, const struct commo
|
||||
}
|
||||
|
||||
|
||||
// Test 2: state load
|
||||
// Test 2: sequence removal isolation
|
||||
// - decode the same prefix into two sequences
|
||||
// - remove sequence 0
|
||||
// - verify that sequence 1 remains unchanged
|
||||
static bool test_seq_rm_isolated(
|
||||
struct llama_model * model,
|
||||
const struct common_params & params,
|
||||
const llama_tokens & tokens) {
|
||||
auto params_ctx = common_context_params_to_llama(params);
|
||||
params_ctx.n_ctx = 256;
|
||||
params_ctx.n_seq_max = 2;
|
||||
params_ctx.kv_unified = true;
|
||||
|
||||
auto ctx = llama_context_ptr{llama_init_from_model(model, params_ctx)};
|
||||
if (!ctx) {
|
||||
LOG_ERR("%s: failed to create context\n", __func__);
|
||||
return false;
|
||||
}
|
||||
|
||||
LOG("\n=== Test 2: sequence removal isolation ===\n");
|
||||
|
||||
const size_t n_tokens = tokens.size() < 128 ? tokens.size() : 128;
|
||||
for (llama_seq_id seq_id = 0; seq_id < 2; ++seq_id) {
|
||||
llama_batch_ptr batch(n_tokens, 0, 1);
|
||||
for (size_t i = 0; i < n_tokens; ++i) {
|
||||
common_batch_add(batch.get(), tokens[i], i, { seq_id }, false);
|
||||
}
|
||||
|
||||
if (llama_decode(ctx.get(), batch.get())) {
|
||||
LOG_ERR("%s: failed to decode prompt for sequence %d\n", __func__, seq_id);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
const auto get_seq_state = [&](llama_seq_id seq_id, std::vector<uint8_t> & state) {
|
||||
const size_t state_size = llama_state_seq_get_size(ctx.get(), seq_id);
|
||||
if (state_size == 0) {
|
||||
LOG_ERR("%s: sequence state is empty\n", __func__);
|
||||
return false;
|
||||
}
|
||||
|
||||
state.resize(state_size);
|
||||
const size_t ncopy = llama_state_seq_get_data(ctx.get(), state.data(), state.size(), seq_id);
|
||||
if (ncopy != state.size()) {
|
||||
LOG_ERR("%s: sequence state length %zu does not match expected length %zu\n",
|
||||
__func__, ncopy, state.size());
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
};
|
||||
|
||||
std::vector<uint8_t> state_before;
|
||||
if (!get_seq_state(1, state_before)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!llama_memory_seq_rm(llama_get_memory(ctx.get()), 0, -1, -1)) {
|
||||
LOG_ERR("%s: failed to remove sequence 0\n", __func__);
|
||||
return false;
|
||||
}
|
||||
|
||||
std::vector<uint8_t> state_after;
|
||||
if (!get_seq_state(1, state_after)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (state_before != state_after) {
|
||||
LOG_ERR("%s: removing sequence 0 changed sequence 1\n", __func__);
|
||||
return false;
|
||||
}
|
||||
|
||||
LOG("PASS\n");
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
// Test 3: state load
|
||||
// - create a new context
|
||||
// - load state from file
|
||||
// - replay the last prompt token
|
||||
@@ -90,7 +167,7 @@ static bool test_state_load(struct llama_model * model, const struct common_para
|
||||
auto smpl = llama_sampler_ptr{llama_sampler_chain_init(sparams)};
|
||||
llama_sampler_chain_add(smpl.get(), llama_sampler_init_dist(params.sampling.seed));
|
||||
|
||||
LOG("\n=== Test 2: state load ===\n");
|
||||
LOG("\n=== Test 3: state load ===\n");
|
||||
|
||||
// Load state from file
|
||||
llama_tokens unused_sts(tokens.size());
|
||||
@@ -126,7 +203,7 @@ static bool test_state_load(struct llama_model * model, const struct common_para
|
||||
}
|
||||
|
||||
|
||||
// Test 3: seq copy (host)
|
||||
// Test 4: seq copy (host)
|
||||
// - create a multi-seq context
|
||||
// - load state from file
|
||||
// - replay the last prompt token
|
||||
@@ -141,7 +218,7 @@ static bool test_seq_cp_host(struct llama_model * model, const struct common_par
|
||||
auto smpl = llama_sampler_ptr{llama_sampler_chain_init(sparams)};
|
||||
llama_sampler_chain_add(smpl.get(), llama_sampler_init_dist(params.sampling.seed));
|
||||
|
||||
LOG("\n=== Test 3: seq copy (host) ===\n");
|
||||
LOG("\n=== Test 4: seq copy (host) ===\n");
|
||||
|
||||
// Load state from file
|
||||
llama_tokens unused_sts(tokens.size());
|
||||
@@ -198,7 +275,7 @@ static bool test_seq_cp_host(struct llama_model * model, const struct common_par
|
||||
}
|
||||
|
||||
|
||||
// Test 4: seq copy (device)
|
||||
// Test 5: seq copy (device)
|
||||
// - create a multi-seq context
|
||||
// - load state from file
|
||||
// - replay the last prompt token
|
||||
@@ -213,7 +290,7 @@ static bool test_seq_cp_device(struct llama_model * model, const struct common_p
|
||||
auto smpl = llama_sampler_ptr{llama_sampler_chain_init(sparams)};
|
||||
llama_sampler_chain_add(smpl.get(), llama_sampler_init_dist(params.sampling.seed));
|
||||
|
||||
LOG("\n=== Test 4: seq copy (device) ===\n");
|
||||
LOG("\n=== Test 5: seq copy (device) ===\n");
|
||||
|
||||
// Load state from file
|
||||
llama_tokens unused_sts(tokens.size());
|
||||
@@ -337,17 +414,22 @@ int main(int argc, char ** argv) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
// Test 2: state load
|
||||
// Test 2: sequence removal isolation
|
||||
if (!test_seq_rm_isolated(model, params, tokens)) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
// Test 3: state load
|
||||
if (!test_state_load(model, params, tokens, result_baseline)) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
// Test 3: seq copy (host)
|
||||
// Test 4: seq copy (host)
|
||||
if (!test_seq_cp_host(model, params, tokens, result_baseline)) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
// Test 4: seq copy (device)
|
||||
// Test 5: seq copy (device)
|
||||
if (!test_seq_cp_device(model, params, tokens, result_baseline)) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
@@ -250,7 +250,8 @@ static int eval_message(mtmd_cli_context & ctx, common_chat_msg & msg) {
|
||||
LOG_DBG("formatted_chat.prompt: %s\n", formatted_chat.c_str());
|
||||
|
||||
mtmd_input_text text;
|
||||
text.text = formatted_chat.c_str();
|
||||
text.text = formatted_chat.data();
|
||||
text.text_len = formatted_chat.size();
|
||||
text.add_special = add_bos;
|
||||
text.parse_special = true;
|
||||
|
||||
|
||||
+3
-2
@@ -809,7 +809,7 @@ void mtmd_free(mtmd_context * ctx) {
|
||||
struct mtmd_tokenizer {
|
||||
mtmd_context * ctx;
|
||||
|
||||
std::string input_text;
|
||||
std::string input_text; // note: can contain null bytes; do not use c_str()
|
||||
bool add_special;
|
||||
bool parse_special;
|
||||
const llama_vocab * vocab;
|
||||
@@ -839,9 +839,10 @@ struct mtmd_tokenizer {
|
||||
size_t n_bitmaps) : ctx(ctx) {
|
||||
add_special = text->add_special;
|
||||
parse_special = text->parse_special;
|
||||
input_text = text->text;
|
||||
vocab = ctx->vocab;
|
||||
|
||||
input_text.assign(text->text, text->text_len);
|
||||
|
||||
std::vector<const mtmd_bitmap *> bitmaps(bmps, bmps + n_bitmaps);
|
||||
auto parts_str = split_text(input_text, ctx->media_marker);
|
||||
size_t i_bm = 0;
|
||||
|
||||
@@ -67,6 +67,7 @@ struct mtmd_batch;
|
||||
|
||||
struct mtmd_input_text {
|
||||
const char * text;
|
||||
size_t text_len;
|
||||
bool add_special;
|
||||
bool parse_special;
|
||||
};
|
||||
|
||||
@@ -126,15 +126,15 @@ It is opt in via the `X-Conversation-Id` header on `POST /v1/chat/completions`.
|
||||
|
||||
The feature lives entirely in `server-stream.{h,cpp}` and rests on three types:
|
||||
|
||||
- `stream_session`: a bounded ring buffer (4 MiB cap, oldest bytes drop first) plus a condvar. `append` pushes raw SSE bytes, `read_from` drains from any offset and blocks for live bytes or finalize, `finalize` wakes readers, `cancel` stops the producer. One conv maps to at most one live session.
|
||||
- `stream_session`: a bounded ring buffer (4 MiB cap, oldest bytes drop first) plus a condvar. `append` pushes raw SSE bytes, `read_from` drains from any offset and blocks for live bytes or finalize, `finalize` wakes readers, `cancel` sets the flag the producer polls. One conv maps to at most one live session.
|
||||
- `stream_session_manager`: a file-static singleton (`g_stream_sessions`) inside `server-stream.cpp`, owns all sessions keyed by conv id, enforces the one conv one session invariant via `create_or_replace`, and runs a GC thread that drops completed sessions past their TTL. Exposed to main only through `server_stream_session_manager_start/stop`.
|
||||
- `stream_pipe_producer` / `stream_pipe_consumer`: the write and read ends. The producer owns the session lifetime and finalizes it on destruction; the consumer is read only and never finalizes, so a reader detaching cannot kill a running generation.
|
||||
|
||||
The implementation is hidden in `server-stream.cpp` (pimpl). The header exposes only the route handler factories, `server_stream_session_attach_pipe`, `server_stream_aware_should_stop`, `server_stream_conv_id_from_headers` and the GC lifecycle; the session, manager and consumer types stay in the `.cpp`.
|
||||
The implementation is hidden in `server-stream.cpp` (pimpl). The header exposes only the route handler factories, the `server_res_spipe` response base, `server_stream_conv_id_from_headers` and the GC lifecycle; the session, manager, consumer and the `server_stream_create_spipe` factory stay in the `.cpp`.
|
||||
|
||||
Producer side: `server_res_generator` attaches a producer pipe when the header is present. The HTTP content provider mirrors every chunk into the ring before writing it to the socket. While a pipe is attached, `server_stream_aware_should_stop` ignores peer disconnect, so a dropped socket does not stop generation: only an explicit `DELETE` does. When the peer leaves early, `on_complete` calls `close()`, which drains the rest of the generation into the ring on the http worker.
|
||||
Producer side: `server_res_generator` extends `server_res_spipe`, which keeps all spipe logic out of the generic `server_http_res`. `set_req` attaches a producer when the header is present, and the wrapped `next` tees each chunk into the ring before the socket, so a chunk lost to a dead wire is already buffered. While attached, `should_stop` ignores peer disconnect: only a `DELETE` stops generation. On an early peer drop, `on_complete` drains the tail into the ring on the http worker.
|
||||
|
||||
Lifetime safety: the producer pipe holds a shared `alive` flag also captured by the session cancel hook. `~server_res_generator` calls `cleanup()` to clear that hook while the reader is still alive, so a `cancel` arriving during teardown can never call `stop()` on a freed response. This ordering is the most fragile part of the feature: finalizing or destroying the producer before `cleanup()` runs reintroduces a use after free.
|
||||
Lifetime safety: the session holds no back reference to the response, so `spipe` is a plain `unique_ptr` touched only by the http worker. `cancel` raises an atomic the producer polls; the producer finalizes the session from its destructor, which also runs `~server_response_reader::stop()` to cancel the generation at the queue level. A `DELETE` stops work by raising the flag and letting the worker unwind.
|
||||
|
||||
Consumer side: `GET /v1/stream/<conv_id>?from=N` opens a `text/event-stream` that replays buffered bytes from offset `N` and blocks for live bytes, so the browser reattaches like a fresh EventSource. An offset below the dropped prefix returns 400.
|
||||
|
||||
@@ -235,6 +235,29 @@ That requires `JSON.stringify` when formatted to message content:
|
||||
}
|
||||
```
|
||||
|
||||
Set `stream: true` in the request body to stream a tool's output as it runs, instead of waiting for it to finish. Only certain tools accept this (for ex. `exec_shell_command`);
|
||||
returns 404 if tool doesn't support it.
|
||||
|
||||
Response is SSE stream, one `data: <json>` line per chunk:
|
||||
|
||||
```json
|
||||
{"chunk": "hello\n"}
|
||||
```
|
||||
|
||||
followed by a final event once the tool returns:
|
||||
|
||||
```json
|
||||
{"done": true}
|
||||
```
|
||||
|
||||
or, if `invoke()` threw:
|
||||
|
||||
```json
|
||||
{"done": true, "error": "..."}
|
||||
```
|
||||
|
||||
There is no `[DONE]` sentinel (unlike `/chat/completions`), the stream ends after the `done`
|
||||
|
||||
### Router mode: how child <--> router communicates
|
||||
|
||||
Upon spawning a new child process using `subprocess`, both child and router listen to the stdout/stderr (combined)
|
||||
|
||||
@@ -431,22 +431,70 @@ json server_chat_convert_anthropic_to_oai(const json & body) {
|
||||
std::string tool_use_id = json_value(block, "tool_use_id", std::string());
|
||||
|
||||
auto result_content = json_value(block, "content", json());
|
||||
std::string result_text;
|
||||
if (result_content.is_string()) {
|
||||
result_text = result_content.get<std::string>();
|
||||
tool_results.push_back({
|
||||
{"role", "tool"},
|
||||
{"tool_call_id", tool_use_id},
|
||||
{"content", result_content.get<std::string>()}
|
||||
});
|
||||
} else if (result_content.is_array()) {
|
||||
// Single-pass: build both text and content_parts, decide format at the end
|
||||
std::string result_text;
|
||||
json content_parts = json::array();
|
||||
bool has_images = false;
|
||||
|
||||
for (const auto & c : result_content) {
|
||||
if (json_value(c, "type", std::string()) == "text") {
|
||||
result_text += json_value(c, "text", std::string());
|
||||
std::string c_type = json_value(c, "type", std::string());
|
||||
if (c_type == "text") {
|
||||
std::string text = json_value(c, "text", std::string());
|
||||
result_text += text;
|
||||
content_parts.push_back({
|
||||
{"type", "text"},
|
||||
{"text", text}
|
||||
});
|
||||
} else if (c_type == "image") {
|
||||
has_images = true;
|
||||
json source = json_value(c, "source", json::object());
|
||||
std::string source_type = json_value(source, "type", std::string());
|
||||
if (source_type == "base64") {
|
||||
std::string media_type = json_value(source, "media_type", std::string("image/jpeg"));
|
||||
std::string data = json_value(source, "data", std::string());
|
||||
std::string url = "data:" + media_type + ";base64," + data;
|
||||
content_parts.push_back({
|
||||
{"type", "image_url"},
|
||||
{"image_url", {{"url", url}}}
|
||||
});
|
||||
} else if (source_type == "url") {
|
||||
content_parts.push_back({
|
||||
{"type", "image_url"},
|
||||
{"image_url", {{"url", json_value(source, "url", std::string())}}}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
tool_results.push_back({
|
||||
{"role", "tool"},
|
||||
{"tool_call_id", tool_use_id},
|
||||
{"content", result_text}
|
||||
});
|
||||
if (!has_images) {
|
||||
// Text-only: collapse to a plain string for maximum compatibility
|
||||
tool_results.push_back({
|
||||
{"role", "tool"},
|
||||
{"tool_call_id", tool_use_id},
|
||||
{"content", result_text}
|
||||
});
|
||||
} else {
|
||||
// Mixed or image-only: use array content parts (OpenAI multimodal tool format)
|
||||
tool_results.push_back({
|
||||
{"role", "tool"},
|
||||
{"tool_call_id", tool_use_id},
|
||||
{"content", content_parts}
|
||||
});
|
||||
}
|
||||
} else {
|
||||
tool_results.push_back({
|
||||
{"role", "tool"},
|
||||
{"tool_call_id", tool_use_id},
|
||||
{"content", ""}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -705,7 +705,8 @@ server_tokens process_mtmd_prompt(mtmd_context * mctx, const std::string & promp
|
||||
std::vector<server_tokens> inputs;
|
||||
// multimodal
|
||||
mtmd_input_text inp_txt = {
|
||||
prompt.c_str(),
|
||||
prompt.data(),
|
||||
prompt.size(),
|
||||
/* add_special */ true,
|
||||
/* parse_special */ true,
|
||||
};
|
||||
@@ -1116,7 +1117,8 @@ json oaicompat_chat_params_parse(
|
||||
|
||||
// Reasoning budget: pass parameters through to sampling layer
|
||||
{
|
||||
int reasoning_budget = json_value(body, "thinking_budget_tokens", -1);
|
||||
int reasoning_budget = json_value(body, "reasoning_budget_tokens",
|
||||
json_value(body, "thinking_budget_tokens", -1));
|
||||
if (reasoning_budget == -1) {
|
||||
reasoning_budget = opt.reasoning_budget;
|
||||
}
|
||||
@@ -1125,7 +1127,7 @@ json oaicompat_chat_params_parse(
|
||||
llama_params["reasoning_budget_tokens"] = reasoning_budget;
|
||||
llama_params["reasoning_budget_start_tag"] = chat_params.thinking_start_tag;
|
||||
llama_params["reasoning_budget_end_tag"] = chat_params.thinking_end_tag;
|
||||
llama_params["reasoning_budget_message"] = opt.reasoning_budget_message;
|
||||
llama_params["reasoning_budget_message"] = json_value(body, "reasoning_budget_message", opt.reasoning_budget_message);
|
||||
llama_params["reasoning_control"] = json_value(body, "reasoning_control", false);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2290,6 +2290,24 @@ private:
|
||||
|
||||
// n_tokens_cur: the number of tokens added to the batch for the current slot
|
||||
void create_checkpoint(server_slot & slot, const int64_t n_tokens_cur, llama_pos pos_min, llama_pos pos_max) {
|
||||
const int id_task = slot.task->id;
|
||||
|
||||
// evict checkpoints within min-step of a previous checkpoint, unless they were
|
||||
// created by the current task
|
||||
int64_t last = -1;
|
||||
for (auto it = slot.prompt.checkpoints.begin(); it != slot.prompt.checkpoints.end(); ) {
|
||||
if (it->id_task != id_task && last >= 0 && it->n_tokens <= last + params_base.checkpoint_min_step) {
|
||||
SLT_TRC(slot, "erasing context checkpoint too close to an earlier one (pos_min = %d, pos_max = %d, n_tokens = %" PRId64 ", size = %.3f MiB)\n",
|
||||
it->pos_min, it->pos_max, it->n_tokens, (float) it->size() / 1024 / 1024);
|
||||
|
||||
it = slot.prompt.checkpoints.erase(it);
|
||||
continue;
|
||||
}
|
||||
|
||||
last = it->n_tokens;
|
||||
++it;
|
||||
}
|
||||
|
||||
while (slot.prompt.checkpoints.size() >= (size_t) params_base.n_ctx_checkpoints) {
|
||||
// make room for the new checkpoint, if needed
|
||||
const auto & cur = slot.prompt.checkpoints.front();
|
||||
@@ -2302,6 +2320,8 @@ private:
|
||||
|
||||
auto & cur = slot.prompt.checkpoints.emplace_back();
|
||||
|
||||
cur.id_task = id_task;
|
||||
|
||||
// [TAG_CHECKPOINTS_FIX_POS_MIN]
|
||||
// TODO: here we incorrectly deterimne that the saved checkpoint data covers the [pos_min, pos_max] range
|
||||
// this is not true for SWA models: https://github.com/ggml-org/llama.cpp/pull/24411#issuecomment-4677983225
|
||||
@@ -3511,7 +3531,10 @@ private:
|
||||
do_checkpoint = do_checkpoint && !has_mtmd;
|
||||
|
||||
// no need to create checkpoints that are too close together, unless it's the last user message
|
||||
do_checkpoint = do_checkpoint && (slot.prompt.checkpoints.empty() || is_last_user_message || n_tokens_start > slot.prompt.checkpoints.back().n_tokens + params_base.checkpoint_min_step);
|
||||
do_checkpoint = do_checkpoint && (
|
||||
slot.prompt.checkpoints.empty() ||
|
||||
is_last_user_message || near_prompt_end ||
|
||||
n_tokens_start > slot.prompt.checkpoints.back().n_tokens + params_base.checkpoint_min_step);
|
||||
SLT_DBG(slot, "main/do_checkpoint = %s, pos_min = %d, pos_max = %d\n", do_checkpoint ? "yes" : "no", pos_min, pos_max);
|
||||
|
||||
// note: we create the checkpoint before calling llama_decode(), so the current batch is not
|
||||
@@ -3979,11 +4002,9 @@ server_context_meta server_context::get_meta() const {
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
|
||||
// generator-like API for HTTP response generation
|
||||
// may have bypass_sleep = true if the task does not use ctx_server
|
||||
struct server_res_generator : server_http_res {
|
||||
struct server_res_generator : server_res_spipe {
|
||||
server_response_reader rd;
|
||||
server_res_generator(server_queue & queue_tasks, server_response & queue_results, int sleep_idle_seconds, bool bypass_sleep = false)
|
||||
: rd(queue_tasks, queue_results, HTTP_POLLING_SECONDS) {
|
||||
@@ -3993,15 +4014,6 @@ struct server_res_generator : server_http_res {
|
||||
queue_tasks.wait_until_no_sleep();
|
||||
}
|
||||
}
|
||||
~server_res_generator() override {
|
||||
// cleanup() must run while rd is still alive (rd is destroyed after this body returns)
|
||||
if (spipe) {
|
||||
spipe->cleanup();
|
||||
}
|
||||
}
|
||||
void stop() override {
|
||||
rd.stop();
|
||||
}
|
||||
void ok(const json & response_data) {
|
||||
status = 200;
|
||||
data = safe_json_to_str(response_data);
|
||||
@@ -4039,6 +4051,8 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
|
||||
auto & rd = res->rd;
|
||||
auto & params = this->params;
|
||||
|
||||
res->set_req(&req); // will also set spipe if needed
|
||||
|
||||
int32_t sse_ping_interval = params.sse_ping_interval;
|
||||
|
||||
try {
|
||||
@@ -4181,7 +4195,7 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
|
||||
}
|
||||
res->status = 200;
|
||||
res->content_type = "text/event-stream";
|
||||
res->next = [res_this = res.get(), res_type, sse_ping_interval, &req](std::string & output) -> bool {
|
||||
res->set_next([res_this = res.get(), res_type, sse_ping_interval](std::string & output) -> bool {
|
||||
static auto format_error = [](task_response_type res_type, const json & res_json) {
|
||||
if (res_type == TASK_RESPONSE_TYPE_ANTHROPIC) {
|
||||
return format_anthropic_sse({
|
||||
@@ -4193,7 +4207,9 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
|
||||
}
|
||||
};
|
||||
|
||||
auto effective_should_stop = server_stream_aware_should_stop(res_this, req.should_stop);
|
||||
auto effective_should_stop = [&res_this]() {
|
||||
return res_this->should_stop();
|
||||
};
|
||||
|
||||
try {
|
||||
if (effective_should_stop()) {
|
||||
@@ -4284,13 +4300,9 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
|
||||
// terminate on exception
|
||||
return false;
|
||||
}
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
// attach a producer pipe to the response when X-Conversation-Id is present.
|
||||
// the pipe mirrors SSE chunks into the ring buffer and wires up the cancel hook.
|
||||
server_stream_session_attach_pipe(*res, req.headers);
|
||||
|
||||
return res;
|
||||
}
|
||||
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
#include "common.h"
|
||||
#include "http.h"
|
||||
#include "server-http.h"
|
||||
#include "server-stream.h"
|
||||
#include "server-common.h"
|
||||
#include "ui.h"
|
||||
|
||||
@@ -530,33 +529,20 @@ static void process_handler_response(server_http_req_ptr && request, server_http
|
||||
std::string chunk;
|
||||
const bool has_next = response->next(chunk);
|
||||
if (!chunk.empty()) {
|
||||
// mirror into the ring buffer first, the session must reflect every SSE chunk
|
||||
// whether or not the wire write below succeeds
|
||||
if (response->spipe) {
|
||||
response->spipe->write(chunk.data(), chunk.size());
|
||||
}
|
||||
if (!sink.write(chunk.data(), chunk.size())) {
|
||||
// peer is gone, stop the wire path here
|
||||
return false;
|
||||
}
|
||||
SRV_DBG("http: streamed chunk: %s\n", chunk.c_str());
|
||||
}
|
||||
if (!has_next) {
|
||||
// producer reached its natural end on the wire, a later close() skips the drain
|
||||
if (response->spipe) {
|
||||
response->spipe->done();
|
||||
}
|
||||
sink.done();
|
||||
SRV_DBG("%s", "http: stream ended\n");
|
||||
}
|
||||
return has_next;
|
||||
};
|
||||
const auto on_complete = [request = q_ptr, response = r_ptr](bool) mutable {
|
||||
// on a dropped peer, close() drains the rest of the generation into the ring buffer
|
||||
if (response->spipe) {
|
||||
response->spipe->close();
|
||||
}
|
||||
response.reset(); // spipe destructor finalizes the session if attached
|
||||
response->on_complete();
|
||||
response.reset();
|
||||
request.reset();
|
||||
};
|
||||
res.set_chunked_content_provider(content_type, chunked_content_provider, on_complete);
|
||||
@@ -564,6 +550,7 @@ static void process_handler_response(server_http_req_ptr && request, server_http
|
||||
res.status = response->status;
|
||||
set_headers(res, response->headers);
|
||||
res.set_content(response->data, response->content_type);
|
||||
response->on_complete();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -11,7 +11,6 @@
|
||||
#include <unordered_map>
|
||||
|
||||
struct common_params;
|
||||
struct stream_pipe_producer; // defined in server-stream.h
|
||||
|
||||
// generator-like API for HTTP response generation
|
||||
// this object response with one of the 2 modes:
|
||||
@@ -25,19 +24,13 @@ struct server_http_res {
|
||||
std::string data;
|
||||
std::map<std::string, std::string> headers;
|
||||
|
||||
// if set, the stream survives a client disconnect: the producer pipe keeps draining into the
|
||||
// ring buffer and finalizes the session on destruction, so no explicit on_stream_end is needed.
|
||||
// shared_ptr (not unique_ptr) so the forward-declared type is safe to delete here.
|
||||
std::shared_ptr<stream_pipe_producer> spipe;
|
||||
|
||||
std::function<bool(std::string &)> next = nullptr;
|
||||
bool is_stream() const {
|
||||
return next != nullptr;
|
||||
}
|
||||
|
||||
// called when the session is cancelled (e.g. DELETE /v1/stream/<conv_id>).
|
||||
// server_res_generator overrides this to stop its reader; the default is a no-op.
|
||||
virtual void stop() {}
|
||||
// fired before req and res are destroyed
|
||||
virtual void on_complete() {}
|
||||
|
||||
virtual ~server_http_res() = default;
|
||||
};
|
||||
|
||||
@@ -219,13 +219,14 @@ void server_model_meta::update_caps() {
|
||||
"LLAMA_ARG_MODEL_URL",
|
||||
"LLAMA_ARG_MMPROJ",
|
||||
"LLAMA_ARG_MMPROJ_URL",
|
||||
"LLAMA_ARG_MMPROJ_AUTO",
|
||||
"LLAMA_ARG_HF_REPO",
|
||||
"LLAMA_ARG_HF_REPO_FILE",
|
||||
});
|
||||
params.offline = true;
|
||||
common_models_handler handler = common_models_handler_init(params, LLAMA_EXAMPLE_SERVER);
|
||||
common_models_handler_apply(handler, params); // note: this won't download the model because offline=true
|
||||
if (params.mmproj.path.empty()) {
|
||||
if (params.no_mmproj || params.mmproj.path.empty()) {
|
||||
multimodal = { false, false };
|
||||
} else {
|
||||
multimodal = mtmd_get_cap_from_file(params.mmproj.path.c_str());
|
||||
|
||||
@@ -96,8 +96,6 @@ struct stream_session {
|
||||
size_t dropped_prefix() const; // bytes evicted from the front due to cap
|
||||
int64_t completed_at() const; // 0 while alive, unix seconds after finalize
|
||||
|
||||
void set_stop_producer(std::function<void()> fn);
|
||||
|
||||
void cancel();
|
||||
|
||||
private:
|
||||
@@ -109,7 +107,6 @@ private:
|
||||
bool done;
|
||||
std::atomic<bool> cancelled; // polled lock-free by the should_stop closure, no mu
|
||||
int64_t completed_ts;
|
||||
std::function<void()> stop_producer;
|
||||
};
|
||||
stream_session::stream_session(std::string conversation_id_, size_t max_bytes_)
|
||||
: conversation_id(std::move(conversation_id_))
|
||||
@@ -217,26 +214,10 @@ int64_t stream_session::completed_at() const {
|
||||
return completed_ts;
|
||||
}
|
||||
|
||||
void stream_session::set_stop_producer(std::function<void()> fn) {
|
||||
std::lock_guard<std::mutex> lock(mu);
|
||||
stop_producer = std::move(fn);
|
||||
}
|
||||
|
||||
void stream_session::cancel() {
|
||||
// flip cancelled first so the producer-side server_stream_aware_should_stop can break out of the
|
||||
// recv() wait even if remove_waiting_task_ids does not notify the condvar (the cancel task
|
||||
// posted by rd.stop() will eventually notify, but we do not want to depend on that timing)
|
||||
// the should_stop closure on both the producer and any HTTP reader polls is_cancelled()
|
||||
// so flipping this is the only signal needed to unwind both sides
|
||||
cancelled.store(true, std::memory_order_release);
|
||||
// copy the hook under the lock then invoke outside, the producer side may grab queue locks
|
||||
// and we do not want to hold our mu across that path
|
||||
std::function<void()> fn;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(mu);
|
||||
fn = stop_producer;
|
||||
}
|
||||
if (fn) {
|
||||
fn();
|
||||
}
|
||||
}
|
||||
|
||||
bool stream_session::is_cancelled() const {
|
||||
@@ -325,8 +306,10 @@ void stream_session_manager::evict_and_cancel(const std::string & conversation_i
|
||||
s = it->second;
|
||||
sessions.erase(it);
|
||||
}
|
||||
// signal the producer side first so the inference is cancelled at the queue level,
|
||||
// then finalize, which wakes any pending HTTP reader and lets the drain exit naturally
|
||||
// cancel first so the producer's on_complete() drain loop and any pending HTTP reader
|
||||
// observe is_cancelled() and stop pulling further output, then finalize to wake readers
|
||||
// blocked in read_from(). note: this does not interrupt the underlying generation itself,
|
||||
// which keeps running to its own natural stop condition (EOS/max_tokens)
|
||||
s->cancel();
|
||||
s->finalize();
|
||||
}
|
||||
@@ -431,65 +414,15 @@ stream_pipe_producer::stream_pipe_producer(stream_session_ptr session)
|
||||
}
|
||||
|
||||
stream_pipe_producer::~stream_pipe_producer() {
|
||||
cleanup();
|
||||
session_->finalize();
|
||||
}
|
||||
|
||||
void stream_pipe_producer::cleanup() {
|
||||
if (!alive_) {
|
||||
return;
|
||||
}
|
||||
alive_->store(false, std::memory_order_release);
|
||||
session_->set_stop_producer(nullptr);
|
||||
alive_.reset();
|
||||
}
|
||||
|
||||
bool stream_pipe_producer::write(const char * data, size_t len) {
|
||||
return session_->append(data, len);
|
||||
}
|
||||
|
||||
void stream_pipe_producer::done() {
|
||||
done_ = true;
|
||||
}
|
||||
|
||||
void stream_pipe_producer::close() {
|
||||
// httplib bails its content provider the moment is_peer_alive() goes false, so pump the rest
|
||||
// of the generation into the ring buffer here. a DELETE flips is_cancelled and cuts it short
|
||||
if (done_ || session_->is_cancelled()) {
|
||||
SRV_TRC("stream_pipe close: skip drain (done=%d cancelled=%d) conv=%s\n",
|
||||
done_ ? 1 : 0, session_->is_cancelled() ? 1 : 0, session_->conversation_id.c_str());
|
||||
return;
|
||||
}
|
||||
SRV_TRC("stream_pipe close: draining conv=%s\n", session_->conversation_id.c_str());
|
||||
size_t drained = 0;
|
||||
std::string chunk;
|
||||
while (true) {
|
||||
chunk.clear();
|
||||
bool has_next = res_->next(chunk);
|
||||
if (!chunk.empty()) {
|
||||
write(chunk.data(), chunk.size());
|
||||
drained += chunk.size();
|
||||
}
|
||||
if (!has_next) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
SRV_TRC("stream_pipe close: drain ended conv=%s bytes=%zu\n", session_->conversation_id.c_str(), drained);
|
||||
}
|
||||
|
||||
std::shared_ptr<stream_pipe_producer> stream_pipe_producer::create(stream_session_ptr session,
|
||||
server_http_res & res) {
|
||||
auto alive = std::make_shared<std::atomic<bool>>(true);
|
||||
auto * res_ptr = &res;
|
||||
session->set_stop_producer([alive, res_ptr]() {
|
||||
if (alive->load(std::memory_order_acquire)) {
|
||||
res_ptr->stop();
|
||||
}
|
||||
});
|
||||
auto pipe = std::shared_ptr<stream_pipe_producer>(new stream_pipe_producer(std::move(session)));
|
||||
pipe->alive_ = std::move(alive);
|
||||
pipe->res_ = res_ptr;
|
||||
return pipe;
|
||||
stream_pipe_producer * stream_pipe_producer::create(stream_session_ptr session) {
|
||||
return new stream_pipe_producer(std::move(session));
|
||||
}
|
||||
|
||||
// stream_pipe_consumer
|
||||
@@ -661,21 +594,68 @@ std::string server_stream_conv_id_from_headers(const std::map<std::string, std::
|
||||
return std::string();
|
||||
}
|
||||
|
||||
void server_stream_session_attach_pipe(server_http_res & res, const std::map<std::string, std::string> & headers) {
|
||||
static stream_pipe_producer * server_stream_create_spipe(const std::map<std::string, std::string> & headers) {
|
||||
std::string conversation_id = server_stream_conv_id_from_headers(headers);
|
||||
SRV_TRC("conv_id=%s (empty=%d)\n", conversation_id.c_str(), conversation_id.empty() ? 1 : 0);
|
||||
if (conversation_id.empty()) {
|
||||
return;
|
||||
return nullptr;
|
||||
}
|
||||
auto session = g_stream_sessions.create_or_replace(conversation_id);
|
||||
res.spipe = stream_pipe_producer::create(session, res);
|
||||
return stream_pipe_producer::create(session);
|
||||
}
|
||||
|
||||
std::function<bool()> server_stream_aware_should_stop(server_http_res * res, std::function<bool()> fallback) {
|
||||
return [res, fallback = std::move(fallback)]() -> bool {
|
||||
if (res->spipe) {
|
||||
return res->spipe->is_cancelled();
|
||||
//
|
||||
// server_res_spipe
|
||||
//
|
||||
|
||||
void server_res_spipe::set_req(const server_http_req * req) {
|
||||
this->req = req;
|
||||
// optionally attach spipe to the response when X-Conversation-Id is present
|
||||
spipe.reset(server_stream_create_spipe(req->headers));
|
||||
}
|
||||
|
||||
bool server_res_spipe::conn_alive() {
|
||||
GGML_ASSERT(req != nullptr);
|
||||
return !req->should_stop();
|
||||
}
|
||||
|
||||
bool server_res_spipe::should_stop() {
|
||||
if (spipe) {
|
||||
// note: if DELETE /v1/stream/<conv_id> is called, is_cancelled() will be true
|
||||
return spipe->is_cancelled();
|
||||
} else {
|
||||
return !conn_alive();
|
||||
}
|
||||
}
|
||||
|
||||
void server_res_spipe::on_complete() {
|
||||
if (!spipe || next_finished) {
|
||||
return;
|
||||
}
|
||||
std::string chunk;
|
||||
while (!spipe->is_cancelled()) {
|
||||
chunk.clear();
|
||||
bool has_next = next_orig(chunk);
|
||||
if (!chunk.empty()) {
|
||||
spipe->write(chunk.data(), chunk.size());
|
||||
}
|
||||
return fallback();
|
||||
if (!has_next) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void server_res_spipe::set_next(std::function<bool(std::string &)> next_fn) {
|
||||
next_orig = std::move(next_fn);
|
||||
next = [this](std::string & out) {
|
||||
bool has_next = next_orig(out);
|
||||
if (spipe) {
|
||||
// if spipe is set, tee-style pipe input to both HTTP and spipe
|
||||
spipe->write(out.data(), out.size());
|
||||
}
|
||||
if (!has_next) {
|
||||
next_finished = true;
|
||||
}
|
||||
return has_next;
|
||||
};
|
||||
}
|
||||
|
||||
@@ -30,36 +30,15 @@ protected:
|
||||
|
||||
// producer end: writes chunks into the ring buffer and owns the session lifetime, finalizing it
|
||||
// on destruction.
|
||||
//
|
||||
// lifetime safety: holds a shared_ptr<atomic<bool>> alive also captured by the session's
|
||||
// stop_producer hook. cleanup() sets alive=false and clears the hook; it must run while the
|
||||
// response the hook calls stop() on is still alive. ~server_res_generator() does this explicitly.
|
||||
struct stream_pipe_producer : stream_pipe {
|
||||
~stream_pipe_producer() override;
|
||||
|
||||
bool write(const char * data, size_t len);
|
||||
|
||||
// mark the natural end on the wire so a later close() is a no-op
|
||||
void done();
|
||||
|
||||
// on a peer drop, pump the response next() into the ring buffer until done. runs on the http
|
||||
// worker from on_complete, no-op after done() or cancel
|
||||
void close();
|
||||
|
||||
// disarm the stop hook and drop the alive guard, must run while the response the hook
|
||||
// references is still alive. idempotent, the destructor calls it too
|
||||
void cleanup();
|
||||
|
||||
// res.stop() is invoked when the session is cancelled, the alive guard ensures stop() is not
|
||||
// called after cleanup() has run
|
||||
static std::shared_ptr<stream_pipe_producer> create(stream_session_ptr session, server_http_res & res);
|
||||
static stream_pipe_producer * create(stream_session_ptr session);
|
||||
|
||||
private:
|
||||
explicit stream_pipe_producer(stream_session_ptr session);
|
||||
|
||||
bool done_ = false;
|
||||
std::shared_ptr<std::atomic<bool>> alive_;
|
||||
server_http_res * res_ = nullptr;
|
||||
};
|
||||
|
||||
void server_stream_session_manager_start();
|
||||
@@ -73,10 +52,22 @@ server_http_context::handler_t server_stream_make_delete_handler();
|
||||
// extract the X-Conversation-Id header value (case-insensitive), empty when absent
|
||||
std::string server_stream_conv_id_from_headers(const std::map<std::string, std::string> & headers);
|
||||
|
||||
// on an X-Conversation-Id header, create or replace the session and attach a producer pipe to res
|
||||
void server_stream_session_attach_pipe(server_http_res & res, const std::map<std::string, std::string> & headers);
|
||||
// implement tee-style pipe (spipe) for "stream replay" functionality
|
||||
struct server_res_spipe : server_http_res {
|
||||
private:
|
||||
// if set, the stream survives a client disconnect:
|
||||
// connection kept alive, output is forwarded to spipe and reuse later
|
||||
std::unique_ptr<stream_pipe_producer> spipe;
|
||||
// if spipe is set, use this next_orig to implement tee-style pipe
|
||||
std::function<bool(std::string &)> next_orig;
|
||||
const server_http_req * req = nullptr;
|
||||
// set once next_orig reports no more data, so on_complete() doesn't re-drain a finished stream
|
||||
bool next_finished = false;
|
||||
|
||||
// should_stop closure that ignores peer disconnect when a pipe is attached, so only an explicit
|
||||
// DELETE stops the producer and generation keeps flowing into the ring buffer. without a pipe it
|
||||
// delegates to fallback, the legacy non-resumable flow
|
||||
std::function<bool()> server_stream_aware_should_stop(server_http_res * res, std::function<bool()> fallback);
|
||||
public:
|
||||
void set_req(const server_http_req * req);
|
||||
bool conn_alive();
|
||||
bool should_stop();
|
||||
void on_complete() override;
|
||||
void set_next(std::function<bool(std::string &)> next_fn);
|
||||
};
|
||||
|
||||
+149
-23
@@ -12,6 +12,7 @@
|
||||
#include <climits>
|
||||
#include <algorithm>
|
||||
#include <unordered_set>
|
||||
#include <functional>
|
||||
|
||||
namespace fs = std::filesystem;
|
||||
|
||||
@@ -51,7 +52,13 @@ public:
|
||||
virtual bool write_file(const std::string & path, const std::string & content) const = 0;
|
||||
// paths relative to `base`, '/'-separated; sets `err` if `base` isn't a directory
|
||||
virtual std::vector<std::string> list_files(const std::string & base, std::string & err) const = 0;
|
||||
virtual exec_result run(const std::vector<std::string> & args, size_t max_output, int timeout_secs) const = 0;
|
||||
// on_chunk, if set, is called with each chunk of output as it is read (before truncation cuts in);
|
||||
// returning false terminates the process early (e.g. the client disconnected)
|
||||
virtual exec_result run(
|
||||
const std::vector<std::string> & args,
|
||||
size_t max_output,
|
||||
int timeout_secs,
|
||||
const std::function<bool(const std::string &)> & on_chunk = nullptr) const = 0;
|
||||
};
|
||||
|
||||
class tools_io_basic : public tools_io {
|
||||
@@ -123,7 +130,11 @@ public:
|
||||
return list_files_fallback(base);
|
||||
}
|
||||
|
||||
exec_result run(const std::vector<std::string> & args, size_t max_output, int timeout_secs) const override {
|
||||
exec_result run(
|
||||
const std::vector<std::string> & args,
|
||||
size_t max_output,
|
||||
int timeout_secs,
|
||||
const std::function<bool(const std::string &)> & on_chunk = nullptr) const override {
|
||||
exec_result res;
|
||||
|
||||
subprocess_s proc;
|
||||
@@ -164,8 +175,14 @@ public:
|
||||
size_t len = strlen(buf);
|
||||
if (output.size() + len <= max_output) {
|
||||
output.append(buf, len);
|
||||
if (on_chunk && !on_chunk(std::string(buf, len))) {
|
||||
subprocess_terminate(&proc);
|
||||
break;
|
||||
}
|
||||
} else {
|
||||
output.append(buf, max_output - output.size());
|
||||
size_t remaining = max_output - output.size();
|
||||
output.append(buf, remaining);
|
||||
if (on_chunk && remaining > 0) on_chunk(std::string(buf, remaining));
|
||||
truncated = true;
|
||||
}
|
||||
}
|
||||
@@ -287,7 +304,7 @@ struct server_tool_read_file : server_tool {
|
||||
};
|
||||
}
|
||||
|
||||
json invoke(json params) const override {
|
||||
json invoke(json params, server_tool::stream *) const override {
|
||||
std::string path = params.at("path").get<std::string>();
|
||||
int start_line = json_value(params, "start_line", 1);
|
||||
int end_line = json_value(params, "end_line", -1); // -1 = no limit
|
||||
@@ -376,7 +393,7 @@ struct server_tool_file_glob_search : server_tool {
|
||||
};
|
||||
}
|
||||
|
||||
json invoke(json params) const override {
|
||||
json invoke(json params, server_tool::stream *) const override {
|
||||
std::string base = params.at("path").get<std::string>();
|
||||
std::string include = json_value(params, "include", std::string("**"));
|
||||
std::string exclude = json_value(params, "exclude", std::string(""));
|
||||
@@ -457,7 +474,7 @@ struct server_tool_grep_search : server_tool {
|
||||
};
|
||||
}
|
||||
|
||||
json invoke(json params) const override {
|
||||
json invoke(json params, server_tool::stream *) const override {
|
||||
std::string path = params.at("path").get<std::string>();
|
||||
std::string pat_str = params.at("pattern").get<std::string>();
|
||||
std::string include = json_value(params, "include", std::string("**"));
|
||||
@@ -577,6 +594,7 @@ struct server_tool_exec_shell_command : server_tool {
|
||||
name = "exec_shell_command";
|
||||
display_name = "Execute shell command";
|
||||
permission_write = true;
|
||||
support_stream = true;
|
||||
}
|
||||
|
||||
json get_definition() const override {
|
||||
@@ -598,7 +616,7 @@ struct server_tool_exec_shell_command : server_tool {
|
||||
};
|
||||
}
|
||||
|
||||
json invoke(json params) const override {
|
||||
json invoke(json params, server_tool::stream * st) const override {
|
||||
std::string command = params.at("command").get<std::string>();
|
||||
int timeout = json_value(params, "timeout", 10);
|
||||
size_t max_output = (size_t) json_value(params, "max_output_size", (int) SERVER_TOOL_EXEC_SHELL_COMMAND_MAX_OUTPUT_SIZE);
|
||||
@@ -612,7 +630,24 @@ struct server_tool_exec_shell_command : server_tool {
|
||||
std::vector<std::string> args = {"sh", "-c", command};
|
||||
#endif
|
||||
|
||||
auto io = make_tools_io(params);
|
||||
auto io = make_tools_io(params);
|
||||
|
||||
if (st) {
|
||||
auto res = io->run(args, max_output, timeout, [st](const std::string & chunk) {
|
||||
st->push(chunk);
|
||||
return !st->alive || st->alive();
|
||||
});
|
||||
if (st->alive && !st->alive()) {
|
||||
return json();
|
||||
}
|
||||
std::string tail = string_format("\n[exit code: %d]", res.exit_code);
|
||||
if (res.timed_out) {
|
||||
tail += " [exit due to timed out]";
|
||||
}
|
||||
st->push(tail);
|
||||
return json();
|
||||
}
|
||||
|
||||
auto res = io->run(args, max_output, timeout);
|
||||
|
||||
std::string text_output = res.output;
|
||||
@@ -654,7 +689,7 @@ struct server_tool_write_file : server_tool {
|
||||
};
|
||||
}
|
||||
|
||||
json invoke(json params) const override {
|
||||
json invoke(json params, server_tool::stream *) const override {
|
||||
std::string path = params.at("path").get<std::string>();
|
||||
std::string content = params.at("content").get<std::string>();
|
||||
|
||||
@@ -710,7 +745,7 @@ struct server_tool_edit_file : server_tool {
|
||||
};
|
||||
}
|
||||
|
||||
json invoke(json params) const override {
|
||||
json invoke(json params, server_tool::stream *) const override {
|
||||
std::string path = params.at("path").get<std::string>();
|
||||
const json & edits_json = params.at("edits");
|
||||
|
||||
@@ -1018,7 +1053,7 @@ struct server_tool_get_datetime : server_tool {
|
||||
};
|
||||
}
|
||||
|
||||
json invoke(json) const override {
|
||||
json invoke(json, server_tool::stream *) const override {
|
||||
auto now = std::chrono::system_clock::now();
|
||||
auto time = std::chrono::system_clock::to_time_t(now);
|
||||
|
||||
@@ -1026,6 +1061,59 @@ struct server_tool_get_datetime : server_tool {
|
||||
}
|
||||
};
|
||||
|
||||
struct server_tool_stream_result : server_task_result {
|
||||
std::string chunk;
|
||||
bool done = false;
|
||||
std::string error_msg;
|
||||
|
||||
json to_json() override {
|
||||
if (!done) {
|
||||
return {{"chunk", chunk}};
|
||||
} else {
|
||||
json result = {{"done", true}};
|
||||
if (!error_msg.empty()) {
|
||||
result["error"] = error_msg;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
void server_tool::stream::push(const std::string & chunk) {
|
||||
if (chunk.empty()) return;
|
||||
auto r = std::make_unique<server_tool_stream_result>();
|
||||
r->id = id;
|
||||
r->chunk = chunk;
|
||||
qr.send(std::move(r));
|
||||
}
|
||||
|
||||
struct server_tools_res : server_http_res {
|
||||
std::thread worker;
|
||||
server_response * qr = nullptr; // set only for streaming responses
|
||||
int id = -1;
|
||||
|
||||
~server_tools_res() override {
|
||||
if (worker.joinable()) {
|
||||
worker.join();
|
||||
}
|
||||
if (qr) {
|
||||
qr->remove_waiting_task_id(id);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
static server_tool & find_tool(std::vector<std::unique_ptr<server_tool>> & tools, const std::string & name, bool require_stream) {
|
||||
for (auto & t : tools) {
|
||||
if (t->name == name) {
|
||||
if (require_stream && !t->support_stream) {
|
||||
throw std::invalid_argument(string_format("tool \"%s\" does not support stream = true", name.c_str()));
|
||||
}
|
||||
return *t;
|
||||
}
|
||||
}
|
||||
throw std::invalid_argument(string_format("unknown tool \"%s\"", name.c_str()));
|
||||
}
|
||||
|
||||
//
|
||||
// public API
|
||||
//
|
||||
@@ -1090,16 +1178,63 @@ void server_tools::setup(const std::vector<std::string> & enabled_tools) {
|
||||
};
|
||||
|
||||
handle_post = [this](const server_http_req & req) -> server_http_res_ptr {
|
||||
auto res = std::make_unique<server_http_res>();
|
||||
auto res = std::make_unique<server_tools_res>();
|
||||
try {
|
||||
json body = json::parse(req.body);
|
||||
std::string tool_name = body.at("tool").get<std::string>();
|
||||
json params = body.value("params", json::object());
|
||||
json result = invoke(tool_name, params);
|
||||
res->data = safe_json_to_str(result);
|
||||
bool stream = body.value("stream", false);
|
||||
|
||||
server_tool & tool = find_tool(tools, tool_name, stream);
|
||||
|
||||
if (stream) {
|
||||
int id = res_id.fetch_add(1);
|
||||
queue_res.add_waiting_task_id(id);
|
||||
res->qr = &queue_res;
|
||||
res->id = id;
|
||||
|
||||
res->worker = std::thread([this, id, &req, &tool, params]() mutable {
|
||||
server_tool::stream st{queue_res, id, [&req]() {
|
||||
return !req.should_stop();
|
||||
}};
|
||||
|
||||
auto done = std::make_unique<server_tool_stream_result>();
|
||||
try {
|
||||
tool.invoke(params, &st);
|
||||
} catch (const std::exception & e) {
|
||||
done->error_msg = e.what();
|
||||
} catch (...) {
|
||||
done->error_msg = "An unknown error occurred";
|
||||
}
|
||||
done->id = st.id;
|
||||
done->done = true;
|
||||
st.qr.send(std::move(done));
|
||||
});
|
||||
|
||||
res->content_type = "text/event-stream";
|
||||
res->status = 200;
|
||||
res->next = [this, id](std::string & output) -> bool {
|
||||
auto result = queue_res.recv(id);
|
||||
auto * r = dynamic_cast<server_tool_stream_result *>(result.get());
|
||||
GGML_ASSERT(r != nullptr);
|
||||
output = "data: " + safe_json_to_str(r->to_json()) + "\n\n";
|
||||
if (r->done) {
|
||||
queue_res.remove_waiting_task_id(id);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
};
|
||||
} else {
|
||||
json result = tool.invoke(params, nullptr);
|
||||
res->status = 200;
|
||||
res->data = safe_json_to_str(result);
|
||||
}
|
||||
} catch (const json::exception & e) {
|
||||
res->status = 400;
|
||||
res->data = safe_json_to_str(format_error_response(e.what(), ERROR_TYPE_INVALID_REQUEST));
|
||||
} catch (const std::invalid_argument & e) {
|
||||
res->status = 404;
|
||||
res->data = safe_json_to_str(format_error_response(e.what(), ERROR_TYPE_INVALID_REQUEST));
|
||||
} catch (const std::exception & e) {
|
||||
SRV_ERR("got exception: %s\n", e.what());
|
||||
res->status = 500;
|
||||
@@ -1108,12 +1243,3 @@ void server_tools::setup(const std::vector<std::string> & enabled_tools) {
|
||||
return res;
|
||||
};
|
||||
}
|
||||
|
||||
json server_tools::invoke(const std::string & name, const json & params) {
|
||||
for (auto & t : tools) {
|
||||
if (t->name == name) {
|
||||
return t->invoke(params);
|
||||
}
|
||||
}
|
||||
return {{"error", "unknown tool: " + name}};
|
||||
}
|
||||
|
||||
@@ -2,15 +2,27 @@
|
||||
|
||||
#include "server-common.h"
|
||||
#include "server-http.h"
|
||||
#include "server-queue.h"
|
||||
|
||||
#include <atomic>
|
||||
#include <functional>
|
||||
|
||||
struct server_tool {
|
||||
std::string name;
|
||||
std::string display_name;
|
||||
bool permission_write = false;
|
||||
bool support_stream = false; // if true, output can be streamed
|
||||
|
||||
virtual ~server_tool() = default;
|
||||
virtual json get_definition() const = 0;
|
||||
virtual json invoke(json params) const = 0;
|
||||
|
||||
struct stream {
|
||||
server_response & qr;
|
||||
int id;
|
||||
std::function<bool()> alive;
|
||||
void push(const std::string & chunk);
|
||||
};
|
||||
virtual json invoke(json params, stream * st = nullptr) const = 0;
|
||||
|
||||
json to_json() const;
|
||||
};
|
||||
@@ -18,8 +30,11 @@ struct server_tool {
|
||||
struct server_tools {
|
||||
std::vector<std::unique_ptr<server_tool>> tools;
|
||||
|
||||
// for streaming
|
||||
server_response queue_res;
|
||||
std::atomic<int> res_id{0};
|
||||
|
||||
void setup(const std::vector<std::string> & enabled_tools);
|
||||
json invoke(const std::string & name, const json & params);
|
||||
|
||||
server_http_context::handler_t handle_get;
|
||||
server_http_context::handler_t handle_post;
|
||||
|
||||
@@ -402,6 +402,65 @@ def test_anthropic_tool_result_with_text():
|
||||
assert len(res.body["content"]) > 0
|
||||
|
||||
|
||||
def test_anthropic_tool_result_with_image():
|
||||
"""Test tool result containing mixed text and image blocks
|
||||
|
||||
Verifies that image blocks inside Anthropic tool_result content are
|
||||
properly converted to OpenAI image_url format rather than being
|
||||
silently dropped. With a non-multimodal model, the converted image
|
||||
triggers a clear error message instead of being ignored.
|
||||
"""
|
||||
server.jinja = True
|
||||
server.start()
|
||||
|
||||
# Small 1x1 red PNG image in base64 (same as vision tests)
|
||||
red_pixel_png = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="
|
||||
|
||||
res = server.make_request("POST", "/v1/messages", data={
|
||||
"model": "test",
|
||||
"max_tokens": 100,
|
||||
"messages": [
|
||||
{"role": "user", "content": "What is in this image?"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{
|
||||
"type": "tool_use",
|
||||
"id": "tool_1",
|
||||
"name": "read",
|
||||
"input": {"file": "test.png"}
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{
|
||||
"type": "tool_result",
|
||||
"tool_use_id": "tool_1",
|
||||
"content": [
|
||||
{"type": "text", "text": "File: test.png"},
|
||||
{
|
||||
"type": "image",
|
||||
"source": {
|
||||
"type": "base64",
|
||||
"media_type": "image/png",
|
||||
"data": red_pixel_png
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
# Without the fix, image block would cause "unsupported content[].type"
|
||||
# With the fix, image is converted to image_url but tinyllama doesn't support images
|
||||
assert res.status_code == 500
|
||||
assert "image input is not supported" in res.body.get("error", {}).get("message", "").lower()
|
||||
|
||||
|
||||
def test_anthropic_tool_result_error():
|
||||
"""Test tool result with error flag"""
|
||||
server.jinja = True
|
||||
|
||||
@@ -105,6 +105,24 @@ def test_tools_builtin_edit_file_rejects_non_unique_old_text():
|
||||
os.remove(log_path)
|
||||
|
||||
|
||||
def test_tools_builtin_exec_shell_command_stream():
|
||||
global server
|
||||
server.start()
|
||||
|
||||
events = list(server.make_stream_request("POST", "/tools", data={
|
||||
"tool": "exec_shell_command",
|
||||
"params": {"command": "echo hello"},
|
||||
"stream": True,
|
||||
}))
|
||||
|
||||
assert len(events) >= 2
|
||||
assert events[-1]["done"] is True
|
||||
assert not events[-1].get("error")
|
||||
chunks = "".join(e["chunk"] for e in events[:-1])
|
||||
assert "hello" in chunks
|
||||
assert "[exit code: 0]" in chunks
|
||||
|
||||
|
||||
def test_tools_builtin_edit_file_rejects_overlapping_edits():
|
||||
global server
|
||||
server.start()
|
||||
|
||||
Vendored
+30
-5
@@ -3705,6 +3705,12 @@ write_content_chunked(Stream &strm, const ContentProvider &content_provider,
|
||||
// Trailer
|
||||
if (trailer) {
|
||||
for (const auto &kv : *trailer) {
|
||||
// Skip fields with invalid names or values to prevent response
|
||||
// splitting via CR/LF injection, matching set_header().
|
||||
if (!fields::is_field_name(kv.first) ||
|
||||
!fields::is_field_value(kv.second)) {
|
||||
continue;
|
||||
}
|
||||
std::string field_line = kv.first + ": " + kv.second + "\r\n";
|
||||
if (!write_data(strm, field_line.data(), field_line.size())) {
|
||||
ok = false;
|
||||
@@ -8301,8 +8307,8 @@ void Server::apply_ranges(const Request &req, Response &res,
|
||||
}
|
||||
}
|
||||
|
||||
auto length = std::to_string(res.body.size());
|
||||
res.set_header("Content-Length", length);
|
||||
res.content_length_ = res.body.size();
|
||||
res.set_header("Content-Length", std::to_string(res.content_length_));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10270,6 +10276,11 @@ Result ClientImpl::Get(const std::string &path,
|
||||
return Get(path, Headers(), std::move(progress));
|
||||
}
|
||||
|
||||
Result ClientImpl::Get(const std::string &path, const Params ¶ms,
|
||||
DownloadProgress progress) {
|
||||
return Get(path, params, Headers(), std::move(progress));
|
||||
}
|
||||
|
||||
Result ClientImpl::Get(const std::string &path, const Params ¶ms,
|
||||
const Headers &headers,
|
||||
DownloadProgress progress) {
|
||||
@@ -11348,6 +11359,10 @@ Result Client::Get(const std::string &path, const Headers &headers,
|
||||
return cli_->Get(path, headers, std::move(response_handler),
|
||||
std::move(content_receiver), std::move(progress));
|
||||
}
|
||||
Result Client::Get(const std::string &path, const Params ¶ms,
|
||||
DownloadProgress progress) {
|
||||
return cli_->Get(path, params, std::move(progress));
|
||||
}
|
||||
Result Client::Get(const std::string &path, const Params ¶ms,
|
||||
const Headers &headers, DownloadProgress progress) {
|
||||
return cli_->Get(path, params, headers, std::move(progress));
|
||||
@@ -12076,11 +12091,18 @@ bool SSLServer::update_certs_pem(const char *cert_pem,
|
||||
|
||||
// SSL HTTP client implementation
|
||||
SSLClient::~SSLClient() {
|
||||
if (ctx_) { tls::free_context(ctx_); }
|
||||
// Make sure to shut down SSL since shutdown_ssl will resolve to the
|
||||
// base function rather than the derived function once we get to the
|
||||
// base class destructor, and won't free the SSL (causing a leak).
|
||||
// This must happen before the context is freed below: some backends
|
||||
// (e.g. mbedTLS) have the SSL session borrow a raw pointer into the
|
||||
// context, so freeing the context first leaves close_notify reading
|
||||
// freed memory.
|
||||
shutdown_ssl_impl(socket_, true);
|
||||
if (ctx_) {
|
||||
tls::free_context(ctx_);
|
||||
ctx_ = nullptr;
|
||||
}
|
||||
}
|
||||
|
||||
bool SSLClient::is_valid() const { return ctx_ != nullptr; }
|
||||
@@ -16501,6 +16523,11 @@ WebSocketClient::~WebSocketClient() {
|
||||
bool WebSocketClient::is_valid() const { return is_valid_; }
|
||||
|
||||
void WebSocketClient::shutdown_and_close() {
|
||||
// Send the close frame while the TLS session is still alive: ws_ holds an
|
||||
// SSLSocketStream that keeps a raw pointer to tls_session_, so the session
|
||||
// must outlive ws_->close() and ws_.reset() to avoid a use-after-free.
|
||||
if (ws_ && ws_->is_open()) { ws_->close(); }
|
||||
ws_.reset();
|
||||
#ifdef CPPHTTPLIB_SSL_ENABLED
|
||||
if (is_ssl_) {
|
||||
if (tls_session_) {
|
||||
@@ -16510,8 +16537,6 @@ void WebSocketClient::shutdown_and_close() {
|
||||
}
|
||||
}
|
||||
#endif
|
||||
if (ws_ && ws_->is_open()) { ws_->close(); }
|
||||
ws_.reset();
|
||||
if (sock_ != INVALID_SOCKET) {
|
||||
detail::shutdown_socket(sock_);
|
||||
detail::close_socket(sock_);
|
||||
|
||||
Vendored
+4
-2
@@ -8,8 +8,8 @@
|
||||
#ifndef CPPHTTPLIB_HTTPLIB_H
|
||||
#define CPPHTTPLIB_HTTPLIB_H
|
||||
|
||||
#define CPPHTTPLIB_VERSION "0.49.0"
|
||||
#define CPPHTTPLIB_VERSION_NUM "0x003100"
|
||||
#define CPPHTTPLIB_VERSION "0.50.1"
|
||||
#define CPPHTTPLIB_VERSION_NUM "0x003201"
|
||||
|
||||
#ifdef _WIN32
|
||||
#if defined(_WIN32_WINNT) && _WIN32_WINNT < 0x0A00
|
||||
@@ -2219,6 +2219,7 @@ public:
|
||||
Result Get(const std::string &path, const Headers &headers, DownloadProgress progress = nullptr);
|
||||
Result Get(const std::string &path, const Headers &headers, ContentReceiver content_receiver, DownloadProgress progress = nullptr);
|
||||
Result Get(const std::string &path, const Headers &headers, ResponseHandler response_handler, ContentReceiver content_receiver, DownloadProgress progress = nullptr);
|
||||
Result Get(const std::string &path, const Params ¶ms, DownloadProgress progress = nullptr);
|
||||
Result Get(const std::string &path, const Params ¶ms, const Headers &headers, DownloadProgress progress = nullptr);
|
||||
Result Get(const std::string &path, const Params ¶ms, const Headers &headers, ContentReceiver content_receiver, DownloadProgress progress = nullptr);
|
||||
Result Get(const std::string &path, const Params ¶ms, const Headers &headers, ResponseHandler response_handler, ContentReceiver content_receiver, DownloadProgress progress = nullptr);
|
||||
@@ -2602,6 +2603,7 @@ public:
|
||||
Result Get(const std::string &path, const Headers &headers, DownloadProgress progress = nullptr);
|
||||
Result Get(const std::string &path, const Headers &headers, ContentReceiver content_receiver, DownloadProgress progress = nullptr);
|
||||
Result Get(const std::string &path, const Headers &headers, ResponseHandler response_handler, ContentReceiver content_receiver, DownloadProgress progress = nullptr);
|
||||
Result Get(const std::string &path, const Params ¶ms, DownloadProgress progress = nullptr);
|
||||
Result Get(const std::string &path, const Params ¶ms, const Headers &headers, DownloadProgress progress = nullptr);
|
||||
Result Get(const std::string &path, const Params ¶ms, const Headers &headers, ContentReceiver content_receiver, DownloadProgress progress = nullptr);
|
||||
Result Get(const std::string &path, const Params ¶ms, const Headers &headers, ResponseHandler response_handler, ContentReceiver content_receiver, DownloadProgress progress = nullptr);
|
||||
|
||||
Reference in New Issue
Block a user