luxferre/bantam-cpp
1
0
Fork 0
Code Actions Activity

Optimize startup, HTTP/SSE streaming, and fix macOS libedit bindings

Browse Source
This commit is contained in:
Luxferre
2026-10-14 15:30:12 +03:00
parent 105d283e01
commit a1d5f4bf48
10 changed files with 497 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files
llm.cpp
+185 -83
View File
@@ -18,15 +18,29 @@ static size_t mem_cb(char *ptr, size_t sz, size_t nm, void *ud) {
return total;
}
/* ----------------------------- curl RAII ------------------------------ */
struct CurlHandle {
CURL *c;
CurlHandle() : c(curl_easy_init()) {}
~CurlHandle() { if (c) curl_easy_cleanup(c); }
CurlHandle(const CurlHandle &) = delete;
CurlHandle &operator=(const CurlHandle &) = delete;
explicit operator bool() const { return c != nullptr; }
};
static int xferinfo_cb(void *clientp, curl_off_t, curl_off_t, curl_off_t, curl_off_t) {
volatile int *interrupted = (volatile int *)clientp;
return (interrupted && *interrupted) ? 1 : 0;
}
/* ----------------------------- curl handle & headers ------------------ */
static CURL *g_shared_curl = nullptr;
static CURL *get_shared_curl() {
if (!g_shared_curl) {
g_shared_curl = curl_easy_init();
} else {
curl_easy_reset(g_shared_curl);
}
return g_shared_curl;
}
void http_cleanup() {
if (g_shared_curl) {
curl_easy_cleanup(g_shared_curl);
g_shared_curl = nullptr;
}
}
struct CurlHeaders {
curl_slist *list = nullptr;
@@ -53,8 +67,23 @@ static void set_proxy(CURL *c) {
}
}
static void apply_common_curl_opts(CURL *c, volatile int *interrupted) {
curl_easy_setopt(c, CURLOPT_NOSIGNAL, 1L);
curl_easy_setopt(c, CURLOPT_NOPROGRESS, 0L);
curl_easy_setopt(c, CURLOPT_XFERINFOFUNCTION, xferinfo_cb);
curl_easy_setopt(c, CURLOPT_XFERINFODATA, (void *)interrupted);
curl_easy_setopt(c, CURLOPT_TCP_KEEPALIVE, 1L);
curl_easy_setopt(c, CURLOPT_TCP_KEEPIDLE, 30L);
curl_easy_setopt(c, CURLOPT_TCP_KEEPINTVL, 15L);
curl_easy_setopt(c, CURLOPT_USERAGENT, "Mozilla/5.0 Bantam/nightly-cpp");
set_proxy(c);
}
static void apply_llm_headers(CurlHeaders &h, const Cfg &cfg) {
h.append("Content-Type: application/json");
/* Suppress Expect: 100-continue so large POST bodies are sent immediately
* without waiting for a 100 Continue round-trip or hanging behind proxies. */
h.append("Expect:");
if (!cfg.api_key.empty() && cfg.api_key != "-") {
h.append(sfmt("Authorization: Bearer %s", cfg.api_key.c_str()));
}
@@ -63,21 +92,32 @@ static void apply_llm_headers(CurlHeaders &h, const Cfg &cfg) {
/* ----------------------------- http_get ------------------------------- */
int http_get(Cfg &cfg, std::string_view url, std::string &body, long &code,
std::string &errmsg) {
CurlHandle curl;
if (!curl) { errmsg = "curl init failed"; return -1; }
CURL *c = get_shared_curl();
if (!c) { errmsg = "curl init failed"; return -1; }
std::string url_s(url);
curl_easy_setopt(curl.c, CURLOPT_URL, url_s.c_str());
curl_easy_setopt(curl.c, CURLOPT_WRITEFUNCTION, mem_cb);
curl_easy_setopt(curl.c, CURLOPT_WRITEDATA, &body);
set_proxy(curl.c);
long t = cfg.timeout > 0 ? cfg.timeout : 30;
curl_easy_setopt(curl.c, CURLOPT_CONNECTTIMEOUT, t);
curl_easy_setopt(curl.c, CURLOPT_TIMEOUT, t);
curl_easy_setopt(curl.c, CURLOPT_USERAGENT, "Mozilla/5.0 Bantam/nightly-cpp");
CURLcode res = curl_easy_perform(curl.c);
curl_easy_getinfo(curl.c, CURLINFO_RESPONSE_CODE, &code);
CurlHeaders headers;
if (!cfg.api_key.empty() && cfg.api_key != "-") {
headers.append(sfmt("Authorization: Bearer %s", cfg.api_key.c_str()));
}
apply_common_curl_opts(c, &g_interrupted);
curl_easy_setopt(c, CURLOPT_URL, url_s.c_str());
curl_easy_setopt(c, CURLOPT_HTTPGET, 1L);
if (headers.list) curl_easy_setopt(c, CURLOPT_HTTPHEADER, headers.list);
curl_easy_setopt(c, CURLOPT_WRITEFUNCTION, mem_cb);
curl_easy_setopt(c, CURLOPT_WRITEDATA, &body);
long cfg_t = cfg.timeout > 0 ? cfg.timeout : 15;
long conn_t = (cfg_t < 10) ? cfg_t : 10;
long tot_t = (cfg_t < 15) ? cfg_t : 15;
curl_easy_setopt(c, CURLOPT_CONNECTTIMEOUT, conn_t);
curl_easy_setopt(c, CURLOPT_TIMEOUT, tot_t);
curl_easy_setopt(c, CURLOPT_LOW_SPEED_LIMIT, 1L);
curl_easy_setopt(c, CURLOPT_LOW_SPEED_TIME, tot_t);
CURLcode res = curl_easy_perform(c);
curl_easy_getinfo(c, CURLINFO_RESPONSE_CODE, &code);
if (res != CURLE_OK) {
errmsg = curl_easy_strerror(res);
errmsg = (res == CURLE_ABORTED_BY_CALLBACK || g_interrupted)
? "interrupted"
: curl_easy_strerror(res);
return -1;
}
return 0;
@@ -113,11 +153,14 @@ static int est_tokens_msg(const Message &m) {
/* ----------------------------- streaming parser ----------------------- */
struct StreamCtx {
std::string content, reas, line_buf, live_printed, buf, raw;
std::string content, reas, line_buf, buf, raw;
std::vector<std::string> tbl;
Usage last_usage;
std::vector<ToolCall> tcs;
std::vector<std::string> tcs_filt_args;
std::vector<size_t> tcs_filt_pos;
std::vector<int> tcs_idx;
size_t live_printed_len = 0;
int tool_calls_shown = 0;
int tty = 0;
int live_idx = -1;
@@ -128,6 +171,32 @@ struct StreamCtx {
volatile int *interrupted = nullptr;
};
static void advance_filtered_args(StreamCtx *sc, int slot, bool flush_all) {
const std::string &raw_args = sc->tcs[slot].arguments;
size_t pos = sc->tcs_filt_pos[slot];
if (pos >= raw_args.size()) return;
std::string_view rem = std::string_view(raw_args).substr(pos);
if (!flush_all) {
/* Hold back an incomplete trailing multi-byte UTF-8 sequence until the
* next chunk arrives. */
size_t back = 0;
while (back < 3 && back < rem.size()) {
unsigned char b = (unsigned char)rem[rem.size() - 1 - back];
if ((b & 0x80) == 0) break; /* ASCII */
if ((b & 0xC0) == 0xC0) { /* UTF-8 lead byte */
size_t need = ((b & 0xE0) == 0xC0) ? 2 : ((b & 0xF0) == 0xE0) ? 3 : 4;
if (back + 1 < need) rem.remove_suffix(back + 1);
break;
}
back++;
}
}
if (!rem.empty()) {
sc->tcs_filt_args[slot] += filter_text(rem);
sc->tcs_filt_pos[slot] = pos + rem.size();
}
}
static void stream_upsert_tc(StreamCtx *sc, int idx, std::string_view id,
std::string_view name, std::string_view args) {
int slot = -1;
@@ -136,30 +205,54 @@ static void stream_upsert_tc(StreamCtx *sc, int idx, std::string_view id,
if (slot < 0) {
slot = (int)sc->tcs.size();
sc->tcs.emplace_back();
sc->tcs_filt_args.emplace_back();
sc->tcs_filt_pos.push_back(0);
sc->tcs_idx.push_back(idx);
}
if (!id.empty()) sc->tcs[slot].id = std::string(id);
if (!name.empty()) sc->tcs[slot].name += std::string(name);
if (!args.empty()) sc->tcs[slot].arguments += std::string(args);
if (!args.empty()) {
sc->tcs[slot].arguments += std::string(args);
advance_filtered_args(sc, slot, false);
}
}
static void live_target(StreamCtx *sc, int slot, bool closed, std::string &out) {
ToolCall &t = sc->tcs[slot];
str_appendf(out, "[tool call: %s", t.name.c_str());
if (closed || !t.arguments.empty()) {
std::string fa = filter_text(t.arguments);
str_appendf(out, "(%s", fa.c_str());
if (closed) out += ")]";
static void print_live_delta(StreamCtx *sc, int slot, bool closed) {
if (closed) advance_filtered_args(sc, slot, true);
const ToolCall &t = sc->tcs[slot];
const std::string &fa = sc->tcs_filt_args[slot];
bool has_args = closed || !t.arguments.empty();
size_t prefix_len = 12 + t.name.size(); /* "[tool call: " + name */
size_t total_len = prefix_len + (has_args ? (1 + fa.size() + (closed ? 2 : 0)) : 0);
size_t cur = sc->live_printed_len;
if (total_len <= cur) return;
std::string delta;
delta.reserve(total_len - cur);
auto emit_part = [&](std::string_view part, size_t part_start) {
size_t part_end = part_start + part.size();
if (part_end <= cur) return;
size_t off = (cur > part_start) ? (cur - part_start) : 0;
delta.append(part.substr(off));
};
size_t off = 0;
emit_part("[tool call: ", off); off += 12;
emit_part(t.name, off); off += t.name.size();
if (has_args) {
emit_part("(", off); off += 1;
emit_part(fa, off); off += fa.size();
if (closed) {
emit_part(")]", off); off += 2;
}
}
cprint(delta, std::array{33});
sc->live_printed_len = total_len;
}
static void finish_live(StreamCtx *sc, bool closed) {
if (sc->live_idx < 0) return;
std::string target;
live_target(sc, sc->live_idx, closed, target);
if (target.size() > sc->live_printed.size()) {
cprint(target.substr(sc->live_printed.size()), std::array{33});
}
sc->live_printed.clear();
print_live_delta(sc, sc->live_idx, closed);
sc->live_printed_len = 0;
sc->live_idx = -1;
/* The tool call marker must always terminate its own line so the
* following [tool result: ...] header is separated from it by a
@@ -172,13 +265,8 @@ static void step_live(StreamCtx *sc) {
if (!sc->tty || sc->tcs.empty()) return;
int cur = (int)sc->tcs.size() - 1;
if (sc->live_idx != -1 && sc->live_idx != cur) finish_live(sc, true);
if (sc->live_idx == -1) { sc->live_idx = cur; sc->live_printed.clear(); }
std::string target;
live_target(sc, cur, false, target);
if (target.size() > sc->live_printed.size()) {
cprint(target.substr(sc->live_printed.size()), std::array{33});
sc->live_printed += target.substr(sc->live_printed.size());
}
if (sc->live_idx == -1) { sc->live_idx = cur; sc->live_printed_len = 0; }
print_live_delta(sc, cur, false);
sc->tool_calls_shown = 1;
}
@@ -196,7 +284,7 @@ static void stream_maybe_erase(StreamCtx *sc) {
}
}
static void stream_handle_data(StreamCtx *sc, const std::string &data) {
static void stream_handle_data(StreamCtx *sc, std::string_view data) {
if (data == "[DONE]") { sc->done = 1; return; }
Json d_mut;
if (!Json::parse(data, d_mut) || !d_mut.is_object()) return;
@@ -253,21 +341,23 @@ static void stream_handle_data(StreamCtx *sc, const std::string &data) {
if (sc->in_reasoning) { cprint("\n--- reasoning end ---\n", std::array{36}); sc->in_reasoning = false; }
sc->content += ct;
sc->line_buf += ct;
size_t lpos = 0;
while (true) {
auto nl = sc->line_buf.find('\n');
auto nl = sc->line_buf.find('\n', lpos);
if (nl == std::string::npos) break;
std::string cur = sc->line_buf.substr(0, nl);
sc->line_buf.erase(0, nl + 1);
std::string_view cur_sv = std::string_view(sc->line_buf).substr(lpos, nl - lpos);
lpos = nl + 1;
/* Inside a code block every line is rendered verbatim as code; only
* outside a code block do table-shaped lines get accumulated for
* box-drawing table rendering. */
if (sc->md.in_code || !is_table_line(cur)) {
if (sc->md.in_code || !is_table_line(cur_sv)) {
flush_tbl(sc);
render_md_line_print(cur, sc->md);
render_md_line_print(cur_sv, sc->md);
} else {
sc->tbl.push_back(cur);
sc->tbl.emplace_back(cur_sv);
}
}
if (lpos > 0) sc->line_buf.erase(0, lpos);
}
const Json &tcs = delta["tool_calls"];
if (tcs.is_array()) {
@@ -297,22 +387,26 @@ static int stream_feed(StreamCtx *sc, const char *data, size_t len,
volatile int *interrupted) {
sc->interrupted = interrupted;
sc->buf.append(data, len);
size_t pos = 0;
while (true) {
auto nl = sc->buf.find('\n');
auto nl = sc->buf.find('\n', pos);
if (nl == std::string::npos) break;
std::string line = sc->buf.substr(0, nl);
std::string_view line = std::string_view(sc->buf).substr(pos, nl - pos);
pos = nl + 1;
while (!line.empty() && (line.back() == '\r' || line.back() == ' ' || line.back() == '\t'))
line.pop_back();
if (line.rfind("data:", 0) == 0) {
std::string_view dd = line;
dd.remove_prefix(5);
while (!dd.empty() && dd.front() == ' ') dd.remove_prefix(1);
stream_handle_data(sc, std::string(dd));
line.remove_suffix(1);
if (line.starts_with("data:")) {
line.remove_prefix(5);
while (!line.empty() && line.front() == ' ') line.remove_prefix(1);
stream_handle_data(sc, line);
}
sc->buf.erase(0, nl + 1);
if (sc->done) break;
if (*interrupted) return -1;
if (*interrupted) {
sc->buf.erase(0, pos);
return -1;
}
}
if (pos > 0) sc->buf.erase(0, pos);
return 0;
}
@@ -336,9 +430,9 @@ static void stream_result(StreamCtx *sc, Message &out, Usage &usage) {
out = Message{};
out.role = "assistant";
out.calls_shown = sc->tool_calls_shown;
if (!sc->content.empty()) out.content = sc->content;
if (!sc->reas.empty()) out.reasoning = sc->reas;
for (auto &t : sc->tcs) out.calls.push_back(t);
if (!sc->content.empty()) out.content = std::move(sc->content);
if (!sc->reas.empty()) out.reasoning = std::move(sc->reas);
for (auto &t : sc->tcs) out.calls.push_back(std::move(t));
usage = sc->last_usage;
if (usage.total_tokens == 0 && usage.prompt_tokens > 0)
usage.total_tokens = usage.prompt_tokens + usage.completion_tokens;
@@ -349,7 +443,11 @@ static size_t stream_write_cb(char *ptr, size_t sz, size_t nm, void *ud) {
StreamCtx *sc = (StreamCtx *)ud;
size_t total = sz * nm;
if (*sc->interrupted) return 0;
sc->raw.append((const char *)ptr, total);
/* Cap raw error buffer at 4096 bytes; http_err_body only uses 4000 bytes. */
if (sc->raw.size() < 4096) {
size_t rem = 4096 - sc->raw.size();
sc->raw.append((const char *)ptr, total < rem ? total : rem);
}
int r = stream_feed(sc, ptr, total, sc->interrupted);
if (r != 0 || *sc->interrupted) return 0;
return total;
@@ -388,20 +486,23 @@ int llm(Cfg &cfg, Messages &msgs, bool use_tools, Message &out, Usage &usage,
int nf = 9;
std::string pend = cstr("...requesting...", std::array{1, 2});
CurlHandle curl;
if (!curl) { errmsg = "curl init failed"; return -1; }
curl_easy_setopt(curl.c, CURLOPT_USERAGENT, "Mozilla/5.0 Bantam/nightly-cpp");
CURL *c = get_shared_curl();
if (!c) { errmsg = "curl init failed"; return -1; }
CurlHeaders headers;
apply_llm_headers(headers, cfg);
curl_easy_setopt(curl.c, CURLOPT_URL, url.c_str());
curl_easy_setopt(curl.c, CURLOPT_POST, 1L);
curl_easy_setopt(curl.c, CURLOPT_POSTFIELDS, payload.c_str());
curl_easy_setopt(curl.c, CURLOPT_POSTFIELDSIZE, (long)payload.size());
curl_easy_setopt(curl.c, CURLOPT_HTTPHEADER, headers.list);
set_proxy(curl.c);
curl_easy_setopt(curl.c, CURLOPT_CONNECTTIMEOUT, (long)(cfg.timeout > 0 ? cfg.timeout : 30));
if (!stream) curl_easy_setopt(curl.c, CURLOPT_TIMEOUT, (long)(cfg.timeout > 0 ? cfg.timeout : 300));
apply_common_curl_opts(c, interrupted);
curl_easy_setopt(c, CURLOPT_URL, url.c_str());
curl_easy_setopt(c, CURLOPT_POST, 1L);
curl_easy_setopt(c, CURLOPT_POSTFIELDS, payload.c_str());
curl_easy_setopt(c, CURLOPT_POSTFIELDSIZE, (long)payload.size());
curl_easy_setopt(c, CURLOPT_HTTPHEADER, headers.list);
long t = (long)(cfg.timeout > 0 ? cfg.timeout : 300);
long conn_t = (t < 20) ? t : 20;
curl_easy_setopt(c, CURLOPT_CONNECTTIMEOUT, conn_t);
curl_easy_setopt(c, CURLOPT_LOW_SPEED_LIMIT, 1L);
curl_easy_setopt(c, CURLOPT_LOW_SPEED_TIME, t);
if (!stream) curl_easy_setopt(c, CURLOPT_TIMEOUT, t);
int attempt, rc = -1;
for (attempt = 0; attempt <= nf; attempt++) {
@@ -416,20 +517,21 @@ int llm(Cfg &cfg, Messages &msgs, bool use_tools, Message &out, Usage &usage,
sc.tty = is_terminal(STDOUT_FILENO);
sc.interrupted = interrupted;
sc.erase_pending = 1;
curl_easy_setopt(curl.c, CURLOPT_WRITEFUNCTION, stream_write_cb);
curl_easy_setopt(curl.c, CURLOPT_WRITEDATA, &sc);
curl_easy_setopt(c, CURLOPT_WRITEFUNCTION, stream_write_cb);
curl_easy_setopt(c, CURLOPT_WRITEDATA, &sc);
} else {
curl_easy_setopt(curl.c, CURLOPT_WRITEFUNCTION, mem_cb);
curl_easy_setopt(curl.c, CURLOPT_WRITEDATA, &mbody);
curl_easy_setopt(c, CURLOPT_WRITEFUNCTION, mem_cb);
curl_easy_setopt(c, CURLOPT_WRITEDATA, &mbody);
}
res = curl_easy_perform(curl.c);
if (*interrupted) {
res = curl_easy_perform(c);
if (*interrupted || res == CURLE_ABORTED_BY_CALLBACK) {
*interrupted = 1;
if (g_col && (!stream || sc.erase_pending)) printf("\r\033[K");
errmsg = "interrupted";
rc = -1;
break;
}
curl_easy_getinfo(curl.c, CURLINFO_RESPONSE_CODE, &code);
curl_easy_getinfo(c, CURLINFO_RESPONSE_CODE, &code);
int transport_err = (res != CURLE_OK);
int is4xx = (!transport_err && code >= 400 && code < 500 && code != 408 && code != 429);