1888 lines
59 KiB
C++
1888 lines
59 KiB
C++
/*
|
|
Dynagate LLM request proxying handlers (C++20 port)
|
|
Created by Luxferre in 2026, released into the public domain
|
|
*/
|
|
|
|
#ifndef _DARWIN_C_SOURCE
|
|
#define _DARWIN_C_SOURCE
|
|
#endif
|
|
|
|
#include "dynagate.hpp"
|
|
|
|
#include <algorithm>
|
|
#include <chrono>
|
|
#include <cstdio>
|
|
#include <cstdlib>
|
|
#include <ctime>
|
|
#include <mutex>
|
|
#include <thread>
|
|
|
|
#include <curl/curl.h>
|
|
|
|
namespace {
|
|
|
|
std::once_flag g_curl_init_flag;
|
|
|
|
struct ScopedCurl {
|
|
CURL *handle = nullptr;
|
|
ScopedCurl() : handle(curl_easy_init()) {}
|
|
~ScopedCurl() {
|
|
if (handle) {
|
|
curl_easy_cleanup(handle);
|
|
}
|
|
}
|
|
ScopedCurl(const ScopedCurl &) = delete;
|
|
ScopedCurl &operator=(const ScopedCurl &) = delete;
|
|
};
|
|
|
|
struct ScopedCurlSlist {
|
|
struct curl_slist *list = nullptr;
|
|
ScopedCurlSlist() = default;
|
|
~ScopedCurlSlist() {
|
|
if (list) {
|
|
curl_slist_free_all(list);
|
|
}
|
|
}
|
|
ScopedCurlSlist(const ScopedCurlSlist &) = delete;
|
|
ScopedCurlSlist &operator=(const ScopedCurlSlist &) = delete;
|
|
|
|
void append(const std::string &s) {
|
|
list = curl_slist_append(list, s.c_str());
|
|
}
|
|
};
|
|
|
|
int64_t unix_nano_now() {
|
|
auto now = std::chrono::system_clock::now().time_since_epoch();
|
|
return std::chrono::duration_cast<std::chrono::nanoseconds>(now).count();
|
|
}
|
|
|
|
int64_t unix_sec_now() {
|
|
auto now = std::chrono::system_clock::now().time_since_epoch();
|
|
return std::chrono::duration_cast<std::chrono::seconds>(now).count();
|
|
}
|
|
|
|
void send_json_error(ResponseWriter &w,
|
|
int status_code,
|
|
std::string_view message,
|
|
std::string_view type,
|
|
bool include_code_and_param = false,
|
|
std::string_view code = "") {
|
|
Json err_obj = Json::object({
|
|
{"message", std::string(message)},
|
|
{"type", std::string(type)},
|
|
});
|
|
if (include_code_and_param) {
|
|
err_obj["param"] = nullptr;
|
|
err_obj["code"] = std::string(code);
|
|
}
|
|
Json resp = Json::object({
|
|
{"error", std::move(err_obj)},
|
|
});
|
|
std::string body = resp.dump() + "\n";
|
|
w.header().set("Content-Type", "application/json");
|
|
w.write_header(status_code);
|
|
w.write(body);
|
|
w.flush();
|
|
}
|
|
|
|
bool wait_backoff_or_cancelled(std::chrono::milliseconds delay,
|
|
const HttpRequest &req) {
|
|
auto start = std::chrono::steady_clock::now();
|
|
while (true) {
|
|
if (req.is_cancelled && req.is_cancelled()) {
|
|
return false;
|
|
}
|
|
auto elapsed = std::chrono::steady_clock::now() - start;
|
|
if (elapsed >= delay) {
|
|
return true;
|
|
}
|
|
auto remaining =
|
|
std::chrono::duration_cast<std::chrono::milliseconds>(delay - elapsed);
|
|
auto step = std::min(remaining, std::chrono::milliseconds(20));
|
|
if (step.count() > 0) {
|
|
std::this_thread::sleep_for(step);
|
|
}
|
|
}
|
|
}
|
|
|
|
void build_outgoing_headers(const HttpRequest &req,
|
|
const ModelConfig &trial,
|
|
size_t trial_num,
|
|
ScopedCurlSlist &slist) {
|
|
HeaderList out_headers;
|
|
for (const auto &kv : req.headers.items) {
|
|
std::string k_lower = str_to_lower(kv.first);
|
|
if (k_lower == "authorization" || k_lower == "host" ||
|
|
k_lower == "content-length" || k_lower == "transfer-encoding" ||
|
|
k_lower == "connection" || k_lower == "expect") {
|
|
continue;
|
|
}
|
|
out_headers.add(kv.first, kv.second);
|
|
}
|
|
|
|
if (trial.key == "-blank-") {
|
|
out_headers.set("Authorization", "Bearer");
|
|
} else if (!trial.key.empty() && trial.key != "-") {
|
|
out_headers.set("Authorization", "Bearer " + trial.key);
|
|
}
|
|
out_headers.set("Content-Type", "application/json");
|
|
|
|
if (!trial.extra.empty()) {
|
|
Json extra_headers;
|
|
if (!Json::parse(trial.extra, extra_headers)) {
|
|
log_msg("Trial %zu: Failed to parse extra headers JSON", trial_num);
|
|
} else if (extra_headers.is_object()) {
|
|
for (const auto &kv : extra_headers.oval) {
|
|
std::string val_str =
|
|
kv.second.is_string() ? kv.second.sval : kv.second.dump();
|
|
out_headers.set(kv.first, val_str);
|
|
}
|
|
} else {
|
|
log_msg("Trial %zu: Extra headers JSON is not an object", trial_num);
|
|
}
|
|
}
|
|
|
|
// Disable libcurl's automatic Expect: 100-continue delay
|
|
slist.append("Expect:");
|
|
for (const auto &kv : out_headers.items) {
|
|
slist.append(kv.first + ": " + kv.second);
|
|
}
|
|
}
|
|
|
|
struct UpstreamTransferCtx {
|
|
bool is_stream = false;
|
|
int status_code = 0;
|
|
bool headers_complete = false;
|
|
bool stream_started = false;
|
|
bool write_failed = false;
|
|
HeaderList resp_headers;
|
|
std::string resp_body;
|
|
std::string err_body;
|
|
std::function<void()> on_stream_start;
|
|
std::function<bool(std::string_view)> on_stream_data;
|
|
};
|
|
|
|
size_t curl_header_cb(char *buffer, size_t size, size_t nitems, void *userdata) {
|
|
size_t total = size * nitems;
|
|
auto *ctx = static_cast<UpstreamTransferCtx *>(userdata);
|
|
std::string_view line(buffer, total);
|
|
|
|
if (line.size() >= 5 && line.substr(0, 5) == "HTTP/") {
|
|
size_t sp = line.find(' ');
|
|
if (sp != std::string_view::npos) {
|
|
std::string code_str = str_trim(line.substr(sp + 1));
|
|
ctx->status_code = std::atoi(code_str.c_str());
|
|
}
|
|
ctx->resp_headers.items.clear();
|
|
ctx->headers_complete = false;
|
|
return total;
|
|
}
|
|
|
|
if (line == "\r\n" || line == "\n") {
|
|
if (ctx->status_code >= 200 && ctx->status_code < 300) {
|
|
ctx->headers_complete = true;
|
|
if (ctx->is_stream && !ctx->stream_started && ctx->on_stream_start) {
|
|
ctx->stream_started = true;
|
|
ctx->on_stream_start();
|
|
}
|
|
}
|
|
return total;
|
|
}
|
|
|
|
if (ctx->status_code >= 100 && ctx->status_code < 200) {
|
|
return total;
|
|
}
|
|
|
|
size_t colon = line.find(':');
|
|
if (colon != std::string_view::npos) {
|
|
std::string key = str_trim(line.substr(0, colon));
|
|
std::string val = str_trim(line.substr(colon + 1));
|
|
if (!key.empty()) {
|
|
ctx->resp_headers.add(key, val);
|
|
}
|
|
}
|
|
return total;
|
|
}
|
|
|
|
size_t curl_write_cb(char *ptr, size_t size, size_t nmemb, void *userdata) {
|
|
size_t total = size * nmemb;
|
|
auto *ctx = static_cast<UpstreamTransferCtx *>(userdata);
|
|
|
|
if (ctx->status_code < 200 || ctx->status_code >= 300) {
|
|
if (ctx->err_body.size() < 1024) {
|
|
size_t space = 1024 - ctx->err_body.size();
|
|
ctx->err_body.append(ptr, std::min(total, space));
|
|
}
|
|
return total;
|
|
}
|
|
|
|
if (ctx->is_stream) {
|
|
if (!ctx->stream_started && ctx->on_stream_start) {
|
|
ctx->stream_started = true;
|
|
ctx->on_stream_start();
|
|
}
|
|
if (ctx->on_stream_data) {
|
|
if (!ctx->on_stream_data(std::string_view(ptr, total))) {
|
|
ctx->write_failed = true;
|
|
return 0;
|
|
}
|
|
}
|
|
return total;
|
|
}
|
|
|
|
ctx->resp_body.append(ptr, total);
|
|
return total;
|
|
}
|
|
|
|
bool should_skip_upstream_resp_header(std::string_view key,
|
|
bool skip_content_type) {
|
|
std::string k_lower = str_to_lower(key);
|
|
if (k_lower == "transfer-encoding" || k_lower == "content-length" ||
|
|
k_lower == "connection") {
|
|
return true;
|
|
}
|
|
if (skip_content_type && k_lower == "content-type") {
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
} // namespace
|
|
|
|
void ensure_curl_initialized() {
|
|
std::call_once(g_curl_init_flag, []() {
|
|
curl_global_init(CURL_GLOBAL_ALL);
|
|
});
|
|
}
|
|
|
|
bool check_auth(const std::vector<std::string> &expected_tokens,
|
|
const HttpRequest &req) {
|
|
if (expected_tokens.empty()) {
|
|
return true;
|
|
}
|
|
std::string auth_header = req.headers.get("Authorization");
|
|
constexpr std::string_view kBearerPrefix = "Bearer ";
|
|
if (auth_header.size() < kBearerPrefix.size() ||
|
|
std::string_view(auth_header).substr(0, kBearerPrefix.size()) !=
|
|
kBearerPrefix) {
|
|
return false;
|
|
}
|
|
std::string_view token =
|
|
std::string_view(auth_header).substr(kBearerPrefix.size());
|
|
for (const auto &expected : expected_tokens) {
|
|
if (token == expected) {
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}
|
|
|
|
void send_unauthorized(ResponseWriter &w) {
|
|
send_json_error(w, 401, "Incorrect API key provided.",
|
|
"invalid_request_error", true, "invalid_api_key");
|
|
}
|
|
|
|
void handle_models(ConfigManager &cm,
|
|
const std::vector<std::string> &expected_tokens,
|
|
const HttpRequest &req,
|
|
ResponseWriter &w) {
|
|
if (!check_auth(expected_tokens, req)) {
|
|
send_unauthorized(w);
|
|
return;
|
|
}
|
|
|
|
if (req.method != "GET") {
|
|
send_json_error(w, 405, "Method not allowed", "invalid_request_error");
|
|
return;
|
|
}
|
|
|
|
auto models = cm.get_unique_models();
|
|
Json data = Json::array();
|
|
for (const auto &m : models) {
|
|
data.push_back(Json::object({
|
|
{"id", m},
|
|
{"object", "model"},
|
|
{"created", 1686935002},
|
|
{"owned_by", "dynagate"},
|
|
}));
|
|
}
|
|
|
|
Json resp = Json::object({
|
|
{"object", "list"},
|
|
{"data", std::move(data)},
|
|
});
|
|
|
|
std::string body = resp.dump() + "\n";
|
|
w.header().set("Content-Type", "application/json");
|
|
w.write_header(200);
|
|
w.write(body);
|
|
w.flush();
|
|
}
|
|
|
|
std::vector<ModelConfig> get_trial_configs(
|
|
const std::vector<ModelConfig> &configs,
|
|
const std::vector<std::string> &unique_models,
|
|
std::string_view requested_model) {
|
|
if (configs.empty()) {
|
|
return {};
|
|
}
|
|
|
|
int req_idx = -1;
|
|
for (size_t i = 0; i < unique_models.size(); ++i) {
|
|
if (unique_models[i] == requested_model) {
|
|
req_idx = static_cast<int>(i);
|
|
break;
|
|
}
|
|
}
|
|
|
|
std::vector<std::string> ordered_models;
|
|
ordered_models.reserve(unique_models.size());
|
|
if (req_idx != -1) {
|
|
for (size_t i = static_cast<size_t>(req_idx); i < unique_models.size();
|
|
++i) {
|
|
ordered_models.push_back(unique_models[i]);
|
|
}
|
|
for (size_t i = 0; i < static_cast<size_t>(req_idx); ++i) {
|
|
ordered_models.push_back(unique_models[i]);
|
|
}
|
|
} else {
|
|
ordered_models = unique_models;
|
|
}
|
|
|
|
std::vector<ModelConfig> trial_configs;
|
|
trial_configs.reserve(configs.size());
|
|
for (const auto &model_name : ordered_models) {
|
|
for (const auto &cfg : configs) {
|
|
if (cfg.model == model_name) {
|
|
trial_configs.push_back(cfg);
|
|
}
|
|
}
|
|
}
|
|
|
|
return trial_configs;
|
|
}
|
|
|
|
std::string build_url(std::string_view endpoint, std::string_view req_path) {
|
|
if (!endpoint.empty() && endpoint.back() == '/') {
|
|
endpoint.remove_suffix(1);
|
|
}
|
|
if (!req_path.empty() && req_path.front() == '/') {
|
|
req_path.remove_prefix(1);
|
|
}
|
|
|
|
if (endpoint.size() >= 3 &&
|
|
endpoint.substr(endpoint.size() - 3) == "/v1") {
|
|
if (req_path.size() >= 3 && req_path.substr(0, 3) == "v1/") {
|
|
req_path.remove_prefix(3);
|
|
}
|
|
}
|
|
|
|
return std::string(endpoint) + "/" + std::string(req_path);
|
|
}
|
|
|
|
int64_t fibonacci(int n) {
|
|
if (n <= 0) {
|
|
return 0;
|
|
}
|
|
if (n == 1 || n == 2) {
|
|
return 1;
|
|
}
|
|
int64_t a = 1;
|
|
int64_t b = 1;
|
|
for (int i = 3; i <= n; ++i) {
|
|
int64_t next = a + b;
|
|
a = b;
|
|
b = next;
|
|
}
|
|
return b;
|
|
}
|
|
|
|
bool is_media_content(const Json &m) {
|
|
if (!m.is_object()) {
|
|
return false;
|
|
}
|
|
if (m["type"].is_string()) {
|
|
std::string t_lower = str_to_lower(m["type"].sval);
|
|
if (t_lower == "image_url" || t_lower == "image" ||
|
|
t_lower == "input_audio" || t_lower == "audio" ||
|
|
t_lower == "file" || t_lower == "document" || t_lower == "video") {
|
|
return true;
|
|
}
|
|
}
|
|
for (const auto &kv : m.oval) {
|
|
std::string k_lower = str_to_lower(kv.first);
|
|
if (k_lower == "image_url" || k_lower == "input_audio" ||
|
|
k_lower == "inline_data" || k_lower == "file_data") {
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}
|
|
|
|
Json normalize_content(const Json &content_any) {
|
|
if (content_any.is_null()) {
|
|
return nullptr;
|
|
}
|
|
|
|
if (content_any.is_string()) {
|
|
return content_any;
|
|
}
|
|
|
|
if (content_any.is_array()) {
|
|
if (content_any.empty()) {
|
|
return "";
|
|
}
|
|
bool has_media = false;
|
|
for (const auto &item : content_any.aval) {
|
|
if (item.is_object() && is_media_content(item)) {
|
|
has_media = true;
|
|
break;
|
|
}
|
|
}
|
|
|
|
if (has_media) {
|
|
Json norm_slice = Json::array();
|
|
for (const auto &item : content_any.aval) {
|
|
if (item.is_object()) {
|
|
if (item["type"].is_string() &&
|
|
str_to_lower(item["type"].sval) == "text") {
|
|
std::string text_val = item["text"].str();
|
|
norm_slice.push_back(Json::object({
|
|
{"type", "text"},
|
|
{"text", text_val},
|
|
}));
|
|
} else {
|
|
norm_slice.push_back(item);
|
|
}
|
|
} else {
|
|
norm_slice.push_back(item);
|
|
}
|
|
}
|
|
return norm_slice;
|
|
}
|
|
|
|
std::vector<std::string> text_parts;
|
|
for (const auto &item : content_any.aval) {
|
|
if (item.is_string()) {
|
|
text_parts.push_back(item.sval);
|
|
} else if (item.is_object()) {
|
|
if (item["text"].is_string()) {
|
|
text_parts.push_back(item["text"].sval);
|
|
} else if (item["content"].is_string()) {
|
|
text_parts.push_back(item["content"].sval);
|
|
}
|
|
}
|
|
}
|
|
|
|
std::string sb;
|
|
for (const auto &part : text_parts) {
|
|
if (part.empty()) {
|
|
continue;
|
|
}
|
|
if (!sb.empty()) {
|
|
char last_char = sb.back();
|
|
char first_char = part.front();
|
|
if (last_char != '\n' && last_char != ' ' && first_char != '\n' &&
|
|
first_char != ' ') {
|
|
sb.push_back('\n');
|
|
}
|
|
}
|
|
sb.append(part);
|
|
}
|
|
return sb;
|
|
}
|
|
|
|
if (content_any.is_object()) {
|
|
if (is_media_content(content_any)) {
|
|
return Json::array({content_any});
|
|
}
|
|
if (content_any["text"].is_string()) {
|
|
return content_any["text"];
|
|
}
|
|
if (content_any["content"].is_string()) {
|
|
return content_any["content"];
|
|
}
|
|
return content_any;
|
|
}
|
|
|
|
return content_any;
|
|
}
|
|
|
|
void autodetect_and_normalize_messages(Json &body_map) {
|
|
if (!body_map.is_object()) {
|
|
return;
|
|
}
|
|
|
|
Json system_msg = nullptr;
|
|
if (body_map.contains("system") && !body_map["system"].is_null()) {
|
|
Json sys_text = normalize_content(body_map["system"]);
|
|
if (sys_text.is_string() && !sys_text.sval.empty()) {
|
|
system_msg = Json::object({
|
|
{"role", "system"},
|
|
{"content", sys_text},
|
|
});
|
|
} else if (sys_text.is_array() && !sys_text.empty()) {
|
|
system_msg = Json::object({
|
|
{"role", "system"},
|
|
{"content", sys_text},
|
|
});
|
|
}
|
|
body_map.erase("system");
|
|
}
|
|
|
|
Json raw_messages = nullptr;
|
|
std::string source_key;
|
|
|
|
if (body_map.contains("messages") && !body_map["messages"].is_null()) {
|
|
raw_messages = body_map["messages"];
|
|
source_key = "messages";
|
|
} else if (body_map.contains("prompt") && !body_map["prompt"].is_null()) {
|
|
raw_messages = body_map["prompt"];
|
|
source_key = "prompt";
|
|
body_map.erase("prompt");
|
|
} else if (body_map.contains("contents") && !body_map["contents"].is_null()) {
|
|
raw_messages = body_map["contents"];
|
|
source_key = "contents";
|
|
body_map.erase("contents");
|
|
} else if (body_map.contains("input") && !body_map["input"].is_null()) {
|
|
raw_messages = body_map["input"];
|
|
source_key = "input";
|
|
body_map.erase("input");
|
|
}
|
|
|
|
if (raw_messages.is_null() && system_msg.is_null()) {
|
|
return;
|
|
}
|
|
|
|
std::vector<Json> msg_list;
|
|
if (raw_messages.is_array()) {
|
|
for (const auto &elem : raw_messages.aval) {
|
|
if (elem.is_object()) {
|
|
msg_list.push_back(elem);
|
|
} else if (elem.is_string()) {
|
|
msg_list.push_back(Json::object({
|
|
{"role", "user"},
|
|
{"content", elem},
|
|
}));
|
|
}
|
|
}
|
|
} else if (raw_messages.is_object()) {
|
|
msg_list.push_back(raw_messages);
|
|
} else if (raw_messages.is_string()) {
|
|
if (!raw_messages.sval.empty()) {
|
|
msg_list.push_back(Json::object({
|
|
{"role", "user"},
|
|
{"content", raw_messages},
|
|
}));
|
|
}
|
|
}
|
|
|
|
Json normalized_list = Json::array();
|
|
bool has_system_in_list = false;
|
|
|
|
for (const auto &msg : msg_list) {
|
|
std::string role = msg["role"].str();
|
|
if (role.empty()) {
|
|
role = "user";
|
|
}
|
|
if (role == "system") {
|
|
has_system_in_list = true;
|
|
}
|
|
|
|
Json norm_msg = msg;
|
|
norm_msg["role"] = role;
|
|
|
|
if (msg.contains("content")) {
|
|
norm_msg["content"] = normalize_content(msg["content"]);
|
|
} else if (msg.contains("parts")) {
|
|
norm_msg["content"] = normalize_content(msg["parts"]);
|
|
norm_msg.erase("parts");
|
|
}
|
|
|
|
normalized_list.push_back(std::move(norm_msg));
|
|
}
|
|
|
|
if (!system_msg.is_null()) {
|
|
if (!has_system_in_list) {
|
|
normalized_list.aval.insert(normalized_list.aval.begin(), system_msg);
|
|
} else if (!normalized_list.empty()) {
|
|
auto &first_msg = normalized_list.at(0);
|
|
if (first_msg.is_object() &&
|
|
first_msg.contains("role") &&
|
|
first_msg["role"].str() == "system") {
|
|
Json existing_sys = normalize_content(
|
|
first_msg.contains("content") ? first_msg["content"] : Json(nullptr));
|
|
if (system_msg["content"].is_string()) {
|
|
std::string sys_str = system_msg["content"].sval;
|
|
if (existing_sys.is_string() && !existing_sys.sval.empty()) {
|
|
first_msg["content"] = sys_str + "\n" + existing_sys.sval;
|
|
} else {
|
|
first_msg["content"] = sys_str;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if (!normalized_list.empty() || !source_key.empty()) {
|
|
body_map["messages"] = std::move(normalized_list);
|
|
}
|
|
}
|
|
|
|
Json convert_anthropic_content_to_openai(const Json &cnt_any) {
|
|
if (cnt_any.is_array()) {
|
|
Json new_slice = Json::array();
|
|
for (const auto &item : cnt_any.aval) {
|
|
if (item.is_object() &&
|
|
item["type"].str() == "image" &&
|
|
item["source"].is_object()) {
|
|
const auto &src = item["source"];
|
|
std::string media_type = src["media_type"].str();
|
|
if (media_type.empty()) {
|
|
media_type = "image/png";
|
|
}
|
|
std::string b64_data = src["data"].str();
|
|
std::string data_url = "data:" + media_type + ";base64," + b64_data;
|
|
new_slice.push_back(Json::object({
|
|
{"type", "image_url"},
|
|
{"image_url", Json::object({{"url", data_url}})},
|
|
}));
|
|
continue;
|
|
}
|
|
new_slice.push_back(item);
|
|
}
|
|
return normalize_content(new_slice);
|
|
}
|
|
return normalize_content(cnt_any);
|
|
}
|
|
|
|
Json convert_anthropic_to_openai(const Json &body_map) {
|
|
Json openai_body = Json::object();
|
|
|
|
if (body_map.contains("model")) {
|
|
openai_body["model"] = body_map["model"];
|
|
}
|
|
if (body_map["stream"].is_bool()) {
|
|
openai_body["stream"] = body_map["stream"];
|
|
}
|
|
if (body_map.contains("temperature")) {
|
|
openai_body["temperature"] = body_map["temperature"];
|
|
}
|
|
if (body_map.contains("top_p")) {
|
|
openai_body["top_p"] = body_map["top_p"];
|
|
}
|
|
if (body_map.contains("max_tokens")) {
|
|
openai_body["max_tokens"] = body_map["max_tokens"];
|
|
}
|
|
if (body_map.contains("stop_sequences")) {
|
|
openai_body["stop"] = body_map["stop_sequences"];
|
|
}
|
|
|
|
Json openai_messages = Json::array();
|
|
|
|
if (body_map.contains("system") && !body_map["system"].is_null()) {
|
|
Json sys_content = normalize_content(body_map["system"]);
|
|
if (!sys_content.is_null()) {
|
|
openai_messages.push_back(Json::object({
|
|
{"role", "system"},
|
|
{"content", sys_content},
|
|
}));
|
|
}
|
|
}
|
|
|
|
if (body_map["messages"].is_array()) {
|
|
for (const auto &item : body_map["messages"].aval) {
|
|
if (item.is_object()) {
|
|
std::string role = item["role"].str();
|
|
if (role.empty()) {
|
|
role = "user";
|
|
}
|
|
Json cnt = item.contains("content") ? item["content"] : Json(nullptr);
|
|
Json norm_cnt = convert_anthropic_content_to_openai(cnt);
|
|
openai_messages.push_back(Json::object({
|
|
{"role", role},
|
|
{"content", norm_cnt},
|
|
}));
|
|
}
|
|
}
|
|
}
|
|
|
|
openai_body["messages"] = std::move(openai_messages);
|
|
autodetect_and_normalize_messages(openai_body);
|
|
return openai_body;
|
|
}
|
|
|
|
Json convert_openai_to_anthropic_response(const Json &openai_resp,
|
|
const std::string &requested_model) {
|
|
std::string id = openai_resp["id"].str();
|
|
if (id.empty()) {
|
|
id = "msg_" + std::to_string(unix_nano_now());
|
|
} else if (id.rfind("msg_", 0) != 0) {
|
|
id = "msg_" + id;
|
|
}
|
|
|
|
std::string model = openai_resp["model"].str();
|
|
if (model.empty()) {
|
|
model = requested_model;
|
|
}
|
|
|
|
std::string text_content;
|
|
std::string finish_reason;
|
|
|
|
if (openai_resp["choices"].is_array() && !openai_resp["choices"].empty()) {
|
|
const auto &choice = openai_resp["choices"].at(0);
|
|
if (choice.is_object()) {
|
|
finish_reason = choice["finish_reason"].str();
|
|
if (choice["message"].is_object()) {
|
|
const auto &msg = choice["message"];
|
|
if (msg["content"].is_string()) {
|
|
text_content = msg["content"].sval;
|
|
} else if (msg.contains("content")) {
|
|
Json cnt_norm = normalize_content(msg["content"]);
|
|
if (cnt_norm.is_string()) {
|
|
text_content = cnt_norm.sval;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
std::string stop_reason = "end_turn";
|
|
if (finish_reason == "length") {
|
|
stop_reason = "max_tokens";
|
|
} else if (finish_reason == "tool_calls" ||
|
|
finish_reason == "function_call") {
|
|
stop_reason = "tool_use";
|
|
} else if (finish_reason == "stop") {
|
|
stop_reason = "end_turn";
|
|
}
|
|
|
|
int input_tokens = 0;
|
|
int output_tokens = 0;
|
|
if (openai_resp["usage"].is_object()) {
|
|
const auto &usage = openai_resp["usage"];
|
|
input_tokens = usage["prompt_tokens"].as_int();
|
|
output_tokens = usage["completion_tokens"].as_int();
|
|
}
|
|
|
|
return Json::object({
|
|
{"id", id},
|
|
{"type", "message"},
|
|
{"role", "assistant"},
|
|
{"model", model},
|
|
{"content", Json::array({
|
|
Json::object({
|
|
{"type", "text"},
|
|
{"text", text_content},
|
|
}),
|
|
})},
|
|
{"stop_reason", stop_reason},
|
|
{"stop_sequence", nullptr},
|
|
{"usage", Json::object({
|
|
{"input_tokens", input_tokens},
|
|
{"output_tokens", output_tokens},
|
|
})},
|
|
});
|
|
}
|
|
|
|
Json convert_responses_to_openai(const Json &body_map) {
|
|
Json openai_body = Json::object();
|
|
|
|
if (body_map.contains("model")) {
|
|
openai_body["model"] = body_map["model"];
|
|
}
|
|
if (body_map["stream"].is_bool()) {
|
|
openai_body["stream"] = body_map["stream"];
|
|
}
|
|
if (body_map.contains("temperature")) {
|
|
openai_body["temperature"] = body_map["temperature"];
|
|
}
|
|
if (body_map.contains("top_p")) {
|
|
openai_body["top_p"] = body_map["top_p"];
|
|
}
|
|
if (body_map.contains("max_output_tokens")) {
|
|
openai_body["max_tokens"] = body_map["max_output_tokens"];
|
|
} else if (body_map.contains("max_tokens")) {
|
|
openai_body["max_tokens"] = body_map["max_tokens"];
|
|
}
|
|
|
|
Json openai_messages = Json::array();
|
|
|
|
if (body_map.contains("instructions") &&
|
|
!body_map["instructions"].is_null()) {
|
|
Json sys_content = normalize_content(body_map["instructions"]);
|
|
if (!sys_content.is_null()) {
|
|
openai_messages.push_back(Json::object({
|
|
{"role", "system"},
|
|
{"content", sys_content},
|
|
}));
|
|
}
|
|
}
|
|
|
|
if (body_map.contains("input") && !body_map["input"].is_null()) {
|
|
const auto &inp = body_map["input"];
|
|
if (inp.is_string()) {
|
|
openai_messages.push_back(Json::object({
|
|
{"role", "user"},
|
|
{"content", inp},
|
|
}));
|
|
} else if (inp.is_array()) {
|
|
for (const auto &elem : inp.aval) {
|
|
if (elem.is_string()) {
|
|
openai_messages.push_back(Json::object({
|
|
{"role", "user"},
|
|
{"content", elem},
|
|
}));
|
|
} else if (elem.is_object()) {
|
|
std::string role = elem["role"].str();
|
|
if (role.empty()) {
|
|
role = "user";
|
|
}
|
|
Json cnt =
|
|
(elem.contains("content") && !elem["content"].is_null())
|
|
? elem["content"]
|
|
: Json(nullptr);
|
|
if (cnt.is_null() && elem["text"].is_string()) {
|
|
cnt = elem["text"];
|
|
}
|
|
openai_messages.push_back(Json::object({
|
|
{"role", role},
|
|
{"content", normalize_content(cnt)},
|
|
}));
|
|
}
|
|
}
|
|
} else if (inp.is_object()) {
|
|
std::string role = inp["role"].str();
|
|
if (role.empty()) {
|
|
role = "user";
|
|
}
|
|
Json cnt =
|
|
(inp.contains("content") && !inp["content"].is_null())
|
|
? inp["content"]
|
|
: Json(nullptr);
|
|
if (cnt.is_null() && inp["text"].is_string()) {
|
|
cnt = inp["text"];
|
|
}
|
|
openai_messages.push_back(Json::object({
|
|
{"role", role},
|
|
{"content", normalize_content(cnt)},
|
|
}));
|
|
}
|
|
} else if (body_map["messages"].is_array()) {
|
|
for (const auto &item : body_map["messages"].aval) {
|
|
if (item.is_object()) {
|
|
std::string role = item["role"].str();
|
|
if (role.empty()) {
|
|
role = "user";
|
|
}
|
|
Json cnt = item.contains("content") ? item["content"] : Json(nullptr);
|
|
openai_messages.push_back(Json::object({
|
|
{"role", role},
|
|
{"content", normalize_content(cnt)},
|
|
}));
|
|
}
|
|
}
|
|
}
|
|
|
|
openai_body["messages"] = std::move(openai_messages);
|
|
autodetect_and_normalize_messages(openai_body);
|
|
return openai_body;
|
|
}
|
|
|
|
Json convert_openai_to_responses_response(const Json &openai_resp,
|
|
const std::string &requested_model) {
|
|
std::string id = openai_resp["id"].str();
|
|
if (id.empty()) {
|
|
id = "resp_" + std::to_string(unix_nano_now());
|
|
} else if (id.rfind("resp_", 0) != 0) {
|
|
id = "resp_" + id;
|
|
}
|
|
|
|
std::string model = openai_resp["model"].str();
|
|
if (model.empty()) {
|
|
model = requested_model;
|
|
}
|
|
|
|
std::string text_content;
|
|
if (openai_resp["choices"].is_array() && !openai_resp["choices"].empty()) {
|
|
const auto &choice = openai_resp["choices"].at(0);
|
|
if (choice.is_object() && choice["message"].is_object()) {
|
|
const auto &msg = choice["message"];
|
|
if (msg["content"].is_string()) {
|
|
text_content = msg["content"].sval;
|
|
} else if (msg.contains("content")) {
|
|
Json cnt_norm = normalize_content(msg["content"]);
|
|
if (cnt_norm.is_string()) {
|
|
text_content = cnt_norm.sval;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
int64_t created = openai_resp["created"].is_number()
|
|
? openai_resp["created"].as_int64()
|
|
: unix_sec_now();
|
|
|
|
int prompt_tokens = 0;
|
|
int completion_tokens = 0;
|
|
int total_tokens = 0;
|
|
if (openai_resp["usage"].is_object()) {
|
|
const auto &usage = openai_resp["usage"];
|
|
prompt_tokens = usage["prompt_tokens"].as_int();
|
|
completion_tokens = usage["completion_tokens"].as_int();
|
|
total_tokens = usage["total_tokens"].as_int();
|
|
}
|
|
|
|
std::string msg_id = "msg_" + std::to_string(unix_nano_now());
|
|
|
|
return Json::object({
|
|
{"id", id},
|
|
{"object", "response"},
|
|
{"created_at", created},
|
|
{"status", "completed"},
|
|
{"model", model},
|
|
{"output", Json::array({
|
|
Json::object({
|
|
{"id", msg_id},
|
|
{"type", "message"},
|
|
{"status", "completed"},
|
|
{"role", "assistant"},
|
|
{"content", Json::array({
|
|
Json::object({
|
|
{"type", "output_text"},
|
|
{"text", text_content},
|
|
{"annotations", Json::array()},
|
|
{"logprobs", Json::array()},
|
|
}),
|
|
})},
|
|
}),
|
|
})},
|
|
{"usage", Json::object({
|
|
{"prompt_tokens", prompt_tokens},
|
|
{"completion_tokens", completion_tokens},
|
|
{"total_tokens", total_tokens},
|
|
})},
|
|
});
|
|
}
|
|
|
|
void handle_chat_completions(
|
|
ConfigManager &cm,
|
|
const std::vector<std::string> &expected_tokens,
|
|
std::chrono::milliseconds retry_base_delay,
|
|
const HttpRequest &req,
|
|
ResponseWriter &w) {
|
|
ensure_curl_initialized();
|
|
|
|
if (!check_auth(expected_tokens, req)) {
|
|
send_unauthorized(w);
|
|
return;
|
|
}
|
|
|
|
if (req.method != "POST") {
|
|
send_json_error(w, 405, "Method not allowed", "invalid_request_error");
|
|
return;
|
|
}
|
|
|
|
Json body_map;
|
|
if (!Json::parse(req.body, body_map) || !body_map.is_object()) {
|
|
send_json_error(w, 400, "Invalid JSON in request body",
|
|
"invalid_request_error");
|
|
return;
|
|
}
|
|
|
|
autodetect_and_normalize_messages(body_map);
|
|
|
|
std::string requested_model;
|
|
if (body_map.contains("model") && body_map["model"].is_string()) {
|
|
requested_model = body_map["model"].sval;
|
|
}
|
|
|
|
auto configs = cm.get_configs();
|
|
auto unique_models = cm.get_unique_models();
|
|
|
|
if (configs.empty()) {
|
|
send_json_error(w, 503, "No model configurations loaded", "gateway_error");
|
|
return;
|
|
}
|
|
|
|
auto trial_configs =
|
|
get_trial_configs(configs, unique_models, requested_model);
|
|
if (trial_configs.empty()) {
|
|
send_json_error(w, 503, "No valid trial configuration candidates",
|
|
"gateway_error");
|
|
return;
|
|
}
|
|
|
|
bool is_stream = false;
|
|
if (body_map.contains("stream") && body_map["stream"].is_bool()) {
|
|
is_stream = body_map["stream"].as_bool();
|
|
}
|
|
|
|
log_msg("Received completion request for model \"%s\" (stream=%s). Found %zu config trials.",
|
|
requested_model.c_str(), is_stream ? "true" : "false",
|
|
trial_configs.size());
|
|
|
|
for (size_t i = 0; i < trial_configs.size(); ++i) {
|
|
const auto &trial = trial_configs[i];
|
|
if (i > 0 && retry_base_delay.count() > 0) {
|
|
auto delay = std::chrono::milliseconds(
|
|
fibonacci(static_cast<int>(i)) * retry_base_delay.count());
|
|
log_msg("Trial %zu/%zu: Fibonacci backoff delay of %lldms before retry...",
|
|
i + 1, trial_configs.size(),
|
|
static_cast<long long>(delay.count()));
|
|
if (!wait_backoff_or_cancelled(delay, req)) {
|
|
log_msg("Request context cancelled during retry delay before trial %zu",
|
|
i + 1);
|
|
send_json_error(w, 499, "Client closed request", "gateway_error");
|
|
return;
|
|
}
|
|
}
|
|
|
|
log_msg("Trial %zu/%zu: model=%s endpoint=%s key_len=%zu",
|
|
i + 1, trial_configs.size(), trial.model.c_str(),
|
|
trial.endpoint.c_str(), trial.key.size());
|
|
|
|
body_map["model"] = trial.model;
|
|
std::string modified_body = body_map.dump();
|
|
std::string target_url = build_url(trial.endpoint, req.path);
|
|
|
|
ScopedCurl curl;
|
|
if (!curl.handle) {
|
|
log_msg("Trial %zu: Failed to initialize curl handle", i + 1);
|
|
continue;
|
|
}
|
|
|
|
ScopedCurlSlist headers_slist;
|
|
build_outgoing_headers(req, trial, i + 1, headers_slist);
|
|
|
|
UpstreamTransferCtx ctx;
|
|
ctx.is_stream = is_stream;
|
|
if (is_stream) {
|
|
ctx.on_stream_start = [&]() {
|
|
log_msg("Trial %zu: Connection established with status %d. Proxying response.",
|
|
i + 1, ctx.status_code);
|
|
for (const auto &kv : ctx.resp_headers.items) {
|
|
if (should_skip_upstream_resp_header(kv.first, false)) {
|
|
continue;
|
|
}
|
|
w.header().add(kv.first, kv.second);
|
|
}
|
|
w.write_header(ctx.status_code);
|
|
w.flush();
|
|
};
|
|
ctx.on_stream_data = [&](std::string_view chunk) -> bool {
|
|
if (!w.write(chunk)) {
|
|
log_msg("Trial %zu: Client disconnected or write error", i + 1);
|
|
return false;
|
|
}
|
|
w.flush();
|
|
return true;
|
|
};
|
|
}
|
|
|
|
curl_easy_setopt(curl.handle, CURLOPT_NOSIGNAL, 1L);
|
|
curl_easy_setopt(curl.handle, CURLOPT_URL, target_url.c_str());
|
|
curl_easy_setopt(curl.handle, CURLOPT_POST, 1L);
|
|
curl_easy_setopt(curl.handle, CURLOPT_POSTFIELDS, modified_body.data());
|
|
curl_easy_setopt(curl.handle, CURLOPT_POSTFIELDSIZE,
|
|
static_cast<long>(modified_body.size()));
|
|
curl_easy_setopt(curl.handle, CURLOPT_HTTPHEADER, headers_slist.list);
|
|
curl_easy_setopt(curl.handle, CURLOPT_HEADERFUNCTION, curl_header_cb);
|
|
curl_easy_setopt(curl.handle, CURLOPT_HEADERDATA, &ctx);
|
|
curl_easy_setopt(curl.handle, CURLOPT_WRITEFUNCTION, curl_write_cb);
|
|
curl_easy_setopt(curl.handle, CURLOPT_WRITEDATA, &ctx);
|
|
|
|
CURLcode res = curl_easy_perform(curl.handle);
|
|
if (res != CURLE_OK) {
|
|
if (ctx.stream_started) {
|
|
return;
|
|
}
|
|
log_msg("Trial %zu: Request to %s failed: %s",
|
|
i + 1, target_url.c_str(), curl_easy_strerror(res));
|
|
continue;
|
|
}
|
|
|
|
if (ctx.status_code < 200 || ctx.status_code >= 300) {
|
|
std::string trimmed_err = str_trim(ctx.err_body);
|
|
log_msg("Trial %zu: Request to %s returned error status %d: %s",
|
|
i + 1, target_url.c_str(), ctx.status_code, trimmed_err.c_str());
|
|
continue;
|
|
}
|
|
|
|
if (is_stream) {
|
|
return;
|
|
}
|
|
|
|
log_msg("Trial %zu: Connection established with status %d. Proxying response.",
|
|
i + 1, ctx.status_code);
|
|
for (const auto &kv : ctx.resp_headers.items) {
|
|
if (should_skip_upstream_resp_header(kv.first, false)) {
|
|
continue;
|
|
}
|
|
w.header().add(kv.first, kv.second);
|
|
}
|
|
w.write_header(ctx.status_code);
|
|
w.write(ctx.resp_body);
|
|
w.flush();
|
|
return;
|
|
}
|
|
|
|
log_msg("All trials failed. Returning Bad Gateway.");
|
|
send_json_error(w, 502, "All configured models and keys failed to respond.",
|
|
"gateway_error", true, "all_endpoints_failed");
|
|
}
|
|
|
|
void handle_image_generations(
|
|
ConfigManager &cm,
|
|
const std::vector<std::string> &expected_tokens,
|
|
std::chrono::milliseconds retry_base_delay,
|
|
const HttpRequest &req,
|
|
ResponseWriter &w) {
|
|
ensure_curl_initialized();
|
|
|
|
if (!check_auth(expected_tokens, req)) {
|
|
send_unauthorized(w);
|
|
return;
|
|
}
|
|
|
|
if (req.method != "POST") {
|
|
send_json_error(w, 405, "Method not allowed", "invalid_request_error");
|
|
return;
|
|
}
|
|
|
|
Json body_map;
|
|
if (!Json::parse(req.body, body_map) || !body_map.is_object()) {
|
|
send_json_error(w, 400, "Invalid JSON in request body",
|
|
"invalid_request_error");
|
|
return;
|
|
}
|
|
|
|
if (body_map.contains("prompt") && !body_map["prompt"].is_null()) {
|
|
body_map["prompt"] = normalize_content(body_map["prompt"]);
|
|
}
|
|
|
|
std::string requested_model;
|
|
if (body_map.contains("model") && body_map["model"].is_string()) {
|
|
requested_model = body_map["model"].sval;
|
|
}
|
|
|
|
auto configs = cm.get_configs();
|
|
auto unique_models = cm.get_unique_models();
|
|
|
|
if (configs.empty()) {
|
|
send_json_error(w, 503, "No model configurations loaded", "gateway_error");
|
|
return;
|
|
}
|
|
|
|
auto trial_configs =
|
|
get_trial_configs(configs, unique_models, requested_model);
|
|
if (trial_configs.empty()) {
|
|
send_json_error(w, 503, "No valid trial configuration candidates",
|
|
"gateway_error");
|
|
return;
|
|
}
|
|
|
|
log_msg("Received image generation request for model \"%s\". Found %zu config trials.",
|
|
requested_model.c_str(), trial_configs.size());
|
|
|
|
for (size_t i = 0; i < trial_configs.size(); ++i) {
|
|
const auto &trial = trial_configs[i];
|
|
if (i > 0 && retry_base_delay.count() > 0) {
|
|
auto delay = std::chrono::milliseconds(
|
|
fibonacci(static_cast<int>(i)) * retry_base_delay.count());
|
|
log_msg("Trial %zu/%zu: Fibonacci backoff delay of %lldms before retry...",
|
|
i + 1, trial_configs.size(),
|
|
static_cast<long long>(delay.count()));
|
|
if (!wait_backoff_or_cancelled(delay, req)) {
|
|
log_msg("Request context cancelled during retry delay before trial %zu",
|
|
i + 1);
|
|
send_json_error(w, 499, "Client closed request", "gateway_error");
|
|
return;
|
|
}
|
|
}
|
|
|
|
log_msg("Trial %zu/%zu: model=%s endpoint=%s key_len=%zu",
|
|
i + 1, trial_configs.size(), trial.model.c_str(),
|
|
trial.endpoint.c_str(), trial.key.size());
|
|
|
|
body_map["model"] = trial.model;
|
|
std::string modified_body = body_map.dump();
|
|
std::string target_url = build_url(trial.endpoint, req.path);
|
|
|
|
ScopedCurl curl;
|
|
if (!curl.handle) {
|
|
log_msg("Trial %zu: Failed to initialize curl handle", i + 1);
|
|
continue;
|
|
}
|
|
|
|
ScopedCurlSlist headers_slist;
|
|
build_outgoing_headers(req, trial, i + 1, headers_slist);
|
|
|
|
UpstreamTransferCtx ctx;
|
|
ctx.is_stream = false;
|
|
|
|
curl_easy_setopt(curl.handle, CURLOPT_NOSIGNAL, 1L);
|
|
curl_easy_setopt(curl.handle, CURLOPT_URL, target_url.c_str());
|
|
curl_easy_setopt(curl.handle, CURLOPT_POST, 1L);
|
|
curl_easy_setopt(curl.handle, CURLOPT_POSTFIELDS, modified_body.data());
|
|
curl_easy_setopt(curl.handle, CURLOPT_POSTFIELDSIZE,
|
|
static_cast<long>(modified_body.size()));
|
|
curl_easy_setopt(curl.handle, CURLOPT_HTTPHEADER, headers_slist.list);
|
|
curl_easy_setopt(curl.handle, CURLOPT_HEADERFUNCTION, curl_header_cb);
|
|
curl_easy_setopt(curl.handle, CURLOPT_HEADERDATA, &ctx);
|
|
curl_easy_setopt(curl.handle, CURLOPT_WRITEFUNCTION, curl_write_cb);
|
|
curl_easy_setopt(curl.handle, CURLOPT_WRITEDATA, &ctx);
|
|
|
|
CURLcode res = curl_easy_perform(curl.handle);
|
|
if (res != CURLE_OK) {
|
|
log_msg("Trial %zu: Request to %s failed: %s",
|
|
i + 1, target_url.c_str(), curl_easy_strerror(res));
|
|
continue;
|
|
}
|
|
|
|
if (ctx.status_code < 200 || ctx.status_code >= 300) {
|
|
std::string trimmed_err = str_trim(ctx.err_body);
|
|
log_msg("Trial %zu: Request to %s returned error status %d: %s",
|
|
i + 1, target_url.c_str(), ctx.status_code, trimmed_err.c_str());
|
|
continue;
|
|
}
|
|
|
|
log_msg("Trial %zu: Connection established with status %d. Proxying response.",
|
|
i + 1, ctx.status_code);
|
|
for (const auto &kv : ctx.resp_headers.items) {
|
|
if (should_skip_upstream_resp_header(kv.first, false)) {
|
|
continue;
|
|
}
|
|
w.header().add(kv.first, kv.second);
|
|
}
|
|
w.write_header(ctx.status_code);
|
|
w.write(ctx.resp_body);
|
|
w.flush();
|
|
return;
|
|
}
|
|
|
|
log_msg("All trials failed. Returning Bad Gateway.");
|
|
send_json_error(w, 502, "All configured models and keys failed to respond.",
|
|
"gateway_error", true, "all_endpoints_failed");
|
|
}
|
|
|
|
void handle_messages(
|
|
ConfigManager &cm,
|
|
const std::vector<std::string> &expected_tokens,
|
|
std::chrono::milliseconds retry_base_delay,
|
|
const HttpRequest &req,
|
|
ResponseWriter &w) {
|
|
ensure_curl_initialized();
|
|
|
|
if (!check_auth(expected_tokens, req)) {
|
|
send_unauthorized(w);
|
|
return;
|
|
}
|
|
|
|
if (req.method != "POST") {
|
|
send_json_error(w, 405, "Method not allowed", "invalid_request_error");
|
|
return;
|
|
}
|
|
|
|
Json body_map;
|
|
if (!Json::parse(req.body, body_map) || !body_map.is_object()) {
|
|
send_json_error(w, 400, "Invalid JSON in request body",
|
|
"invalid_request_error");
|
|
return;
|
|
}
|
|
|
|
Json openai_body_map = convert_anthropic_to_openai(body_map);
|
|
|
|
std::string requested_model;
|
|
if (openai_body_map.contains("model") &&
|
|
openai_body_map["model"].is_string()) {
|
|
requested_model = openai_body_map["model"].sval;
|
|
}
|
|
|
|
auto configs = cm.get_configs();
|
|
auto unique_models = cm.get_unique_models();
|
|
|
|
if (configs.empty()) {
|
|
send_json_error(w, 503, "No model configurations loaded", "gateway_error");
|
|
return;
|
|
}
|
|
|
|
auto trial_configs =
|
|
get_trial_configs(configs, unique_models, requested_model);
|
|
if (trial_configs.empty()) {
|
|
send_json_error(w, 503, "No valid trial configuration candidates",
|
|
"gateway_error");
|
|
return;
|
|
}
|
|
|
|
bool is_stream = false;
|
|
if (openai_body_map.contains("stream") &&
|
|
openai_body_map["stream"].is_bool()) {
|
|
is_stream = openai_body_map["stream"].as_bool();
|
|
}
|
|
|
|
log_msg("Received Anthropic messages request for model \"%s\" (stream=%s). Found %zu config trials.",
|
|
requested_model.c_str(), is_stream ? "true" : "false",
|
|
trial_configs.size());
|
|
|
|
for (size_t i = 0; i < trial_configs.size(); ++i) {
|
|
const auto &trial = trial_configs[i];
|
|
if (i > 0 && retry_base_delay.count() > 0) {
|
|
auto delay = std::chrono::milliseconds(
|
|
fibonacci(static_cast<int>(i)) * retry_base_delay.count());
|
|
log_msg("Trial %zu/%zu: Fibonacci backoff delay of %lldms before retry...",
|
|
i + 1, trial_configs.size(),
|
|
static_cast<long long>(delay.count()));
|
|
if (!wait_backoff_or_cancelled(delay, req)) {
|
|
log_msg("Request context cancelled during retry delay before trial %zu",
|
|
i + 1);
|
|
send_json_error(w, 499, "Client closed request", "gateway_error");
|
|
return;
|
|
}
|
|
}
|
|
|
|
log_msg("Trial %zu/%zu: model=%s endpoint=%s key_len=%zu",
|
|
i + 1, trial_configs.size(), trial.model.c_str(),
|
|
trial.endpoint.c_str(), trial.key.size());
|
|
|
|
openai_body_map["model"] = trial.model;
|
|
std::string modified_body = openai_body_map.dump();
|
|
std::string target_url = build_url(trial.endpoint, "/v1/chat/completions");
|
|
|
|
ScopedCurl curl;
|
|
if (!curl.handle) {
|
|
log_msg("Trial %zu: Failed to initialize curl handle", i + 1);
|
|
continue;
|
|
}
|
|
|
|
ScopedCurlSlist headers_slist;
|
|
build_outgoing_headers(req, trial, i + 1, headers_slist);
|
|
|
|
UpstreamTransferCtx ctx;
|
|
ctx.is_stream = is_stream;
|
|
|
|
std::string sse_line_buf;
|
|
std::string final_stop_reason = "end_turn";
|
|
bool sse_done = false;
|
|
|
|
auto process_anthropic_sse_line = [&](std::string_view raw_line) -> bool {
|
|
std::string line = str_trim(raw_line);
|
|
constexpr std::string_view kDataPrefix = "data: ";
|
|
if (line.size() < kDataPrefix.size() ||
|
|
std::string_view(line).substr(0, kDataPrefix.size()) != kDataPrefix) {
|
|
return true;
|
|
}
|
|
std::string data_str =
|
|
str_trim(std::string_view(line).substr(kDataPrefix.size()));
|
|
if (data_str == "[DONE]") {
|
|
sse_done = true;
|
|
return true;
|
|
}
|
|
Json chunk_mut;
|
|
if (Json::parse(data_str, chunk_mut) && chunk_mut.is_object()) {
|
|
const Json &chunk_map = chunk_mut;
|
|
if (chunk_map["choices"].is_array() && !chunk_map["choices"].empty()) {
|
|
const auto &choice = chunk_map["choices"].at(0);
|
|
if (choice.is_object()) {
|
|
if (choice["finish_reason"].is_string()) {
|
|
const std::string &fr = choice["finish_reason"].sval;
|
|
if (!fr.empty()) {
|
|
if (fr == "length") {
|
|
final_stop_reason = "max_tokens";
|
|
} else if (fr == "tool_calls" || fr == "function_call") {
|
|
final_stop_reason = "tool_use";
|
|
} else if (fr == "stop") {
|
|
final_stop_reason = "end_turn";
|
|
}
|
|
}
|
|
}
|
|
if (choice["delta"].is_object()) {
|
|
const auto &delta = choice["delta"];
|
|
if (delta["content"].is_string()) {
|
|
const std::string &content_str = delta["content"].sval;
|
|
if (!content_str.empty()) {
|
|
Json delta_obj = Json::object({
|
|
{"type", "content_block_delta"},
|
|
{"index", 0},
|
|
{"delta", Json::object({
|
|
{"type", "text_delta"},
|
|
{"text", content_str},
|
|
})},
|
|
});
|
|
std::string ev = "event: content_block_delta\ndata: " +
|
|
delta_obj.dump() + "\n\n";
|
|
if (!w.write(ev)) {
|
|
return false;
|
|
}
|
|
w.flush();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return true;
|
|
};
|
|
|
|
if (is_stream) {
|
|
ctx.on_stream_start = [&]() {
|
|
log_msg("Trial %zu: Connection established with status %d. Proxying response.",
|
|
i + 1, ctx.status_code);
|
|
w.header().set("Content-Type", "text/event-stream");
|
|
w.header().set("Cache-Control", "no-cache");
|
|
w.header().set("Connection", "keep-alive");
|
|
w.write_header(200);
|
|
w.flush();
|
|
|
|
std::string msg_id = "msg_" + std::to_string(unix_nano_now());
|
|
Json msg_start_obj = Json::object({
|
|
{"type", "message_start"},
|
|
{"message", Json::object({
|
|
{"id", msg_id},
|
|
{"type", "message"},
|
|
{"role", "assistant"},
|
|
{"model", requested_model},
|
|
{"content", Json::array()},
|
|
{"stop_reason", nullptr},
|
|
{"stop_sequence", nullptr},
|
|
{"usage", Json::object({
|
|
{"input_tokens", 0},
|
|
{"output_tokens", 0},
|
|
})},
|
|
})},
|
|
});
|
|
std::string ev1 =
|
|
"event: message_start\ndata: " + msg_start_obj.dump() + "\n\n";
|
|
w.write(ev1);
|
|
|
|
Json block_start_obj = Json::object({
|
|
{"type", "content_block_start"},
|
|
{"index", 0},
|
|
{"content_block", Json::object({
|
|
{"type", "text"},
|
|
{"text", ""},
|
|
})},
|
|
});
|
|
std::string ev2 =
|
|
"event: content_block_start\ndata: " + block_start_obj.dump() + "\n\n";
|
|
w.write(ev2);
|
|
w.flush();
|
|
};
|
|
|
|
ctx.on_stream_data = [&](std::string_view chunk) -> bool {
|
|
if (sse_done) {
|
|
return true;
|
|
}
|
|
sse_line_buf.append(chunk.data(), chunk.size());
|
|
for (;;) {
|
|
size_t nl = sse_line_buf.find('\n');
|
|
if (nl == std::string::npos) {
|
|
break;
|
|
}
|
|
std::string line = sse_line_buf.substr(0, nl);
|
|
sse_line_buf.erase(0, nl + 1);
|
|
if (!process_anthropic_sse_line(line)) {
|
|
return false;
|
|
}
|
|
if (sse_done) {
|
|
break;
|
|
}
|
|
}
|
|
return true;
|
|
};
|
|
}
|
|
|
|
curl_easy_setopt(curl.handle, CURLOPT_NOSIGNAL, 1L);
|
|
curl_easy_setopt(curl.handle, CURLOPT_URL, target_url.c_str());
|
|
curl_easy_setopt(curl.handle, CURLOPT_POST, 1L);
|
|
curl_easy_setopt(curl.handle, CURLOPT_POSTFIELDS, modified_body.data());
|
|
curl_easy_setopt(curl.handle, CURLOPT_POSTFIELDSIZE,
|
|
static_cast<long>(modified_body.size()));
|
|
curl_easy_setopt(curl.handle, CURLOPT_HTTPHEADER, headers_slist.list);
|
|
curl_easy_setopt(curl.handle, CURLOPT_HEADERFUNCTION, curl_header_cb);
|
|
curl_easy_setopt(curl.handle, CURLOPT_HEADERDATA, &ctx);
|
|
curl_easy_setopt(curl.handle, CURLOPT_WRITEFUNCTION, curl_write_cb);
|
|
curl_easy_setopt(curl.handle, CURLOPT_WRITEDATA, &ctx);
|
|
|
|
CURLcode res = curl_easy_perform(curl.handle);
|
|
if (res != CURLE_OK && !ctx.stream_started) {
|
|
log_msg("Trial %zu: Request to %s failed: %s",
|
|
i + 1, target_url.c_str(), curl_easy_strerror(res));
|
|
continue;
|
|
}
|
|
|
|
if (ctx.status_code < 200 || ctx.status_code >= 300) {
|
|
std::string trimmed_err = str_trim(ctx.err_body);
|
|
log_msg("Trial %zu: Request to %s returned error status %d: %s",
|
|
i + 1, target_url.c_str(), ctx.status_code, trimmed_err.c_str());
|
|
continue;
|
|
}
|
|
|
|
if (is_stream) {
|
|
if (!sse_done && !sse_line_buf.empty()) {
|
|
process_anthropic_sse_line(sse_line_buf);
|
|
sse_line_buf.clear();
|
|
}
|
|
Json block_stop_obj = Json::object({
|
|
{"type", "content_block_stop"},
|
|
{"index", 0},
|
|
});
|
|
w.write("event: content_block_stop\ndata: " + block_stop_obj.dump() +
|
|
"\n\n");
|
|
|
|
Json msg_delta_obj = Json::object({
|
|
{"type", "message_delta"},
|
|
{"delta", Json::object({
|
|
{"stop_reason", final_stop_reason},
|
|
{"stop_sequence", nullptr},
|
|
})},
|
|
{"usage", Json::object({
|
|
{"output_tokens", 0},
|
|
})},
|
|
});
|
|
w.write("event: message_delta\ndata: " + msg_delta_obj.dump() + "\n\n");
|
|
|
|
Json msg_stop_obj = Json::object({
|
|
{"type", "message_stop"},
|
|
});
|
|
w.write("event: message_stop\ndata: " + msg_stop_obj.dump() + "\n\n");
|
|
w.flush();
|
|
return;
|
|
}
|
|
|
|
log_msg("Trial %zu: Connection established with status %d. Proxying response.",
|
|
i + 1, ctx.status_code);
|
|
|
|
Json openai_resp;
|
|
if (!Json::parse(ctx.resp_body, openai_resp)) {
|
|
log_msg("Trial %zu: Failed to unmarshal upstream response JSON", i + 1);
|
|
continue;
|
|
}
|
|
if (!openai_resp.is_object()) {
|
|
log_msg("Trial %zu: Upstream response JSON is not an object", i + 1);
|
|
continue;
|
|
}
|
|
|
|
Json anthropic_resp =
|
|
convert_openai_to_anthropic_response(openai_resp, requested_model);
|
|
|
|
for (const auto &kv : ctx.resp_headers.items) {
|
|
if (should_skip_upstream_resp_header(kv.first, true)) {
|
|
continue;
|
|
}
|
|
w.header().add(kv.first, kv.second);
|
|
}
|
|
w.header().set("Content-Type", "application/json");
|
|
w.write_header(ctx.status_code);
|
|
w.write(anthropic_resp.dump() + "\n");
|
|
w.flush();
|
|
return;
|
|
}
|
|
|
|
log_msg("All trials failed. Returning Bad Gateway.");
|
|
send_json_error(w, 502, "All configured models and keys failed to respond.",
|
|
"gateway_error", true, "all_endpoints_failed");
|
|
}
|
|
|
|
void handle_responses(
|
|
ConfigManager &cm,
|
|
const std::vector<std::string> &expected_tokens,
|
|
std::chrono::milliseconds retry_base_delay,
|
|
const HttpRequest &req,
|
|
ResponseWriter &w) {
|
|
ensure_curl_initialized();
|
|
|
|
if (!check_auth(expected_tokens, req)) {
|
|
send_unauthorized(w);
|
|
return;
|
|
}
|
|
|
|
if (req.method != "POST") {
|
|
send_json_error(w, 405, "Method not allowed", "invalid_request_error");
|
|
return;
|
|
}
|
|
|
|
Json body_map;
|
|
if (!Json::parse(req.body, body_map) || !body_map.is_object()) {
|
|
send_json_error(w, 400, "Invalid JSON in request body",
|
|
"invalid_request_error");
|
|
return;
|
|
}
|
|
|
|
Json openai_body_map = convert_responses_to_openai(body_map);
|
|
|
|
std::string requested_model;
|
|
if (openai_body_map.contains("model") &&
|
|
openai_body_map["model"].is_string()) {
|
|
requested_model = openai_body_map["model"].sval;
|
|
}
|
|
|
|
auto configs = cm.get_configs();
|
|
auto unique_models = cm.get_unique_models();
|
|
|
|
if (configs.empty()) {
|
|
send_json_error(w, 503, "No model configurations loaded", "gateway_error");
|
|
return;
|
|
}
|
|
|
|
auto trial_configs =
|
|
get_trial_configs(configs, unique_models, requested_model);
|
|
if (trial_configs.empty()) {
|
|
send_json_error(w, 503, "No valid trial configuration candidates",
|
|
"gateway_error");
|
|
return;
|
|
}
|
|
|
|
bool is_stream = false;
|
|
if (openai_body_map.contains("stream") &&
|
|
openai_body_map["stream"].is_bool()) {
|
|
is_stream = openai_body_map["stream"].as_bool();
|
|
}
|
|
|
|
log_msg("Received OpenAI Responses request for model \"%s\" (stream=%s). Found %zu config trials.",
|
|
requested_model.c_str(), is_stream ? "true" : "false",
|
|
trial_configs.size());
|
|
|
|
for (size_t i = 0; i < trial_configs.size(); ++i) {
|
|
const auto &trial = trial_configs[i];
|
|
if (i > 0 && retry_base_delay.count() > 0) {
|
|
auto delay = std::chrono::milliseconds(
|
|
fibonacci(static_cast<int>(i)) * retry_base_delay.count());
|
|
log_msg("Trial %zu/%zu: Fibonacci backoff delay of %lldms before retry...",
|
|
i + 1, trial_configs.size(),
|
|
static_cast<long long>(delay.count()));
|
|
if (!wait_backoff_or_cancelled(delay, req)) {
|
|
log_msg("Request context cancelled during retry delay before trial %zu",
|
|
i + 1);
|
|
send_json_error(w, 499, "Client closed request", "gateway_error");
|
|
return;
|
|
}
|
|
}
|
|
|
|
log_msg("Trial %zu/%zu: model=%s endpoint=%s key_len=%zu",
|
|
i + 1, trial_configs.size(), trial.model.c_str(),
|
|
trial.endpoint.c_str(), trial.key.size());
|
|
|
|
openai_body_map["model"] = trial.model;
|
|
std::string modified_body = openai_body_map.dump();
|
|
std::string target_url = build_url(trial.endpoint, "/v1/chat/completions");
|
|
|
|
ScopedCurl curl;
|
|
if (!curl.handle) {
|
|
log_msg("Trial %zu: Failed to initialize curl handle", i + 1);
|
|
continue;
|
|
}
|
|
|
|
ScopedCurlSlist headers_slist;
|
|
build_outgoing_headers(req, trial, i + 1, headers_slist);
|
|
|
|
UpstreamTransferCtx ctx;
|
|
ctx.is_stream = is_stream;
|
|
|
|
std::string sse_line_buf;
|
|
std::string full_text_buf;
|
|
std::string resp_id;
|
|
std::string msg_id;
|
|
bool sse_done = false;
|
|
|
|
auto process_responses_sse_line = [&](std::string_view raw_line) -> bool {
|
|
std::string line = str_trim(raw_line);
|
|
constexpr std::string_view kDataPrefix = "data: ";
|
|
if (line.size() < kDataPrefix.size() ||
|
|
std::string_view(line).substr(0, kDataPrefix.size()) != kDataPrefix) {
|
|
return true;
|
|
}
|
|
std::string data_str =
|
|
str_trim(std::string_view(line).substr(kDataPrefix.size()));
|
|
if (data_str == "[DONE]") {
|
|
sse_done = true;
|
|
return true;
|
|
}
|
|
Json chunk_mut;
|
|
if (Json::parse(data_str, chunk_mut) && chunk_mut.is_object()) {
|
|
const Json &chunk_map = chunk_mut;
|
|
if (chunk_map["choices"].is_array() && !chunk_map["choices"].empty()) {
|
|
const auto &choice = chunk_map["choices"].at(0);
|
|
if (choice.is_object() && choice["delta"].is_object()) {
|
|
const auto &delta = choice["delta"];
|
|
if (delta["content"].is_string()) {
|
|
const std::string &content_str = delta["content"].sval;
|
|
if (!content_str.empty()) {
|
|
full_text_buf.append(content_str);
|
|
Json delta_obj = Json::object({
|
|
{"type", "response.output_text.delta"},
|
|
{"delta", content_str},
|
|
});
|
|
std::string ev = "event: response.output_text.delta\ndata: " +
|
|
delta_obj.dump() + "\n\n";
|
|
if (!w.write(ev)) {
|
|
return false;
|
|
}
|
|
w.flush();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return true;
|
|
};
|
|
|
|
if (is_stream) {
|
|
ctx.on_stream_start = [&]() {
|
|
log_msg("Trial %zu: Connection established with status %d. Proxying response.",
|
|
i + 1, ctx.status_code);
|
|
w.header().set("Content-Type", "text/event-stream");
|
|
w.header().set("Cache-Control", "no-cache");
|
|
w.header().set("Connection", "keep-alive");
|
|
w.write_header(200);
|
|
w.flush();
|
|
|
|
resp_id = "resp_" + std::to_string(unix_nano_now());
|
|
msg_id = "msg_" + std::to_string(unix_nano_now());
|
|
|
|
Json created_obj = Json::object({
|
|
{"type", "response.created"},
|
|
{"response", Json::object({
|
|
{"id", resp_id},
|
|
{"object", "response"},
|
|
{"status", "in_progress"},
|
|
{"model", requested_model},
|
|
})},
|
|
});
|
|
w.write("event: response.created\ndata: " + created_obj.dump() +
|
|
"\n\n");
|
|
|
|
Json part_added_obj = Json::object({
|
|
{"type", "response.content_part.added"},
|
|
{"part", Json::object({
|
|
{"type", "output_text"},
|
|
{"text", ""},
|
|
})},
|
|
});
|
|
w.write("event: response.content_part.added\ndata: " +
|
|
part_added_obj.dump() + "\n\n");
|
|
w.flush();
|
|
};
|
|
|
|
ctx.on_stream_data = [&](std::string_view chunk) -> bool {
|
|
if (sse_done) {
|
|
return true;
|
|
}
|
|
sse_line_buf.append(chunk.data(), chunk.size());
|
|
for (;;) {
|
|
size_t nl = sse_line_buf.find('\n');
|
|
if (nl == std::string::npos) {
|
|
break;
|
|
}
|
|
std::string line = sse_line_buf.substr(0, nl);
|
|
sse_line_buf.erase(0, nl + 1);
|
|
if (!process_responses_sse_line(line)) {
|
|
return false;
|
|
}
|
|
if (sse_done) {
|
|
break;
|
|
}
|
|
}
|
|
return true;
|
|
};
|
|
}
|
|
|
|
curl_easy_setopt(curl.handle, CURLOPT_NOSIGNAL, 1L);
|
|
curl_easy_setopt(curl.handle, CURLOPT_URL, target_url.c_str());
|
|
curl_easy_setopt(curl.handle, CURLOPT_POST, 1L);
|
|
curl_easy_setopt(curl.handle, CURLOPT_POSTFIELDS, modified_body.data());
|
|
curl_easy_setopt(curl.handle, CURLOPT_POSTFIELDSIZE,
|
|
static_cast<long>(modified_body.size()));
|
|
curl_easy_setopt(curl.handle, CURLOPT_HTTPHEADER, headers_slist.list);
|
|
curl_easy_setopt(curl.handle, CURLOPT_HEADERFUNCTION, curl_header_cb);
|
|
curl_easy_setopt(curl.handle, CURLOPT_HEADERDATA, &ctx);
|
|
curl_easy_setopt(curl.handle, CURLOPT_WRITEFUNCTION, curl_write_cb);
|
|
curl_easy_setopt(curl.handle, CURLOPT_WRITEDATA, &ctx);
|
|
|
|
CURLcode res = curl_easy_perform(curl.handle);
|
|
if (res != CURLE_OK && !ctx.stream_started) {
|
|
log_msg("Trial %zu: Request to %s failed: %s",
|
|
i + 1, target_url.c_str(), curl_easy_strerror(res));
|
|
continue;
|
|
}
|
|
|
|
if (ctx.status_code < 200 || ctx.status_code >= 300) {
|
|
std::string trimmed_err = str_trim(ctx.err_body);
|
|
log_msg("Trial %zu: Request to %s returned error status %d: %s",
|
|
i + 1, target_url.c_str(), ctx.status_code, trimmed_err.c_str());
|
|
continue;
|
|
}
|
|
|
|
if (is_stream) {
|
|
if (!sse_done && !sse_line_buf.empty()) {
|
|
process_responses_sse_line(sse_line_buf);
|
|
sse_line_buf.clear();
|
|
}
|
|
Json completed_obj = Json::object({
|
|
{"type", "response.completed"},
|
|
{"response", Json::object({
|
|
{"id", resp_id},
|
|
{"object", "response"},
|
|
{"status", "completed"},
|
|
{"model", requested_model},
|
|
{"created_at", unix_sec_now()},
|
|
{"output", Json::array({
|
|
Json::object({
|
|
{"id", msg_id},
|
|
{"type", "message"},
|
|
{"status", "completed"},
|
|
{"role", "assistant"},
|
|
{"content", Json::array({
|
|
Json::object({
|
|
{"type", "output_text"},
|
|
{"text", full_text_buf},
|
|
{"annotations",
|
|
Json::array()},
|
|
{"logprobs",
|
|
Json::array()},
|
|
}),
|
|
})},
|
|
}),
|
|
})},
|
|
})},
|
|
});
|
|
w.write("event: response.completed\ndata: " + completed_obj.dump() +
|
|
"\n\n");
|
|
w.flush();
|
|
return;
|
|
}
|
|
|
|
log_msg("Trial %zu: Connection established with status %d. Proxying response.",
|
|
i + 1, ctx.status_code);
|
|
|
|
Json openai_resp;
|
|
if (!Json::parse(ctx.resp_body, openai_resp)) {
|
|
log_msg("Trial %zu: Failed to unmarshal upstream response JSON", i + 1);
|
|
continue;
|
|
}
|
|
if (!openai_resp.is_object()) {
|
|
log_msg("Trial %zu: Upstream response JSON is not an object", i + 1);
|
|
continue;
|
|
}
|
|
|
|
Json responses_resp =
|
|
convert_openai_to_responses_response(openai_resp, requested_model);
|
|
|
|
for (const auto &kv : ctx.resp_headers.items) {
|
|
if (should_skip_upstream_resp_header(kv.first, true)) {
|
|
continue;
|
|
}
|
|
w.header().add(kv.first, kv.second);
|
|
}
|
|
w.header().set("Content-Type", "application/json");
|
|
w.write_header(ctx.status_code);
|
|
w.write(responses_resp.dump() + "\n");
|
|
w.flush();
|
|
return;
|
|
}
|
|
|
|
log_msg("All trials failed. Returning Bad Gateway.");
|
|
send_json_error(w, 502, "All configured models and keys failed to respond.",
|
|
"gateway_error", true, "all_endpoints_failed");
|
|
}
|