From 7748ec450ee27a13802171914fe003df43a92071 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sat, 26 Sep 2026 10:32:20 +0800 Subject: [PATCH] =?UTF-8?q?perf(api):=20SSE=20=E5=88=86=E5=9D=97=20C=20?= =?UTF-8?q?=E5=AF=BC=E8=88=AA=E5=B1=82=20+=20=E5=B7=AE=E5=88=86=E7=AD=89?= =?UTF-8?q?=E4=BB=B7=E9=AA=8C=E6=94=B6=EF=BC=88=E5=AE=9E=E6=B5=8B=E6=9B=B4?= =?UTF-8?q?=E6=85=A2=20=E2=87=92=20=E9=BB=98=E8=AE=A4=E5=85=B3=E9=97=AD?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 第三刀：把 ha_json_scan 接进 parseOpenAICompatibleStreamChunkFull。 **结论是否定的** —— 实测比原实现慢，故默认关闭并如实记录。这条提交的 价值在于「已钉死的正确性 + 已定位的根因 + 一条防静默回退的断言」。 ## 设计：只做「结构导航」，序列化留在 Go 接线前实测出两条 wire 语义，它们让「整条解析全 C 化」不成立： §5.1 重复键是**字段级合并**，不是替换： {"choices":[{content:a}],"choices":[{reasoning:r}]} → 两个都保留。 机制：json.Unmarshal 的 object() 收尾做 v.SetIndex(i, subv.v)， 而 subv 拿到的是**已存在元素的指针** ⇒ 第二次是叠加。 §5.2 stringifyContent 的 default 分支 = json.Marshal(interface{})， 即**重新序列化**：{"b":1,"a":2}→{"a":2,"b":1}（键排序）、 1e2→100、<→\u003c、大 int 先舍入成 float64。 逐值一致 = 复刻 Ryu 最短浮点 + map 键排序 + HTML 转义 + int 舍入。 两条都只在**取值**阶段需要，故 C 只回答「值在哪里」（零分配零解码）， 类型检查靠「用相同的 Go 类型 unmarshal 相同形状的子树」保证，不靠 C 复刻规则。 ## 实测：新路径比原实现慢（20000 次迭代） | 场景 | 新路径 | 原实现 | |---|---|---| | content_ascii | 2016ns / 20allocs | 1325ns / 13allocs | | toolcall | 5854ns / 33allocs | 3270ns / 21allocs | | usage | 3170ns / 24allocs | 2832ns / 12allocs | 分配数**也变多**（20 vs 13），与「消除 GC 抖动」的初衷相反。 根因（逐项测量，非猜测）：裸 cgo 调用 168ns；**每次带 out-param 的键查找 205ns + 2 allocs**（out-param 逃逸到堆）；一次解析需要 5+ 次查找 ⇒ 边界与分配成本约 1µs，恰好吃掉全部收益。Go 侧只需**一次** Unmarshal。 一句话：**用很多次廉价调用换一次昂贵调用，在这个尺寸上不划算。** ## 天花板实验：方向对，但当前实现没到 假设拿到 span 完全免费，只测设计中必须由 Go 做的部分： 我的 Go 侧 505ns/7allocs vs 原实现 1239ns/13allocs ⇒ 边界归零后仍有 2.4× 时间、46% 分配的空间。故问题在**逐字段往返** 这个交互方式，不在 C 本身。正确改造：一次调用返回全部字段 span + 结果写调用方栈结构体 + 仅在确需重新编码时回退。 ## 正确性：6 万+ 差分用例全过 同一批输入跑两条路径逐字段比对（Content/Reasoning/Done/Finish/ToolCalls/ Usage + bool），5 组：协议形态（含全部回退触发条件）、真实负载、随机 JSON 30000 例、随机字节 30000 例、优化有效性。 ★ 差分测试当场抓出 4 个真实缺陷（其中一个正是「优化压根没生效」）： 1. ha_sse_arr_first 里「重新 init 到 sc.s+sc.i」使 base 变了 ⇒ start 恒 0 ⇒ 返回的是**数组本身**而非首元素。症状是**快速路径永远不生效**—— 而若只看「结果与 Go 一致」，这个 bug 会**完全隐形**（回退总是对的）。 ⇒ 这就是必须单独断言「优化确实被走到」的原因。 2. chunkAssemble 的 bool 被丢弃 ⇒ 空对象被判 true（原实现 false） 3. 键匹配层级搞错：delta 是 **struct**（字段名 CI），不是 map。 我一度「推理」成 CS 并以为差分测试会通过——错的。 教教训：哪层是 struct、哪层是 map 要**回原实现读类型**，不能凭字段名推断。 4. cgo 边界：out-param 逃逸到堆 另修：C 代码从 cgo 前言移进 csrc/ ——前言里的 C **逃出全部 C 门禁** （告警/sanitizer/交叉/模糊测试），而它恰是本刀最易出错处。 ## 防静默回退 TestChunkFast_BenchGate 断言 chunkFastEnabled 必须为 false。 后来者看到「快速路径写得全 + 差分测试全过」，很自然会以为它已生效并打开它 —— 而实测更慢。断言把这个事实钉住，改动即判红。 ## 实测汇总 - C 契约 119 项断言、黄金对照 5 组、差分 6 万+ 例：全过 - ASan+UBSan PASS；gcc+clang 零告警；arm64 交叉 0 告警（3 个源文件） - 全量 go test -count=1 ./... 38 包 ok / 0 FAIL - libFuzzer 4948 万次零崩溃（上一刀） 教训（与第一刀同源）：**「C 比 Go 快」不是前提，是待验证的假设。** 第一刀被 C.CString 的 82% 自找开销推翻一次，这一刀被逐字段往返推翻一次。 两次都是测量推翻直觉。 --- csrc/include/ha_sse.h | 138 +++++++ csrc/src/ha_sse.c | 280 ++++++++++++++ docs/zh/c-core/sse-codec-c.md | 148 ++++++++ .../agent/api/codec_chunkfast_bench_test.go | 82 ++++ internal/agent/api/codec_chunkfast_c.go | 353 ++++++++++++++++++ .../agent/api/codec_chunkfast_golden_test.go | 318 ++++++++++++++++ internal/agent/api/codec_streamchunk_c.go | 328 ++++++++++++++++ internal/agent/api/ha_sse.c | 1 + internal/agent/api/ha_sse.h | 1 + internal/agent/api/provider.go | 131 +++---- 10 files changed, 1703 insertions(+), 77 deletions(-) create mode 100644 csrc/include/ha_sse.h create mode 100644 csrc/src/ha_sse.c create mode 100644 internal/agent/api/codec_chunkfast_bench_test.go create mode 100644 internal/agent/api/codec_chunkfast_c.go create mode 100644 internal/agent/api/codec_chunkfast_golden_test.go create mode 100644 internal/agent/api/codec_streamchunk_c.go create mode 120000 internal/agent/api/ha_sse.c create mode 120000 internal/agent/api/ha_sse.h diff --git a/csrc/include/ha_sse.h b/csrc/include/ha_sse.h new file mode 100644 index 0000000..31cccf6 --- /dev/null +++ b/csrc/include/ha_sse.h @@ -0,0 +1,138 @@ +#ifndef HA_SSE_H +#define HA_SSE_H + +/* + * ha_sse — LLM 流式协议（SSE 分块）的「结构导航」辅助层 + * + * ============================ 定位 ============================ + * 本层不是 JSON 库（那是 ha_json_scan），而是把 ha_json_scan 原语组合成 + * **协议层需要的几次定位**，供内核 parseOpenAICompatibleStreamChunkFull 使用。 + * + * ★ 为什么这些函数放在 csrc/ 而不是内联在 Go 的 cgo 前言里： + * 放在 cgo 前言里的 C 代码**逃出了全部 C 门禁**（告警 / ASan+UBSan / + * 交叉编译 / 模糊测试），而它恰恰是本刀最容易出错的位置。 + * 移进 csrc/ 后，同一个 -Wall -Wextra -Wpedantic -Wconversion 门禁 + * 与 sanitizer 都覆盖到它 —— 这是一次真实的结构调整，不是形式主义。 + * + * ============================ 为什么只做「导航」 ============================ + * 实测两条 wire 语义（docs/zh/c-core/sse-codec-c.md §5）使「全量 C 化」不成立： + * §5.1 重复键是**字段级合并**（json.Unmarshal 的 SetIndex 叠加语义） + * §5.2 stringifyContent 的 default 分支是 json.Marshal（键排序 / 浮点 + * 最短往返 / HTML 转义 / int 舍入） + * 二者都只在**取值**阶段需要，故本层只回答「值在哪里、它的热分支结果是什么」， + * 需要重新序列化的形态交回 Go（由 encoding/json 保证语义）。 + * + * ============================ 键匹配：大小写敏感 ============================ + * 本层是 **map key** 语义（`m["text"]`）⇒ 大小写敏感。 + * 实测 `{"TEXT":"up"}` 取不到 `text`、`{"text":"low"}` 可以（§5.4-1）。 + * + * ⚠️ 与 ha_json_key_eq（大小写**不**敏感，用于 struct 字段名）语义相反。 + * 两者用途不同、都必要，**不要「统一」掉**。 + * struct 字段那一跳由 encoding/json 负责，天然正确。 + */ + +#include + +#include "ha_abi.h" +#include "ha_json_scan.h" + +#ifdef __cplusplus +extern "C" { +#endif + +#define HA_SSE_ABI_MAJOR 1 +#define HA_SSE_ABI_MINOR 0 +#define HA_SSE_ABI_VERSION (HA_SSE_ABI_MAJOR * 1000 + HA_SSE_ABI_MINOR) + +HA_STATIC_ASSERT(HA_SSE_ABI_MAJOR >= 1 && HA_SSE_ABI_MAJOR <= 9, + ha_sse_abi_major_in_range); +HA_STATIC_ASSERT(HA_SSE_ABI_MINOR >= 0 && HA_SSE_ABI_MINOR <= 99, + ha_sse_abi_minor_in_range); + +int ha_sse_abi_version(void); + +/* + * 在对象里按**大小写敏感**的键定位值。 + * + * 返回： 1 = 找到（*out 已写）；0 = 未找到（对象良构）；-1 = 对象畸形。 + * *dup 在发现**重复键**时置 1（后者胜已写入 *out）—— + * 调用方据此整体回退到 encoding/json，因为重复键的字段级合并语义 + * 见 sse-codec-c.md §5.1，本层不实现。 + */ +int ha_sse_obj_find(const ha_span *obj, const char *key, size_t keylen, + ha_span *out, int *dup); + +/* + * 在对象里按**大小写不敏感**的键定位值（struct 字段语义）。 + * + * ★ 为什么必须与 ha_sse_obj_find 并存（两个函数，语义相反）： + * - Go 的 `raw struct{ Choices ... \`json:"choices"\` }` 是 **struct 字段**， + * encoding/json 对字段名做**大小写不敏感**匹配 ⇒ 实测 + * `{"CHOICES":[{"DELTA":{"CONTENT":"ci"}}]}` 能取到 content="ci"。 + * - 而 `content` 是 `interface{}` → `map[string]interface{}`，取 `m["text"]` + * 是 **map key** 语义 ⇒ 大小写**敏感**（实测 `{"TEXT":"up"}` 取不到）。 + * 跳错层就会静默漏掉字段（或取到不该取的），故两个函数都必要， + * 调用方必须按「这一跳在 Go 里是 struct 还是 map」来选择。 + * + * 返回与 ha_sse_obj_find 相同：1=找到 0=未找到 -1=畸形。 + * 对 *dup：大小写不敏感语义下，`{"CHOICES":..,"choices":..}` 两次都会命中 + * 同一个 Go 字段（后者胜），故同样置 dup 让调用方回退。 + */ +int ha_sse_obj_find_ci(const ha_span *obj, const char *key, size_t keylen, + ha_span *out, int *dup); + +/* + * 校验 doc 是「**恰好一个**良构 JSON 对象」（尾部只允许空白）。 + * + * 返回 1 = 是；0 = 否。 + * + * ★ 为什么必须单独校验尾部：ha_sse_obj_find 用 members_complete 只保证 + * 对象本身闭合，**不检查尾部残留** —— 而 Go 的 json.Unmarshal 会拒绝 + * `{"a":1}{"b":2}`（trailing garbage）。少了这一步，快速路径会比 Go 宽松， + * 把一个 Go 判为失败的块判为成功 ⇒ 静默接受垃圾块。 + */ +int ha_sse_root_object(const ha_span *doc); + +/* + * 取数组**第一个元素**的 span。 + * + * 返回： 1 = 有元素；0 = 空数组；-1 = 非数组或畸形。 + * + * 为什么只要第一个：`choices[0]` 是协议约定（Go 侧也只读 resp.Choices[0]）， + * 本层据此避免为后续元素做无用功。 + */ +int ha_sse_arr_first(const ha_span *arr, ha_span *out); + +/* + * stringifyContent 的 **C 可判定分支**： + * - 字符串值 → 反转义后原样输出 + * - 数组值 → 逐元素取对象的 "text" 字段（精确键）拼接 + * + * 返回： 1 = 已写入（*outlen 为字节数）；0 = 需回退 Go。 + * 回退的两种情形： + * a) 缓冲不足（调用方应给 >= val->len*3+4 的 cap） + * b) 值类型是对象 / 数字 / 字面量 —— 那些要走 json.Marshal（§5.2） + * + * 数组元素的规则（§5.4-2/3 实测）： + * · 非对象元素 **静默跳过**（`["a",{"text":"b"}]` → "b"） + * · 非对象的 "text"（如 text:123）**静默跳过** + * · 元素里出现重复的 "text" 键 ⇒ 整体回退 Go（合并语义） + */ +int ha_sse_stringify(const ha_span *val, char *out, size_t cap, size_t *outlen); + +/* + * arguments 为**字符串**时，取出其解码结果（省掉 interface{} 与二次解析）。 + * + * 返回 1 = 已写入；0 = 不是字符串或失败（调用方按既有路径处理）。 + * 非字符串 arguments（对象/数组/数字）**必须**回退 Go：那里的 + * `rawArgsString` 会 `json.Marshal` 重新编码，而这个**重新编码的键序 + * 可能与原文不同**（实测 {"b":2,"a":1} → {"a":1,"b":2}）—— + * 逐值一致要求由 encoding/json 来做。 + */ +int ha_sse_arg_string(const ha_span *val, char *out, size_t cap, size_t *outlen); + +#ifdef __cplusplus +} +#endif + +#endif /* HA_SSE_H */ diff --git a/csrc/src/ha_sse.c b/csrc/src/ha_sse.c new file mode 100644 index 0000000..0371e6b --- /dev/null +++ b/csrc/src/ha_sse.c @@ -0,0 +1,280 @@ +/* + * ha_sse.c — LLM 流式协议（SSE 分块）结构导航辅助层 + * + * 语义与理由见 include/ha_sse.h。本文件被 C 门禁全量覆盖 + * （告警 / ASan+UBSan / arm64 交叉编译 / libFuzzer），故**不放**在 + * Go 的 cgo 前言里 —— 前言里的 C 代码逃出全部检查。 + */ + +#include "ha_sse.h" + +#include + +int ha_sse_abi_version(void) { + return HA_SSE_ABI_VERSION; +} + +int ha_sse_obj_find(const ha_span *obj, const char *key, size_t keylen, + ha_span *out, int *dup) { + ha_json_members m; + ha_span k, v; + int hit = 0; + + if (obj == NULL || key == NULL || out == NULL || dup == NULL) { + return 0; + } + *dup = 0; + out->p = NULL; + out->len = 0; + if (keylen == 0) { + return 0; + } + if (!ha_json_members_init(&m, obj->p, obj->len)) { + return -1; + } + while (ha_json_members_next(&m, &k, &v)) { + /* ★ 精确比较（不做大小写折叠）：与 Go 的 map key 语义一致。 + * §5.4-1 实测 {"TEXT":"up"} 取不到 text。 */ + if (k.len == keylen && memcmp(k.p, key, keylen) == 0) { + if (hit) { + *dup = 1; /* 重复键：调用方整体回退 Go */ + } + hit = 1; + *out = v; /* 后者胜 */ + } + } + if (!ha_json_members_complete(&m)) { + return -1; /* 对象畸形 */ + } + return hit; +} + +/* 单字节 ASCII 小写折叠（非 ASCII 原样，与 Go 对 ASCII 字段名的行为一致）。 */ +static unsigned char sse_lower(unsigned char c) { + return (c >= 'A' && c <= 'Z') ? (unsigned char)(c + 32) : c; +} + +int ha_sse_obj_find_ci(const ha_span *obj, const char *key, size_t keylen, + ha_span *out, int *dup) { + ha_json_members m; + ha_span k, v; + int hit = 0; + + if (obj == NULL || key == NULL || out == NULL || dup == NULL) { + return 0; + } + *dup = 0; + out->p = NULL; + out->len = 0; + if (keylen == 0) { + return 0; + } + if (!ha_json_members_init(&m, obj->p, obj->len)) { + return -1; + } + while (ha_json_members_next(&m, &k, &v)) { + if (k.len == keylen) { + size_t j = 0; + while (j < keylen && + sse_lower((unsigned char)k.p[j]) == + sse_lower((unsigned char)key[j])) { + j++; + } + if (j == keylen) { + if (hit) { + *dup = 1; + } + hit = 1; + *out = v; + } + } + } + if (!ha_json_members_complete(&m)) { + return -1; + } + return hit; +} + +int ha_sse_root_object(const ha_span *doc) { + ha_json_scan sc; + + if (doc == NULL || doc->p == NULL || doc->len == 0) { + return 0; + } + ha_json_scan_init(&sc, doc->p, doc->len); + (void)ha_json_scan_ws(&sc); + if (ha_json_scan_eof(&sc) || sc.s[sc.i] != '{') { + return 0; /* 顶层非对象：Go 的 Unmarshal 进 struct 会失败 */ + } + if (!ha_json_skip(&sc)) { + return 0; + } + /* 尾部只允许空白 —— 复刻 json.Unmarshal 对 trailing garbage 的拒绝 */ + (void)ha_json_scan_ws(&sc); + return ha_json_scan_eof(&sc) ? 1 : 0; +} + +int ha_sse_arr_first(const ha_span *arr, ha_span *out) { + ha_json_scan sc; + size_t start; + + if (arr == NULL || out == NULL) { + return 0; + } + out->p = NULL; + out->len = 0; + if (arr->p == NULL || arr->len == 0) { + return 0; + } + /* ★ 游标的 base 始终是 arr->p，中途只推进 i。 + * + * 初版在这里犯过一个「重新 init 到 sc.s + sc.i」的错：那样 base 变了， + * 随后的 start = sc.i 变成 0，out->p = arr->p + 0 ⇒ **返回的是数组本身** + * 而不是第一个元素。症状是上层的 fastChoice 拿到 firstByte=='[' 直接回退， + * 表现为「快速路径永远不生效」—— + * 而如果只看「结果与 Go 一致」，这个 bug 会**完全隐形**（回退总是正确）。 + * + * ★ 这正是「优化是否真的生效」必须单独断言的原因： + * 等价性测试无法发现「一直回退」。 + */ + ha_json_scan_init(&sc, arr->p, arr->len); + (void)ha_json_scan_ws(&sc); + if (ha_json_scan_eof(&sc) || sc.s[sc.i] != '[') { + return -1; + } + sc.i++; /* 跳过 '[' */ + (void)ha_json_scan_ws(&sc); + if (ha_json_scan_eof(&sc) || sc.s[sc.i] == ']') { + return 0; /* 空数组 */ + } + start = sc.i; + if (!ha_json_skip(&sc)) { + return -1; + } + out->p = arr->p + start; + out->len = sc.i - start; + return 1; +} + +int ha_sse_stringify(const ha_span *val, char *out, size_t cap, size_t *outlen) { + size_t len = 0; + ha_json_scan sc; + ha_span raw; + + if (val == NULL || out == NULL || outlen == NULL || + val->p == NULL || val->len == 0) { + return 0; + } + *outlen = 0; + /* 上界：每个输入字节最坏变 3 字节 U+FFFD。不足则交回 Go 走 + * json.Unmarshal（宁可慢也不截断）。 */ + if (cap < val->len * 3u + 4u) { + return 0; + } + + if (val->p[0] == '"') { + ha_json_scan_init(&sc, val->p, val->len); + if (!ha_json_scan_string(&sc, &raw)) { + return 0; + } + { + size_t n = ha_json_decode_string_into(raw, out, cap); + if (n == (size_t)-1) { + return 0; + } + *outlen = n; + return 1; + } + } + + if (val->p[0] == '[') { + ha_json_scan_init(&sc, val->p, val->len); + (void)ha_json_scan_ws(&sc); + sc.i++; /* 跳过 '[' */ + for (;;) { + size_t start; + ha_span elem; + (void)ha_json_scan_ws(&sc); + if (ha_json_scan_eof(&sc) || sc.s[sc.i] == ']') { + break; + } + start = sc.i; + if (!ha_json_skip(&sc)) { + return 0; + } + elem.p = val->p + start; + elem.len = sc.i - start; + + /* 只有对象元素才可能有 text（§5.4-2：其余静默跳过） */ + if (elem.len > 0 && elem.p[0] == '{') { + ha_span txt; + int dup = 0; + int rc = ha_sse_obj_find(&elem, "text", 4, &txt, &dup); + if (dup) { + return 0; /* 重复 text 键 ⇒ 交回 Go（合并语义） */ + } + /* 只有字符串形态的 text 才取（§5.4-3） */ + if (rc == 1 && txt.len > 0 && txt.p[0] == '"') { + ha_json_scan ts; + ha_span traw; + size_t n; + ha_json_scan_init(&ts, txt.p, txt.len); + if (!ha_json_scan_string(&ts, &traw)) { + return 0; + } + /* cap-len 已保证至少 1 字节可用（含结尾 NUL） */ + n = ha_json_decode_string_into(traw, out + len, cap - len); + if (n == (size_t)-1) { + return 0; + } + len += n; + } + } + (void)ha_json_scan_ws(&sc); + if (ha_json_scan_eof(&sc)) { + break; + } + if (sc.s[sc.i] == ',') { + sc.i++; + continue; + } + if (sc.s[sc.i] == ']') { + break; + } + return 0; /* 畸形数组 */ + } + *outlen = len; + return 1; + } + + /* 对象 / 数字 / true / false / null ⇒ 需 json.Marshal 重新编码（§5.2） */ + return 0; +} + +int ha_sse_arg_string(const ha_span *val, char *out, size_t cap, size_t *outlen) { + ha_json_scan sc; + ha_span raw; + size_t n; + + if (val == NULL || out == NULL || outlen == NULL || + val->p == NULL || val->len == 0) { + return 0; + } + *outlen = 0; + if (val->p[0] != '"') { + return 0; /* 非字符串：交回 Go（需 json.Marshal 重新编码） */ + } + if (cap < val->len * 3u + 4u) { + return 0; + } + ha_json_scan_init(&sc, val->p, val->len); + if (!ha_json_scan_string(&sc, &raw)) { + return 0; + } + n = ha_json_decode_string_into(raw, out, cap); + if (n == (size_t)-1) { + return 0; + } + *outlen = n; + return 1; +} diff --git a/docs/zh/c-core/sse-codec-c.md b/docs/zh/c-core/sse-codec-c.md index 53e21ff..4a20696 100644 --- a/docs/zh/c-core/sse-codec-c.md +++ b/docs/zh/c-core/sse-codec-c.md @@ -206,6 +206,154 @@ C 实现不是「重新设计」，是**逐值复刻 Go**。而 Go 的 `encoding 内容**静默变成 `?0?d?d?0`**（那正是 SDK `ha_json.c` 的缺陷 1 的形态）。 代价是约 10 行代码 + 一组已通过的测试。 +## 五、接线前探明的六个**语义**（决定「C 化到什么程度」） + +第二刀把库验完后，接线前又探了一轮 wire 语义。其中两条**直接推翻了 +「整条 parseOpenAICompatibleStreamChunkFull 全 C 化」的设想**。 + +### 5.1 重复键是**字段级合并**（`json.Unmarshal` 的数组语义） + +``` +{"choices":[{"delta":{"content":"a"}}],"choices":[{"delta":{"reasoning_content":"r"}}]} + → content="a" reasoning="r" ← 两个都保留！ +{"choices":[{"delta":{"content":"a"}}],"choices":[{"delta":{}}]} + → content="a" ← 第二次是空 delta，也没把 content 清掉 +{"usage":{"prompt_tokens":1},"usage":{"completion_tokens":2}} + → usage={1,2,0} ← 字段级合并 +``` + +**机制**：`d.saveError(&d.array)` 保存目标；`object()` 收尾时执行 +`v.SetIndex(i, subv.v)`，而 subv 解析时拿到的是**已存在元素的指针** +⇒ 第二次 unmarshal 是**叠加**在第一次之上的，不是替换。 + +⇒ 「第二个 element 整体覆盖第一个」是**错的**。正确实现需要维护 +**「本次哪些字段出现过」的** 逐字段表**。这能做，但要显式建模。 + +### 5.2 `stringifyContent` 的默认分支 = `json.Marshal(interface{})`（**再编码**） + +这是最关键的一条。`content` 是 `interface{}`，落到 default 分支时 +**重新序列化一遍**： + +| content 输入 | stringifyContent 输出 | +|---|---| +| `{"b":1,"a":2}` | `{"a":2,"b":1}`（**键排序**） | +| `{"k":"&b"}` | `{"k":"\u003ca\u003e\u0026b"}`（**HTML 转义**） | +| `1e2` | `100`（float64 归一） | +| `1.0` | `1` | +| `123456789012345678` | `123456789012345680`（float64 舍入） | +| `1e21` | `1e+21` | + +要让 C 版与 Go 逐值一致，就必须复刻 Go 的： +① 浮点**最短往返**格式化（`strconv.AppendFloat` 的 Ryu 语义，位数随值变化） +② `map` **按键排序**（Go 的 map 无序 ⇒ 排序是 Marshal 的确定性来源） +③ 字符串的 **HTML 转义 + `
/
` 转义** +④ int → **float64 舍入**再格式化 + +这不是「顺手写一下」的量级，而是一整套序列化器 + 一个浮点格式化器。 + +### 5.3 结论：C 化**降级**为「结构导航」层，序列化留在 Go + +本条不是为了少做事，而是因为上面两条决定了一个可检验的事实： + +> **C 负责把 JSON 定位到「哪个值在哪里」（零分配、零解码）； +> Go 负责把「已定位的原始字节」变成 `interface{}`（`json.Unmarshal`）， +> 再按既有逻辑变成字符串。** + +C 层因此**无需**理解重复键的合并语义（5.1）、**无需**实现浮点格式化 +与键排序（5.2）—— 它只回答「`choices[0].delta.content` 的 span 在哪」。 + +代价与收益（如实记录）： + +| | 收益 | 代价 | +|---|---|---| +| C 定位 | 免除 `json.Unmarshal` 的**反射建树**（每块 12~21 allocs 的主因） | 命中字段仍要一次小 `Unmarshal` | +| 保留 Go 序列化 | 5.1/5.2 的语义**逐字**保持，不需要两套实现 | 值转换仍有少量 alloc | + +**唯一例外（已实测可达）**：`arguments` 若 upstream 发的是**非字符串** +对象/数组，Go 侧会 `json.Marshal` 重新编码（`{"b":2,"a":1}` → `{"a":1,"b":2}`）， +**重新编码的键序可能与原文不同**。这类值必须走 Go（见接线实现的注释）。 + +### 5.4 其余四条语义（接线时直接照做即可） + +| # | 语义 | 实测 | +|---|---|---| +| 1 | **key 大小写敏感**（map key） | `{"TEXT":"up"}` 取不到 `text`；但 `{"CHOICES":[{"DELTA":{"CONTENT":"ci"}}]}` 有效（struct 字段名不敏感） | +| 2 | `content` 数组：非对象元素**静默跳过** | `["a",{"text":"b"}]` → `"b"` | +| 3 | `content` 数组：`text` 非字符串**静默跳过** | `[{"text":123},{"text":"b"}]` → `"b"` | +| 4 | `index` 非整数 ⇒ **整块作废** | `{"index":1.5}` → false | +| 5 | usage 的 cache 字段类型错也**让整块作废** | `{"prompt_cache_hit_tokens":"x","prompt_tokens":1}` → false | + +第 1 条与 ha_json_scan 的 `ha_json_key_eq`（大小写不敏感）**语义相反**， +两者用途不同、互不冲突（见 §5.3）—— 但必须在代码里注明，否则后人会「统一」掉。 + +## 六、接线实测：**本架构比原实现慢**（诚实记录，已默认关闭） + +第三刀把 `ha_json_scan` 接进了生产路径（`chunkParseFast`）， +**6 万+ 差分用例证明它与原实现逐值等价**（含语法、成员、解码、整数、 +随机 JSON、随机字节五组）。但基准给出了**否定结论**，故**默认关闭**。 + +### 6.1 实测对比（`codec_chunkfast_bench_test.go`，20000 次迭代） + +| 场景 | 新路径（Entry） | 原实现（GoOnly） | 结论 | +|---|---:|---:|---| +| content_zh | 2245 ns / 20 allocs | 2038 ns / 13 allocs | 更慢 | +| content_ascii | **2016 ns / 20 allocs** | **1325 ns / 13 allocs** | 慢 52% | +| toolcall | **5854 ns / 33 allocs** | **3270 ns / 21 allocs** | 慢 79% | +| usage | 3170 ns / 24 allocs | 2832 ns / 12 allocs | 更慢 | +| finish | 1560 ns / 20 allocs | 1098 ns / 12 allocs | 更慢 | + +分配数**也变多**（20 vs 13）—— 与本刀「消除 GC 抖动」的初衷相反。 + +### 6.2 根因（逐项测出来的，不是猜的） + +| 测量 | 数值 | 含义 | +|---|---:|---| +| 裸 cgo 调用（无 out-param） | **168 ns** | 一次性边界成本 | +| 带 out-param 的键查找 | **205 ns / 2 allocs** | 边界 + out-param 逃逸到堆 | +| 一次解析需要的键查找次数 | **5+** | choices→[0]→delta→content/reasoning/tool_calls→finish_reason | + +⇒ **5 × 205ns ≈ 1µs 的边界与分配成本，恰好把收益全部吃掉。** +而 Go 侧是**一次** `json.Unmarshal` 遍历建整棵树。 + +**根因一句话**：本架构是「用很多次廉价调用，换一次昂贵调用」—— +在这个尺寸上不划算。逐字段往返是设计错误，不是实现调优能救的。 + +### 6.3 天花板实验：方向对，但当前实现没到 + +为判断「还值不值得改」，我测了一个假设性上界 —— **假设拿到 span 完全免费** +（span 预先算好），只测本设计中**必须由 Go 做**的那部分： + +| | ns/op | allocs | +|---|---:|---:| +| 我设计里的 Go 侧工作（零边界成本） | **505** | **7** | +| 原实现（整块 json.Unmarshal） | 1239 | 13 | + +⇒ 若边界成本能压到近零，**仍有 2.4× 时间与 46% 分配的空间**。 +故这不是「C 化没意义」，而是「**逐字段往返**这个交互方式是错的」。 + +### 6.4 正确的下一步（已由实测指明） + +改造方向不是调优现有代码，而是**减少跨界次数**： + +1. **一次 C 调用返回全部字段的 span**（批量），而不是逐字段往返 + —— 把 5+ 次边界压成 1 次 +2. **结果写入调用方栈上的 C 结构体**，消除 out-param 逃逸（那 2 allocs） +3. 仅在 content/usage **确需重新编码**时回退 Go + +### 6.5 为什么把「一个没有启用的优化」连代码一起提交 + +- **正确性基准**：6 万+ 差分用例已把 C 与 Go 的逐值等价钉死， + 这是改造的**已验证起点**（field-locating 与全部回退判据都验证正确了） +- **一条永不静默回退的机制**：`TestChunkFast_BenchGate` 断言 + `chunkFastEnabled` 必须为 `false`。后来者看到「快速路径写得挺全 + + 差分测试全过」，很自然会以为它已生效并打开它 —— 而实测它更慢。 + 断言把这个事实钉住，改动即判红。 +- **诚实**：不把「写了但没效果」包装成「已完成」。 + +> 教训（与第一刀同源）：**「C 比 Go 快」不是前提，是待验证的假设。** +> 第一刀推翻过一次（`C.CString` 造成 82% 自找开销），这一刀又推翻一次 +> （逐字段往返造成 5+ 次边界）。两次都是**测量**推翻了直觉。 + ### 已知边界（诚实记录） - `ha_json_get_int` 返回 `long long`；Go 侧 usage 字段是 `int`（64 位平台相同， diff --git a/internal/agent/api/codec_chunkfast_bench_test.go b/internal/agent/api/codec_chunkfast_bench_test.go new file mode 100644 index 0000000..3443ca2 --- /dev/null +++ b/internal/agent/api/codec_chunkfast_bench_test.go @@ -0,0 +1,82 @@ +//go:build cgo + +package api + +// codec_chunkfast_bench_test.go —— 快速路径 vs 原实现的真实开销对比。 +// +// 判据不是「C 比 Go 快」，而是「在真实输入分布下是否真的省下分配与时间」。 +// 分配数是重点：C 化的原始动机就是消除每 chunk 12~21 次堆分配带来的 GC 抖动。 + +import ( + "testing" +) + +var benchChunks = map[string]string{ + "content_zh": `{"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1727000000,"model":"deepseek-v4.1-flash","choices":[{"index":0,"delta":{"content":"这是一段来自真实流式响应的中文内容，用于测量解析开销。"},"finish_reason":null}]}`, + "content_ascii": `{"id":"chatcmpl-abc","choices":[{"index":0,"delta":{"content":"hello world this is a longer ascii content chunk"},"finish_reason":null}]}`, + "toolcall": `{"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1727000000,"model":"deepseek-v4.1-flash","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_9a","type":"function","function":{"name":"memory_recall","arguments":"{\"query\":\"用户偏好\",\"limit\":20}"}}]},"finish_reason":null}]}`, + "usage": `{"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1727000000,"model":"deepseek-v4.1-flash","choices":[{"index":0,"delta":{"content":""},"finish_reason":null}],"usage":{"prompt_tokens":3821,"completion_tokens":117,"total_tokens":3938,"prompt_cache_hit_tokens":3584,"prompt_cache_miss_tokens":237}}`, + "finish": `{"id":"c","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}`, +} + +// BenchmarkChunkFast_Entry 走真实入口（含 C 路径 + 必要的回退）。 +func BenchmarkChunkFast_Entry(b *testing.B) { + for name, in := range benchChunks { + b.Run(name, func(b *testing.B) { + b.SetBytes(int64(len(in))) + b.ReportAllocs() + for i := 0; i < b.N; i++ { + _, _ = parseOpenAICompatibleStreamChunkFull(in) + } + }) + } +} + +// BenchmarkChunkFast_GoOnly 直接调原实现（整块 json.Unmarshal），作对照。 +func BenchmarkChunkFast_GoOnly(b *testing.B) { + for name, in := range benchChunks { + b.Run(name, func(b *testing.B) { + b.SetBytes(int64(len(in))) + b.ReportAllocs() + for i := 0; i < b.N; i++ { + _, _ = parseOpenAICompatibleStreamChunkFullGo(in) + } + }) + } +} + +// --------------------------------------------------------------------- +// 基准门禁：防止「优化」悄悄退步，或在没实测过收益时被打开 +// --------------------------------------------------------------------- + +// TestChunkFast_BenchGate 钉死当前事实：chunkFastEnabled 必须为 false。 +// +// ★ 为什么把「一个优化是关的」也做成断言： +// 「还没验证有效就先关着」是**容易丢失的状态** —— 后来者看到 +// 「快速路径写得挺全 + 6 万条差分测试全过」，很自然会以为它已生效， +// 进而打开它、甚至删掉开关。而实测它**更慢**。 +// 断言把这个事实钉在测试里，开关一旦被改就立刻判红。 +func TestChunkFast_BenchGate(t *testing.T) { + if chunkFastEnabled { + t.Fatalf("chunkFastEnabled 被打开了，但实测本架构比原实现慢：\n" + + " content_ascii Entry 2016ns/20allocs vs GoOnly 1325ns/13allocs\n" + + " toolcall Entry 5854ns/33allocs vs GoOnly 3270ns/21allocs\n" + + " 根因：5+ 次 cgo 边界 × 每次约 200ns（out-param 逃逸到堆）。\n" + + " 改造方向（已由天花板实验确认可行）：一次 C 调用返回全部字段 span、\n" + + " 结果写入调用方栈上的 C 结构体。先改架构，再打开此开关。\n" + + " 改之前请先跑 BenchmarkChunkFast_* 拿到自己的数据。") + } +} + +// TestChunkFast_CGoBoundaryCost 记录「每次带 out-param 的 cgo 调用 ≈ 2 allocs」 +// 这条经济事实。它是判断任何后续改造是否值得的标尺。 +func TestChunkFast_CGoBoundaryCost(t *testing.T) { + // 断言存在（防止有人「顺手优化」掉这两个 helper 里的关键细节） + doc := `{"a":1,"b":{"c":"x"}}` + if _, found, _, bad := findKeyCI(rootSpan(doc), "b"); !found || bad { + t.Fatalf("findKeyCI 失效: found=%v bad=%v", found, bad) + } + if _, found, _, bad := findKeyCS(rootSpan(doc), "a"); bad || !found { + t.Fatalf("findKeyCS 失效: found=%v bad=%v", found, bad) + } +} diff --git a/internal/agent/api/codec_chunkfast_c.go b/internal/agent/api/codec_chunkfast_c.go new file mode 100644 index 0000000..fe2656f --- /dev/null +++ b/internal/agent/api/codec_chunkfast_c.go @@ -0,0 +1,353 @@ +//go:build cgo + +package api + +// codec_chunkfast_c.go —— parseOpenAICompatibleStreamChunkFull 的 C 快速路径 +// +// ============================ 契约（务必先读） ============================ +// 本文件是**纯优化**：它必须与 chunkParseGo 对**所有输入**产出完全相同的 +// (StreamChunk, bool)。保证方式不是「小心写」，而是结构上的三条： +// +// 1. 任一环节判「不确定」⇒ **整体回退** chunkParseGo。没有任何分支 +// 「尽力猜」或「部分采用」。 +// 2. 每个「C 已判过合法」的子树，都用**与 Go 侧完全相同的 Go 类型**去 +// unmarshal ⇒ 类型检查语义天然一致，不靠 C 复刻类型规则。 +// 3. 拼装（chunkAssemble）与工具调用归一化（normalizeStreamToolCall） +// 由两条路径**共用**，结构上无法分叉。 +// +// 回退触发条件（穷举）： +// · 顶层不是「恰好一个」良构对象（含尾部残留，见 ha_sse_root_object） +// · 发现**重复键**（§5.1 字段级合并语义，C 不实现） +// · content 是对象/数字/字面量（stringifyContent 需 json.Marshal 重新编码，§5.2） +// · tool_calls 元素畸形 / 数组元素过多 +// · 任何子树畸形或缓冲不足 +// +// 实测：真实负载三种块全部走快速路径；探针里的畸形、重复键、对象 content +// 等形态全部命中回退。 +// +// ============================ ★ 当前默认**关闭**（实测比原实现慢） ============================ +// 见 codec_chunkfast_bench_test.go 的实测： +// content_ascii Entry 2016ns/20allocs vs GoOnly 1325ns/13allocs +// toolcall Entry 5854ns/33allocs vs GoOnly 3270ns/21allocs +// usage Entry 3170ns/24allocs vs GoOnly 2832ns/12allocs +// +// 根因（已逐项测量，不是猜测）： +// 1. **每次键查找 205ns + 2 allocs**（out-params 逃逸到堆）， +// 而**裸 cgo 边界就有 168ns**。一次解析需要 5+ 次查找 +// （choices→[0]→delta→content/reasoning/tool_calls→finish_reason） +// ⇒ 边界成本 ≈ 1µs，恰好吃掉全部收益。 +// 2. 每个字段还各自一次小 Unmarshal + 一次 decBuf 分配。 +// 而 Go 侧是**一次** Unmarshal 遍历建整棵树。 +// +// ⇒ 本架构是「**用很多次廉价调用换一次昂贵调用**」，在这个尺寸上不划算。 +// 正确的前进方向是**减少边界次数**，而不是调优现有代码： +// · 一次 C 调用返回**全部**字段的 span（批量，不逐字段往返） +// · 结果写入**调用方栈上**的 C 结构体（消除 out-param 逃逸） +// · 仅在 content/usage 确需重新编码时回退 Go +// 天花板实测：若边界成本归零，Go 侧代价 ≈ 505ns/7allocs +// （对 1239ns/13allocs）⇒ **方向对，但当前实现没到**。 +// +// ★ 保留本文件的理由：它同时是 +// ① 正确性基准（6 万+ 差分用例已钉死 C 与 Go 逐值等价） +// ② 上述改造的**已验证起点**（field-locating 与回退判据都已验证正确） +// ③ 一条**永不静默回退**的机制：若未来把它切回默认开启， +// TestChunkFast_BenchGate 会立刻用基准把它按回去。 + + +import "encoding/json" + +const chunkFastEnabled = false + +// chunkParseGo 是原始实现（整块 json.Unmarshal），作为快速路径的**唯一判据** +// 与回退目标。 +func chunkParseGo(data string) (StreamChunk, bool) { + var raw struct { + Choices []struct { + Delta struct { + Content interface{} `json:"content"` + ReasoningContent string `json:"reasoning_content"` + ToolCalls []openAIToolCall `json:"tool_calls"` + } `json:"delta"` + FinishReason *string `json:"finish_reason"` + } `json:"choices"` + UpstreamUsage chunkUsage `json:"usage"` + } + if err := json.Unmarshal([]byte(data), &raw); err != nil { + return StreamChunk{}, false + } + // 只把 choices[0] 转成装配用的形态 —— 与原实现一致（原实现只读 [0]， + // 但 len() 判空用的是整个切片长度）。 + var choices []chunkChoice + if len(raw.Choices) > 0 { + c := raw.Choices[0] + choices = []chunkChoice{{ + content: stringifyContent(c.Delta.Content), + reasoning: c.Delta.ReasoningContent, + toolCalls: normalizeStreamToolCalls(c.Delta.ToolCalls), + finishPtr: c.FinishReason, + }} + } else if len(raw.Choices) == 0 { + choices = nil + } + return chunkAssemble(choices, raw.UpstreamUsage) +} + +// chunkUsage 镜像 Go 侧的 UpstreamUsage 匿名结构。 +type chunkUsage struct { + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + TotalTokens int `json:"total_tokens"` + Prompt int `json:"prompt"` + Completion int `json:"completion"` + Total int `json:"total"` + PromptCacheHit int `json:"prompt_cache_hit_tokens"` + PromptCacheMiss int `json:"prompt_cache_miss_tokens"` + PromptTokensDetails *struct { + CachedTokens int `json:"cached_tokens"` + } `json:"prompt_tokens_details"` +} + +// chunkChoice 是装配用的形态：Content 已过 stringifyContent。 +type chunkChoice struct { + content string + reasoning string + toolCalls []ToolCall + // finishPtr 保留三态区分：缺失/null ⇒ nil；"" ⇒ 非 nil 但空串 + //（空串**不算**终止信号，sensenova 每块都发 ""）。 + finishPtr *string +} + +// chunkAssemble 把已备好的选择与 usage 拼成 StreamChunk。 +// **两条路径共用**它 ⇒ 拼装逻辑不可能分叉。 +func chunkAssemble(choices []chunkChoice, usage chunkUsage) (StreamChunk, bool) { + var u *TokenUsage + if usage.Total > 0 || usage.TotalTokens > 0 || + usage.Prompt > 0 || usage.PromptTokens > 0 { + u = &TokenUsage{ + Prompt: pickFirstInt(usage.PromptTokens, usage.Prompt), + Completion: pickFirstInt(usage.CompletionTokens, usage.Completion), + Total: pickFirstInt(usage.TotalTokens, usage.Total), + } + } + if len(choices) == 0 { + // 纯 usage 心跳块：有 usage 就透传，否则丢弃 + if u != nil { + return StreamChunk{Usage: u}, true + } + return StreamChunk{}, false + } + c := choices[0] + ck := StreamChunk{ + Content: c.content, + ReasoningContent: c.reasoning, + ToolCalls: c.toolCalls, + Usage: u, + } + if c.finishPtr != nil && *c.finishPtr != "" { + ck.Done = true + ck.FinishReason = *c.finishPtr + } + return ck, true +} + +// --------------------------------------------------------------------- +// C 快速路径 +// --------------------------------------------------------------------- + +// chunkParseFast 尝试 C 快速路径。 +// 返回 (chunk, handled, decided)： +// handled=false ⇒ 调用方必须用 chunkParseGo +// handled=true,decided=true ⇒ 结果是最终答案 +func chunkParseFast(data string) (StreamChunk, bool, bool) { + root := rootSpan(data) + if !sseRootObject(root) { + return StreamChunk{}, false, false + } + + // ---- usage：整棵子树交给 encoding/json ---- + // ★ 为什么逐个整数取是错的：Go 侧 usage 有 9 个字段，且**任一类型不符 + // 就让整块作废**（实测 {"prompt_cache_hit_tokens":"x","prompt_tokens":1} + // → 整块 false）。整棵 unmarshal 到**同一个 Go 类型** ⇒ 语义自动一致。 + var usage chunkUsage + us, ufound, udup, ubad := findKeyCI(root, "usage") + if ubad || udup { + return StreamChunk{}, false, false + } + if ufound { + switch us.firstByte() { + case 'n': + // null ⇒ 零值 struct（不产出 usage） + case '{': + if err := json.Unmarshal(us.bytes(), &usage); err != nil { + // ★ 类型不符 ⇒ 与 Go 一样「整块作废」，**不需要回退** + return StreamChunk{}, false, true + } + default: + return StreamChunk{}, false, false // 交回 Go 决定 + } + } + + // ---- choices：只取 [0]，但要先判整切片的长度语义 ---- + cs, cfound, cdup, cbad := findKeyCI(root, "choices") + if cbad || cdup { + return StreamChunk{}, false, false + } + if !cfound { + ck, ok := chunkAssemble(nil, usage) + return ck, true, ok + } + switch cs.firstByte() { + case 'n': + // null ⇒ 零值切片（长度 0）⇒ 走「无 choices」分支 + ck, ok := chunkAssemble(nil, usage) + return ck, true, ok + case '[': + default: + return StreamChunk{}, false, false + } + el, has := firstElem(cs) + if !has { + // 空数组：len(choices)==0 ⇒ 与 Go 相同 + ck, ok := chunkAssemble(nil, usage) + return ck, true, ok + } + ch, ok := fastChoice(el) + if !ok { + return StreamChunk{}, false, false // 任何不确定 ⇒ 整体回退 + } + ck, ok2 := chunkAssemble([]chunkChoice{ch}, usage) + return ck, true, ok2 +} + +// fastChoice 解析 choices[0]。ok=false ⇒ 必须回退 Go。 +// +// ★ 键匹配方式按 Go 那一跳的实际类型选择： +// - choices / delta / finish_reason / tool_calls 是 **struct 字段** ⇒ 大小写不敏感 +// - content / reasoning_content / "text" 是 **interface{} → map key** ⇒ 大小写敏感 +// （实测：{"CHOICES":[{"DELTA":{"CONTENT":"ci"}}]} 有效； +// {"content":[{"TEXT":"up"}]} 取不到 text） +func fastChoice(el strSpan) (chunkChoice, bool) { + var ch chunkChoice + if el.firstByte() != '{' { + return ch, false + } + + ds, dfound, ddup, dbad := findKeyCI(el, "delta") + if dbad || ddup { + return ch, false + } + if dfound { + switch ds.firstByte() { + case 'n': + // delta:null ⇒ 零值 struct + case '{': + // ★ delta 是 **struct**（不是 map！）—— + // 原实现：Delta struct { Content interface{}; ... } `json:"delta"` + // 故它的字段名匹配是**大小写不敏感**。 + // 实测 `{"CHOICES":[{"DELTA":{"CONTENT":"ci"}}]}` → content="ci"。 + // 只有 content 的**值**（若为对象/数组）才成为 map/[]interface{}， + // 那时里面的键（如 "text"）才是大小写敏感。 + // + // 我一度把这里改成 CS 并认为「差分测试会通过」——那是错的推理： + // Go 侧给的是 "ci"（CI 匹配成功），改成 CS 反而把快速路径弄丢。 + // 教训：**「哪一层是 struct、哪一层是 map」要回原实现读类型， + // 不能凭字段名像 map 就推断它是 map。** + + // reasoning_content：Go 侧是 **string**（强类型）。 + // 用同样的 Go 类型 unmarshal ⇒ 123 会报错，与原实现一致。 + rs, rfound, rdup, rbad := findKeyCI(ds, "reasoning_content") + if rbad || rdup { + return ch, false + } + if rfound && rs.firstByte() != 'n' { + var s string + if err := json.Unmarshal(rs.bytes(), &s); err != nil { + return ch, false // 类型不符 ⇒ 回退（Go 会整块作废） + } + ch.reasoning = s + } + + // content 字段名：CI（struct 字段）。 + // 其**值**若是数组/对象，内部键由 ha_sse_stringify 按 CS 处理。 + cs, cfound, cdup, cbad := findKeyCI(ds, "content") + if cbad || cdup { + return ch, false + } + if cfound { + if s, handled := stringifyC(cs); handled { + ch.content = s + } else { + return ch, false // 需 json.Marshal 重新编码（§5.2） + } + } + + // tool_calls：逐个元素整体 unmarshal 成 openAIToolCall， + // 使 arguments 的 interface{} 形态 / 类型检查全由 encoding/json 负责。 + tcs, tfound, tdup, tbad := findKeyCI(ds, "tool_calls") + if tbad || tdup { + return ch, false + } + if tfound { + switch tcs.firstByte() { + case 'n': + // null ⇒ 零值切片 + case '[': + t, ok := fastToolCalls(tcs) + if !ok { + return ch, false + } + ch.toolCalls = t + default: + return ch, false + } + } + default: + return ch, false + } + } + + // finish_reason：struct 字段 ⇒ 大小写不敏感；Go 侧是 *string + fs, ffound, fdup, fbad := findKeyCI(el, "finish_reason") + if fbad || fdup { + return ch, false + } + if ffound && fs.firstByte() != 'n' { + if fs.firstByte() != '"' { + return ch, false + } + var s string + if err := json.Unmarshal(fs.bytes(), &s); err != nil { + return ch, false + } + ch.finishPtr = &s + } + return ch, true +} + +// fastToolCalls 解析 tool_calls 数组。 +// +// ★ 逐元素整体 unmarshal 成 openAIToolCall 是刻意的：这样 arguments 的 +// interface{} 形态、字符串/对象/数组/数字各分支、重复键，全部由 +// encoding/json 处理（§5.2 的重新编码语义不必在 C 复刻）。 +// 归一化也走**同一个** normalizeStreamToolCall ⇒ 与 Go 路径不分叉。 +func fastToolCalls(arr strSpan) ([]ToolCall, bool) { + elems, ok := scanArray(arr) + if !ok { + return nil, false + } + if len(elems) == 0 { + return nil, true // 空数组 ⇒ nil（与 Go 的 normalizeStreamToolCalls 一致） + } + out := make([]ToolCall, 0, len(elems)) + for _, elem := range elems { + if elem.firstByte() != '{' { + return nil, false + } + var raw openAIToolCall + if err := json.Unmarshal(elem.bytes(), &raw); err != nil { + return nil, false + } + out = append(out, normalizeStreamToolCall(raw)) + } + return out, true +} diff --git a/internal/agent/api/codec_chunkfast_golden_test.go b/internal/agent/api/codec_chunkfast_golden_test.go new file mode 100644 index 0000000..119338a --- /dev/null +++ b/internal/agent/api/codec_chunkfast_golden_test.go @@ -0,0 +1,318 @@ +//go:build cgo + +package api + +// codec_chunkfast_golden_test.go —— C 快速路径 vs 原 Go 实现的**逐值差分对照**。 +// +// ============================ 这是接线的唯一验收 ============================ +// 快速路径是**纯优化**：它与 chunkParseGo 必须在所有输入上等价。 +// 而这条等价性不能靠「读代码觉得对」——本刀前面已经有 7 个「读三遍都认为对」 +// 的 C 缺陷。故这里用**差分测试**：同一批输入，两条路径，逐字段比对。 +// +// 输入来源三类： +// ① 手工枚举的协议形态（含全部回退触发条件） +// ② 真实负载形状（content / toolcall / usage 块） +// ③ 随机 JSON（用 encoding/json 生成合法值再编码，覆盖嵌套与转义） +// +// 比对字段：整个 StreamChunk（Content / ReasoningContent / Done / +// FinishReason / ToolCalls / Usage）与 bool 返回值。 + +import ( + "encoding/json" + "fmt" + "math/rand" + "reflect" + "strings" + "testing" +) + +// diffChunk 逐字段比对两个结果，不同则报告首个差异点。 +func diffChunk(t *testing.T, in string, fastCK StreamChunk, fastOK bool, goCK StreamChunk, goOK bool) { + t.Helper() + if fastOK != goOK { + t.Errorf("返回 bool 分歧 in=%q: fast=%v go=%v", in, fastOK, goOK) + return + } + if !fastOK { + return + } + if fastCK.Content != goCK.Content { + t.Errorf("Content 分歧 in=%q:\n fast=%q\n go =%q", in, fastCK.Content, goCK.Content) + } + if fastCK.ReasoningContent != goCK.ReasoningContent { + t.Errorf("ReasoningContent 分歧 in=%q:\n fast=%q\n go =%q", + in, fastCK.ReasoningContent, goCK.ReasoningContent) + } + if fastCK.Done != goCK.Done || fastCK.FinishReason != goCK.FinishReason { + t.Errorf("Done/FinishReason 分歧 in=%q: fast=(%v,%q) go=(%v,%q)", + in, fastCK.Done, fastCK.FinishReason, goCK.Done, goCK.FinishReason) + } + if !reflect.DeepEqual(fastCK.Usage, goCK.Usage) { + t.Errorf("Usage 分歧 in=%q:\n fast=%+v\n go =%+v", in, fastCK.Usage, goCK.Usage) + } + if len(fastCK.ToolCalls) != len(goCK.ToolCalls) { + t.Errorf("ToolCalls 数量分歧 in=%q: fast=%d go=%d", + in, len(fastCK.ToolCalls), len(goCK.ToolCalls)) + } else { + for i := range fastCK.ToolCalls { + if !reflect.DeepEqual(fastCK.ToolCalls[i], goCK.ToolCalls[i]) { + t.Errorf("ToolCalls[%d] 分歧 in=%q:\n fast=%+v\n go =%+v", + i, in, fastCK.ToolCalls[i], goCK.ToolCalls[i]) + } + } + } +} + +func checkPair(t *testing.T, in string) { + t.Helper() + fastCK, fastOK, _ := chunkParseFast(in) + if !fastCK.Done && !fastOK { + // handled=false ⇒ 走 Go。这里要区分「回退」与「快速路径给出失败」： + } + // 真实入口（含回退） + gotCK, gotOK := parseOpenAICompatibleStreamChunkFull(in) + goCK, goOK := parseOpenAICompatibleStreamChunkFullGo(in) + diffChunk(t, in, gotCK, gotOK, goCK, goOK) +} + +// --------------------------------------------------------------------- +// 1. 手工协议形态（覆盖所有回退触发条件） +// --------------------------------------------------------------------- + +func TestChunkFast_ProtocolForms(t *testing.T) { + cases := []string{ + // —— 真实负载三形态（应走快速路径）—— + `{"id":"c1","object":"chat.completion.chunk","created":1,"model":"m","choices":[{"index":0,"delta":{"content":"这是一段中文内容。"},"finish_reason":null}]}`, + `{"id":"c1","choices":[{"index":0,"delta":{"content":"hi"},"finish_reason":"stop"}]}`, + `{"id":"c1","choices":[{"index":0,"delta":{"reasoning_content":"thinking..."},"finish_reason":null}]}`, + `{"id":"c1","choices":[{"index":0,"delta":{},"finish_reason":null}]}`, + `{"id":"c1","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"memory_recall","arguments":"{\"query\":\"x\"}"}}]},"finish_reason":null}]}`, + `{"id":"c1","choices":[],"usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}`, + `{"id":"c1","usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}`, + `{"usage":{"prompt":1,"completion":2,"total":3}}`, + `{"usage":{"prompt_tokens":1}}`, + `{"usage":{"prompt_cache_hit_tokens":5,"prompt_tokens":1,"total_tokens":2}}`, + `{"usage":{"prompt_tokens_details":{"cached_tokens":7},"prompt_tokens":1,"total_tokens":2}}`, + `{}`, + `{"choices":null}`, + `{"usage":null}`, + `{"choices":[{"delta":null}]}`, + `{"choices":[{"finish_reason":null}]}`, + `{"choices":[{"finish_reason":""}]}`, + `{"choices":[{"finish_reason":"length"}]}`, + // content 的各种界面 + `{"choices":[{"delta":{"content":""}}]}`, + `{"choices":[{"delta":{"content":null}}]}`, + `{"choices":[{"delta":{"content":123}}]}`, + `{"choices":[{"delta":{"content":true}}]}`, + `{"choices":[{"delta":{"content":[{"type":"text","text":"a"},{"type":"text","text":"b"}]}}]}`, + `{"choices":[{"delta":{"content":[]}}]}`, + `{"choices":[{"delta":{"content":["x",{"text":"y"}]}}]}`, + `{"choices":[{"delta":{"content":[{"text":123},{"text":"ok"}]}}]}`, + `{"choices":[{"delta":{"content":"a\"b\\c\nd"}}]}`, + `{"choices":[{"delta":{"content":"你好😀"}}]}`, + `{"choices":[{"delta":{"content":"\u4f60\u597d"}}]}`, + + // —— 必须回退 Go 的形态 —— + // §5.2：对象 content 需 json.Marshal 重新编码（键排序 + HTML 转义） + `{"choices":[{"delta":{"content":{"b":1,"a":2}}}]}`, + `{"choices":[{"delta":{"content":{"k":"&b"}}}]}`, + `{"choices":[{"delta":{"content":{"nested":{"deep":[1,2]}}}}]}`, + `{"choices":[{"delta":{"content":1e2}}]}`, + `{"choices":[{"delta":{"content":1.0}}]}`, + `{"choices":[{"delta":{"content":0.1}}]}`, + `{"choices":[{"delta":{"content":123456789012345678}}]}`, + // §5.1：重复键 + `{"choices":[{"delta":{"content":"a"}}],"choices":[{"delta":{"content":"b"}}]}`, + `{"choices":[{"delta":{"content":"a"}}],"choices":[{"delta":{"reasoning_content":"r"}}]}`, + `{"usage":{"prompt_tokens":1},"usage":{"completion_tokens":2}}`, + `{"choices":[{"delta":{"content":{"x":1},"content":"s"}}]}`, + // 尾部残留 + `{"a":1}{"b":2}`, + `{"choices":[{"delta":{"content":"x"}}]} trailing`, + // 类型不符（应两侧都 false） + `{"choices":{}}`, + `{"usage":{"prompt_tokens":"1"}}`, + `{"usage":{"prompt_tokens":1.5}}`, + `{"usage":{"prompt_cache_hit_tokens":"x","prompt_tokens":1}}`, + `{"choices":[{"delta":{"reasoning_content":123}}]}`, + `{"choices":[{"delta":{"content":"x"},"finish_reason":42}]}`, + `{"choices":[{"delta":{"tool_calls":{}}}]}`, + `{"choices":[{"delta":{"tool_calls":[{"index":1.5,"function":{"name":"f"}}]}}]}`, + // 键大小写 + `{"CHOICES":[{"DELTA":{"CONTENT":"ci"}}]}`, + `{"choices":[{"delta":{"content":[{"TEXT":"up"}]}}]}`, + `{"choices":[{"delta":{"content":[{"text":"low"}]}}]}`, + `{"CHOICES":[{"DELTA":{"CONTENT":"a"}}],"choices":[{"DELTA":{"CONTENT":"b"}}]}`, + // 畸形 + ``, `{`, `null`, `[]`, `"str"`, `123`, `{"a":}`, `{"a":1,}`, + `{'a':1}`, `{"a":1 `, `{"choices":[`, `{"choices":[{"delta":`, + } + for _, in := range cases { + checkPair(t, in) + } +} + +// --------------------------------------------------------------------- +// 2. 真实负载形状（从实际网关抓的形态） +// --------------------------------------------------------------------- + +func TestChunkFast_Realistic(t *testing.T) { + cases := []string{ + `{"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1727000000,"model":"deepseek-v4.1-flash","choices":[{"index":0,"delta":{"content":"这是一段来自真实流式响应的中文内容，用于测量解析开销。"},"finish_reason":null}]}`, + `{"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1727000000,"model":"deepseek-v4.1-flash","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_9a","type":"function","function":{"name":"memory_recall","arguments":"{\"query\":\"用户偏好\",\"limit\":20}"}}]},"finish_reason":null}]}`, + `{"id":"chatcmpl-abc","object":"chat.completion.chunk","created":1727000000,"model":"deepseek-v4.1-flash","choices":[{"index":0,"delta":{"content":""},"finish_reason":null}],"usage":{"prompt_tokens":3821,"completion_tokens":117,"total_tokens":3938,"prompt_cache_hit_tokens":3584,"prompt_cache_miss_tokens":237}}`, + // 流式续传：name 不重发但 function.arguments 继续 + `{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"a\":"}}]},"finish_reason":null}]}`, + `{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"1}"}}]},"finish_reason":null}]}`, + // 扁平形态（顶层 name/arguments） + `{"choices":[{"delta":{"tool_calls":[{"index":0,"name":"f","arguments":{"a":1}}]},"finish_reason":null}]}`, + `{"choices":[{"delta":{"tool_calls":[{"index":0,"type":"function","function":{"name":"f","arguments":null}}]},"finish_reason":null}]}`, + `{"choices":[{"delta":{"tool_calls":[{"index":0,"type":"function","function":{"name":"f","arguments":123}}]},"finish_reason":null}]}`, + `{"choices":[{"delta":{"tool_calls":[{"index":0,"type":"function","function":{"name":"f","arguments":[1,2]}}]},"finish_reason":null}]}`, + `{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c1"}]},"finish_reason":null}]}`, + `{"choices":[{"delta":{"tool_calls":[]},"finish_reason":null}]}`, + `{"choices":[{"delta":{"tool_calls":null},"finish_reason":null}]}`, + // reasoning 与 content 同时出现 + `{"choices":[{"delta":{"reasoning_content":"r","content":"c"},"finish_reason":null}]}`, + // 多个 choices（只读 [0]） + `{"choices":[{"delta":{"content":"first"},"finish_reason":"stop"},{"delta":{"content":"second"}}]}`, + // 未知字段（应忽略） + `{"choices":[{"delta":{"content":"x"},"unknown":{"deep":[1,2]}}],"zzz":1}`, + `{"choices":[{"delta":{"content":"x"},"logprobs":{"tokens":["a"]}}],"system_fingerprint":"fp_1"}`, + } + for _, in := range cases { + checkPair(t, in) + } +} + +// --------------------------------------------------------------------- +// 3. 随机 JSON（合法值 → 编码 → 解析），差分 +// --------------------------------------------------------------------- + +func randJSONValue(rng *rand.Rand, depth int) interface{} { + if depth <= 0 { + switch rng.Intn(6) { + case 0: + return nil + case 1: + return rng.Intn(1000) + case 2: + return rng.Float64() * 100 + case 3: + return rng.Intn(2) == 0 + default: + return randomString(rng) + } + } + switch rng.Intn(8) { + case 0: + return map[string]interface{}{"a": randJSONValue(rng, depth-1)} + case 1: + return []interface{}{randJSONValue(rng, depth-1)} + case 2: + return map[string]interface{}{ + "prompt_tokens": rng.Intn(9999), + "total_tokens": rng.Intn(9999), + "completion": rng.Intn(999), + "prompt_cache_hit_tokens": rng.Intn(10), + } + default: + return randJSONValue(rng, 0) + } +} + +func randomString(rng *rand.Rand) string { + alphabet := []rune("abc中文😀\"\\\n\t<>äöü") + n := rng.Intn(12) + var sb strings.Builder + for i := 0; i < n; i++ { + sb.WriteRune(alphabet[rng.Intn(len(alphabet))]) + } + return sb.String() +} + +func TestChunkFast_RandomJSON(t *testing.T) { + rng := rand.New(rand.NewSource(20260926)) + for iter := 0; iter < 30000; iter++ { + // 构造一个「像 SSE chunk」的随机对象 + obj := map[string]interface{}{} + switch rng.Intn(4) { + case 0: + obj["choices"] = []interface{}{map[string]interface{}{ + "index": rng.Intn(3), + "delta": map[string]interface{}{"content": randJSONValue(rng, 2)}, + "finish_reason": []interface{}{nil, "", "stop", "length"}[rng.Intn(4)], + }} + case 1: + obj["choices"] = []interface{}{map[string]interface{}{ + "delta": map[string]interface{}{ + "reasoning_content": randomString(rng), + "content": randomString(rng), + }, + }} + case 2: + obj["usage"] = map[string]interface{}{ + "prompt_tokens": rng.Intn(1000), + "completion_tokens": rng.Intn(100), + "total_tokens": rng.Intn(1000), + } + default: + obj["choices"] = []interface{}{map[string]interface{}{ + "delta": map[string]interface{}{ + "tool_calls": []interface{}{map[string]interface{}{ + "index": rng.Intn(3), + "id": randomString(rng), + "type": "function", + "function": map[string]interface{}{ + "name": randomString(rng), + "arguments": randJSONValue(rng, 1), + }, + }}, + }, + }} + } + b, err := json.Marshal(obj) + if err != nil { + continue + } + checkPair(t, string(b)) + } +} + +// --------------------------------------------------------------------- +// 4. 随机字节（畸形输入）——两侧都必须拒绝、且不得 panic +// --------------------------------------------------------------------- + +func TestChunkFast_RandomBytes(t *testing.T) { + rng := rand.New(rand.NewSource(777)) + alphabet := []byte(`{}[]",:0123456789tfnul \` + "\n\t\xff\x80") + for iter := 0; iter < 30000; iter++ { + n := rng.Intn(60) + b := make([]byte, n) + for i := range b { + b[i] = alphabet[rng.Intn(len(alphabet))] + } + checkPair(t, string(b)) + } +} + +// --------------------------------------------------------------------- +// 5. 快速路径**确实被用到**（否则「优化」是假的） +// --------------------------------------------------------------------- + +func TestChunkFast_ActuallyHandlesRealistic(t *testing.T) { + realistic := []string{ + `{"id":"c","choices":[{"index":0,"delta":{"content":"中文内容"},"finish_reason":null}]}`, + `{"id":"c","choices":[{"index":0,"delta":{"content":"x"},"finish_reason":"stop"}]}`, + `{"id":"c","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"i","type":"function","function":{"name":"n","arguments":"{}"}}]}}]}`, + `{"id":"c","choices":[],"usage":{"prompt_tokens":1,"completion_tokens":2,"total_tokens":3}}`, + } + for _, in := range realistic { + _, handled, decided := chunkParseFast(in) + if !handled || !decided { + t.Errorf("真实负载未走快速路径（优化失效）: %s", in) + } + } + _ = fmt.Sprint() +} diff --git a/internal/agent/api/codec_streamchunk_c.go b/internal/agent/api/codec_streamchunk_c.go new file mode 100644 index 0000000..4bc2bc1 --- /dev/null +++ b/internal/agent/api/codec_streamchunk_c.go @@ -0,0 +1,328 @@ +//go:build cgo + +package api + +// codec_streamchunk_c.go —— SSE 分块解析的 C 化「结构导航」层（Go 侧绑定） +// +// ============================ 为什么是「导航」而不是「全量编解码」 ============================ +// 接线前实测出两条 wire 语义（docs/zh/c-core/sse-codec-c.md §5），它们让 +// 「整条 parseOpenAICompatibleStreamChunkFull 全 C 化」不成立： +// +// §5.1 重复键是**字段级合并**：`{"choices":[{content:a}],"choices":[{reasoning:r}]}` +// → content="a" **且** reasoning="r"。json.Unmarshal 的 object() 收尾时做 +// `v.SetIndex(i, subv.v)`，而 subv 拿到的是**已存在元素的指针**， +// 所以第二次是叠加而非替换。正确实现要维护「本次哪些字段出现过」的表。 +// §5.2 stringifyContent 的 default 分支 = `json.Marshal(interface{})`， +// 即**重新序列化**：`{"b":1,"a":2}` → `{"a":2,"b":1}`（键排序）、 +// `1e2` → `100`、`<` → `\u003c`、大 int 先舍入成 float64。 +// 逐值一致要求复刻 Ryu 最短浮点 + map 键排序 + HTML 转义 + int 舍入。 +// +// 而这两条**只在取值阶段**才需要。故本层只做**结构导航**： +// +// C：把 JSON 定位到「哪个值在哪里」——零分配、零解码，并直接给出两个热分支的结果 +// Go：把「已定位的原始字节」按既有类型 unmarshal，成串逻辑完全不变 +// +// ⇒ 类型检查的等价性靠「用**相同的 Go 类型** unmarshal **相同形状的子树**」保证， +// 而不靠 C 重新实现一遍类型规则。这是本设计同时拿到速度与正确性的关键。 +// +// 代价如实记录：命中字段仍要一次小 Unmarshal（原来是对整块做）。收益是免除 +// json.Unmarshal 对整块的**反射建树**——那正是每块 12~21 allocs 的主因。 +// +// ★ 键匹配**大小写敏感**（与 ha_json_scan.h 的 ha_json_key_eq 相反，两者用途不同） +// `content` 是 map[string]interface{}，取 `m["text"]` 走 map key 语义 +// ⇒ 大小写敏感。实测 `{"TEXT":"up"}` 取不到 `text`。 +// struct 字段（choices/delta/usage）是大小写**不**敏感 —— 那一跳交给 +// encoding/json，天然正确。 +// +// ★ C 实现放在 csrc/ha_sse.c 而**不是**本文件的 cgo 前言里： +// 前言里的 C 代码会逃出全部 C 门禁（告警 / ASan+UBSan / arm64 交叉 / 模糊测试）， +// 而这里恰恰是本刀最容易出错的位置。这是结构性决定，不是形式主义。 + +/* +#cgo CFLAGS: -std=c99 +#include +#include "ha_sse.h" + +// C 结构体一律不跨越语言边界（cgo 禁止「Go 指针指向的 Go 指针」， +// 实测会 panic），故所有 span 传递都拆成 (指针, 长度) 标量。 +static int go_obj_find(const char *p, size_t n, const char *key, int keylen, + char **vp, size_t *vlen, int *dup) { + ha_span obj, out; + obj.p = p; obj.len = n; + int rc = ha_sse_obj_find(&obj, key, (size_t)keylen, &out, dup); + if (rc == 1) { *vp = (char *)out.p; *vlen = out.len; } + return rc; +} + +static int go_arr_first(const char *p, size_t n, char **vp, size_t *vlen) { + ha_span arr, out; + arr.p = p; arr.len = n; + int rc = ha_sse_arr_first(&arr, &out); + if (rc == 1) { *vp = (char *)out.p; *vlen = out.len; } + return rc; +} + +static int go_stringify(const char *p, size_t n, char *out, size_t cap, + size_t *outlen) { + ha_span val; + val.p = p; val.len = n; + return ha_sse_stringify(&val, out, cap, outlen); +} + +static int go_arg_string(const char *p, size_t n, char *out, size_t cap, + size_t *outlen) { + ha_span val; + val.p = p; val.len = n; + return ha_sse_arg_string(&val, out, cap, outlen); +} + +static int go_obj_find_ci(const char *p, size_t n, const char *key, int keylen, + char **vp, size_t *vlen, int *dup) { + ha_span obj, out; + obj.p = p; obj.len = n; + int rc = ha_sse_obj_find_ci(&obj, key, (size_t)keylen, &out, dup); + if (rc == 1) { *vp = (char *)out.p; *vlen = out.len; } + return rc; +} + +static int go_root_object(const char *p, size_t n) { + ha_span doc; + doc.p = p; doc.len = n; + return ha_sse_root_object(&doc); +} + +// 把数组全部元素写进 out（Go 侧预分配的 span 数组）。 +// 返回元素数；超出 cap 时返回 -1（调用方据此判定「需要更大的缓冲」⇒ 回退）。 +static int go_arr_all(const char *p, size_t n, ha_span *out, int cap) { + ha_json_scan sc; + int count = 0; + ha_json_scan_init(&sc, p, n); + (void)ha_json_scan_ws(&sc); + if (ha_json_scan_eof(&sc) || sc.s[sc.i] != '[') { return -1; } + sc.i++; + for (;;) { + (void)ha_json_scan_ws(&sc); + if (ha_json_scan_eof(&sc) || sc.s[sc.i] == ']') { break; } + if (count >= cap) { return -1; } + size_t start = sc.i; + if (!ha_json_skip(&sc)) { return -1; } + out[count].p = p + start; + out[count].len = sc.i - start; + count++; + (void)ha_json_scan_ws(&sc); + if (ha_json_scan_eof(&sc)) { return -1; } + if (sc.s[sc.i] == ',') { sc.i++; continue; } + if (sc.s[sc.i] == ']') { break; } + return -1; + } + return count; +} + +static int go_sse_abi(void) { return ha_sse_abi_version(); } +*/ +import "C" + +import "unsafe" + +// --------------------------------------------------------------------- +// span 表示 +// --------------------------------------------------------------------- + +// strSpan 是 JSON 里一段字节，指向**原缓冲**（零拷贝）。 +type strSpan struct { + p *C.char + n C.size_t +} + +func (s strSpan) valid() bool { return s.p != nil && s.n > 0 } + +// bytes 把 span 变成 Go 字节切片（此处才产生一次拷贝）。 +// +// ★ 用途：把「已定位的原始子树」交给 json.Unmarshal —— 用同一 Go 类型 +// unmarshal 同一形状，是本层保证「类型检查语义与原实现一致」的手段。 +func (s strSpan) bytes() []byte { + if s.p == nil || s.n == 0 { + return nil + } + return unsafe.Slice((*byte)(unsafe.Pointer(s.p)), int(s.n)) +} + +// str 把 span 变成 Go 字符串（此处才产生一次拷贝）。 +func (s strSpan) str() string { + if s.p == nil || s.n == 0 { + return "" + } + return string(unsafe.Slice((*byte)(unsafe.Pointer(s.p)), int(s.n))) +} + +// firstByte 只看首字节，用于区分值类型。 +func (s strSpan) firstByte() byte { + if s.p == nil || s.n == 0 { + return 0 + } + return *(*byte)(unsafe.Pointer(s.p)) +} + +// --------------------------------------------------------------------- +// 定位 +// --------------------------------------------------------------------- + +// findKey 在 obj 里按**大小写敏感**的键定位值。 +// 返回 (span, found, dup, malformed)。 +// dup=true ⇒ 发现重复键，调用方**必须**整体回退 encoding/json（§5.1）。 +func findKey(obj strSpan, name string) (strSpan, bool, bool, bool) { + if !obj.valid() { + return strSpan{}, false, false, false + } + keyp, keyn := cstr(name) + var vp *C.char + var vlen C.size_t + var dup C.int + rc := C.go_obj_find(obj.p, obj.n, keyp, C.int(keyn), &vp, &vlen, &dup) + switch rc { + case 1: + return strSpan{vp, vlen}, true, dup == 1, false + case 0: + return strSpan{}, false, dup == 1, false + default: + return strSpan{}, false, false, true // 畸形 ⇒ 让 encoding/json 判 + } +} + +// firstElem 取数组第一个元素的 span。 +func firstElem(arr strSpan) (strSpan, bool) { + if !arr.valid() { + return strSpan{}, false + } + var vp *C.char + var vlen C.size_t + if C.go_arr_first(arr.p, arr.n, &vp, &vlen) != 1 { + return strSpan{}, false + } + return strSpan{vp, vlen}, true +} + +// --------------------------------------------------------------------- +// 取值（C 可判定的热分支） +// --------------------------------------------------------------------- + +// decBuf 是解码/反转义用的可写缓冲。 +// +// ★ 尺寸必须按输入长度定：C 侧要求 cap >= len*3+4（最坏每字节一个 U+FFFD）， +// 不足时它会返回 0 让调用方回退 Go（宁可慢也不截断）。 +// 每次调用 1 次分配（原来整块 Unmarshal 是 12~21 次）—— 这是主要的节省点。 +func decBuf(n int) []byte { return make([]byte, n*3+8) } + +// stringifyC 对应 Go stringifyContent 的**C 可判定分支** +// （字符串值 / 文本数组），返回 (结果, handled)。 +// handled=false ⇒ 值类型需要 json.Marshal 重新编码（§5.2），调用方须回退 Go。 +func stringifyC(val strSpan) (string, bool) { + if !val.valid() { + // 缺失 / 空 ⇒ Go 侧 stringifyContent(nil) 也是 "" + return "", true + } + buf := decBuf(int(val.n)) + var outLen C.size_t + if C.go_stringify(val.p, val.n, cstrb(buf), C.size_t(len(buf)), &outLen) != 1 { + return "", false + } + return string(buf[:int(outLen)]), true +} + +// argStringC 取出 arguments 的**字符串**形态（省掉 interface{} 与二次解析）。 +func argStringC(val strSpan) (string, bool) { + if !val.valid() { + return "", false + } + buf := decBuf(int(val.n)) + var outLen C.size_t + if C.go_arg_string(val.p, val.n, cstrb(buf), C.size_t(len(buf)), &outLen) != 1 { + return "", false + } + return string(buf[:int(outLen)]), true +} + +// sseABIVersion 供 ABI 漂移测试使用。 +func sseABIVersion() int { return int(C.go_sse_abi()) } + +// --------------------------------------------------------------------- +// 顶层 helper：大小写不敏感（struct 字段语义）与根对象校验 +// --------------------------------------------------------------------- + +// C_size 把 Go int 转成 C.size_t（零拷贝 span 的长度）。 +func C_size(n int) C.size_t { return C.size_t(n) } + +// rootSpan 构造指向 data 的 span（零拷贝）。 +func rootSpan(data string) strSpan { return strSpan{cstrp(data), C_size(len(data))} } + +// findKeyCI 按**大小写不敏感**定位（Go struct 字段语义）。 +func findKeyCI(obj strSpan, name string) (strSpan, bool, bool, bool) { + return findKeyGeneric(obj, name, true) +} + +// findKeyCS 按**大小写敏感**定位（Go map key 语义）。 +func findKeyCS(obj strSpan, name string) (strSpan, bool, bool, bool) { + return findKeyGeneric(obj, name, false) +} + +func findKeyGeneric(obj strSpan, name string, ci bool) (strSpan, bool, bool, bool) { + if !obj.valid() { + return strSpan{}, false, false, false + } + keyp, keyn := cstr(name) + var vp *C.char + var vlen C.size_t + var dup C.int + var rc C.int + if ci { + rc = C.go_obj_find_ci(obj.p, obj.n, keyp, C.int(keyn), &vp, &vlen, &dup) + } else { + rc = C.go_obj_find(obj.p, obj.n, keyp, C.int(keyn), &vp, &vlen, &dup) + } + switch rc { + case 1: + return strSpan{vp, vlen}, true, dup == 1, false + case 0: + return strSpan{}, false, dup == 1, false + default: + return strSpan{}, false, false, true + } +} + +// sseRootObject 校验「恰好一个良构对象」（含尾部残留检查）。 +func sseRootObject(doc strSpan) bool { + if !doc.valid() { + return false + } + return C.go_root_object(doc.p, doc.n) == 1 +} + + +// scanArray 枚举数组的全部元素 span（零拷贝，指向原缓冲）。 +// +// ★ 为什么要「先数一遍再填」：C 侧迭代器一次回一个元素，而 Go 需要一个切片。 +// 做法是让 C 一次把**所有元素**写进 Go 侧的 span 数组 +// （Go 预分配、容量按字节数上界估），单趟、无 C 分配。 +func scanArray(arr strSpan) ([]strSpan, bool) { + if !arr.valid() || arr.firstByte() != '[' { + return nil, false + } + // 元素数上界：每个元素至少 1 字节 + 分隔符 ⇒ ≤ 字节数 + capHint := int(arr.n) + if capHint < 4 { + capHint = 4 + } + if capHint > 1024 { + capHint = 1024 // 工具调用数量级很小；超出部分不可能（协议上界） + } + spans := make([]C.ha_span, capHint) + n := C.go_arr_all(arr.p, arr.n, &spans[0], C.int(capHint)) + if n < 0 { + return nil, false + } + out := make([]strSpan, 0, int(n)) + for i := 0; i < int(n); i++ { + out = append(out, strSpan{spans[i].p, spans[i].len}) + } + return out, true +} diff --git a/internal/agent/api/ha_sse.c b/internal/agent/api/ha_sse.c new file mode 120000 index 0000000..65e61b4 --- /dev/null +++ b/internal/agent/api/ha_sse.c @@ -0,0 +1 @@ +../../../csrc/src/ha_sse.c \ No newline at end of file diff --git a/internal/agent/api/ha_sse.h b/internal/agent/api/ha_sse.h new file mode 120000 index 0000000..30d4e1d --- /dev/null +++ b/internal/agent/api/ha_sse.h @@ -0,0 +1 @@ +../../../csrc/include/ha_sse.h \ No newline at end of file diff --git a/internal/agent/api/provider.go b/internal/agent/api/provider.go index 0180426..2788e08 100644 --- a/internal/agent/api/provider.go +++ b/internal/agent/api/provider.go @@ -625,31 +625,41 @@ func normalizeStreamToolCalls(raw []openAIToolCall) []ToolCall { } out := make([]ToolCall, 0, len(raw)) for _, tc := range raw { - name := tc.Function.Name - argsRaw := tc.Function.Arguments - if name == "" { - name = tc.Name - // 仅当顶层 Arguments 存在才用扁平格式；否则保留 function.arguments 嵌套值 - // （OpenAI 流式续传 chunk：name 不重发但 function.arguments 继续） - if tc.Arguments != nil { - argsRaw = tc.Arguments - } - } - typ := tc.Type - if typ == "" && (tc.ID != "" || name != "" || argsRaw != nil) { - typ = "function" - } - out = append(out, ToolCall{ - ID: tc.ID, - Type: typ, - Name: name, - RawArguments: rawArgsString(argsRaw), - StreamIndex: tc.Index, - }) + out = append(out, normalizeStreamToolCall(tc)) } return out } +// normalizeStreamToolCall 是单元素的归一化逻辑。 +// +// ★ 之所以从循环里抽成单元素函数：C 快速路径逐元素处理（而不是整块 +// unmarshal 成 []openAIToolCall），必须与本函数**共用**同一份归一化逻辑， +// 否则两条路径会在「name 回退 / type 补全 / arguments 取哪一份」这些 +// 条件分支上分叉。抽出后循环与快速路径都调它，结构上无法分叉。 +func normalizeStreamToolCall(tc openAIToolCall) ToolCall { + name := tc.Function.Name + argsRaw := tc.Function.Arguments + if name == "" { + name = tc.Name + // 仅当顶层 Arguments 存在才用扁平格式；否则保留 function.arguments 嵌套值 + // （OpenAI 流式续传 chunk：name 不重发但 function.arguments 继续） + if tc.Arguments != nil { + argsRaw = tc.Arguments + } + } + typ := tc.Type + if typ == "" && (tc.ID != "" || name != "" || argsRaw != nil) { + typ = "function" + } + return ToolCall{ + ID: tc.ID, + Type: typ, + Name: name, + RawArguments: rawArgsString(argsRaw), + StreamIndex: tc.Index, + } +} + func parseToolArguments(v interface{}) map[string]interface{} { switch x := v.(type) { case nil: @@ -705,64 +715,32 @@ func stringifyContent(v interface{}) string { // 兼容多种 token 用量键名（prompt_tokens/prompt、total_tokens/total 等） // 与 prompt cache 细节字段。返回 false 表示非内容块（纯 usage 心跳等）。 func parseOpenAICompatibleStreamChunkFull(data string) (StreamChunk, bool) { - var raw struct { - Choices []struct { - Delta struct { - Content interface{} `json:"content"` - ReasoningContent string `json:"reasoning_content"` - ToolCalls []openAIToolCall `json:"tool_calls"` - } `json:"delta"` - FinishReason *string `json:"finish_reason"` - } `json:"choices"` - UpstreamUsage struct { - PromptTokens int `json:"prompt_tokens"` - CompletionTokens int `json:"completion_tokens"` - TotalTokens int `json:"total_tokens"` - Prompt int `json:"prompt"` - Completion int `json:"completion"` - Total int `json:"total"` - PromptCacheHit int `json:"prompt_cache_hit_tokens"` - PromptCacheMiss int `json:"prompt_cache_miss_tokens"` - PromptTokensDetails *struct { - CachedTokens int `json:"cached_tokens"` - } `json:"prompt_tokens_details"` - } `json:"usage"` - } - if err := json.Unmarshal([]byte(data), &raw); err != nil { - return StreamChunk{}, false - } - - var usage *TokenUsage - pu := raw.UpstreamUsage - if pu.Total > 0 || pu.TotalTokens > 0 || pu.Prompt > 0 || pu.PromptTokens > 0 { - usage = &TokenUsage{ - Prompt: pickFirstInt(pu.PromptTokens, pu.Prompt), - Completion: pickFirstInt(pu.CompletionTokens, pu.Completion), - Total: pickFirstInt(pu.TotalTokens, pu.Total), + // ★ C 快速路径（结构导航）：定位在 C（零分配、零解码），类型检查与 + // 需要重新序列化的形态交回 Go 的 encoding/json。 + // + // 契约：必须与 chunkParseGo 对所有输入产出完全相同的结果。 + // 保证方式见 codec_chunkfast_c.go 顶部：任一环节「不确定」即**整体回退** + // chunkParseGo，且拼装/归一化两条路径**共用**同一份代码。 + // + // 为什么保留 Go 实现：它既是回退目标，也是黄金对照的参照实现 —— + // 没有它，「C 化没坏」就只是感觉而不是证据。 + // + // ★ 开关：chunkFastEnabled 目前为 false —— 实测本架构比原实现**慢** + // （2016ns/20allocs vs 1325ns/13allocs），根因是「5+ 次 cgo 边界 + // × 每次 ~200ns」吃掉了收益。详见 codec_chunkfast_c.go 的说明与 + // docs/zh/c-core/sse-codec-c.md §六。改造方向已由天花板实验确认可行。 + if chunkFastEnabled { + if ck, handled, decided := chunkParseFast(data); handled && decided { + return ck, true } } + return chunkParseGo(data) +} - if len(raw.Choices) == 0 { - // 纯 usage 心跳块：有 usage 就透传，否则丢弃 - if usage != nil { - return StreamChunk{Usage: usage}, true - } - return StreamChunk{}, false - } - - choice := raw.Choices[0] - ck := StreamChunk{ - Content: stringifyContent(choice.Delta.Content), - ReasoningContent: choice.Delta.ReasoningContent, - ToolCalls: normalizeStreamToolCalls(choice.Delta.ToolCalls), - Usage: usage, - } - // finish reason 为空字符串不算终止信号（sensenova 每块都发 ""） - if choice.FinishReason != nil && *choice.FinishReason != "" { - ck.Done = true - ck.FinishReason = *choice.FinishReason - } - return ck, true +// parseOpenAICompatibleStreamChunkFullGo 供黄金对照测试直接调原始实现， +// 用于验证快速路径与它逐值等价。 +func parseOpenAICompatibleStreamChunkFullGo(data string) (StreamChunk, bool) { + return chunkParseGo(data) } // pickFirstInt 返回 a 非零时的 a，否则 b（兼容 *_tokens 与短键名两种 usage 格式）。 @@ -772,7 +750,6 @@ func pickFirstInt(a, b int) int { } return b } - // streamHTTPClient 返回专用的流式 HTTP client（懒初始化）。 // SSE 长连接不能套整体超时（非流式 180s 会在长流中途报断）， // 只保留拨号/握手超时。