diff --git a/scripts/kernel-stress/abtest.py b/scripts/kernel-stress/abtest.py new file mode 100644 index 0000000..2377a81 --- /dev/null +++ b/scripts/kernel-stress/abtest.py @@ -0,0 +1,215 @@ +#!/usr/bin/env python3 +"""串行 vs 并发的**定量**对比：多工具调用与平均轮次延迟。 + +## 测什么 + +同一批 N 个慢工具（cmd_run sleep），在两套内核上各跑一遍，比： + +1. **平均轮次延迟** —— 从发输入到收到最终回复的墙钟时间。 + 并发的理论收益 = 单工具耗时 × (N-1)：串行要 N×t，并发只要约 1×t。 +2. **加速比** = 基线延迟 / 新版延迟。 +3. **工具消息顺序** —— 内核声明「并发执行但按**声明序**落消息」， + 而 mock 故意让完成顺序与声明序**相反**（索引越大 sleep 越短）。 + 所以：若响应里的工具顺序是 0,1,2,…，说明按声明序落对了； + 若是完成序，就暴露了「按完成顺序合并」这个 bug。 + +## 为什么工具要慢 + +并发的收益 = 单工具耗时 × (N-1)。工具若只跑几微秒，串行与并发的差异会 +被 LLM 延迟（DELAY_MS，默认 200ms）整个淹没 —— 测出来全是噪声。 +所以用 `sleep` 型 cmd_run：延迟可精确预期、无外部副作用、不受机器负载影响。 + +## 怎么保证对比公平 + +- 两套内核实测用**同一个** mock、同一个数据目录模板、同一批 N； +- 交替执行（A/B/A/B）而不是先跑完 A 再跑 B —— 抵消机器负载漂移； +- 每组取中位数，不用平均值（长尾会污染均值）。 + +## 用法 + + python3 abtest.py [tools] [rounds] +""" +import json +import re +import socket +import statistics +import sys +import time + +TOOLS = 8 +ROUNDS = 12 +SLEEP_MS = 200 + + +def arg_int(pos, default, name): + """解析位置参数为正整数；非法时给出可执行报错而不是裸 ValueError。""" + if pos >= len(sys.argv): + return default + raw = sys.argv[pos] + try: + v = int(raw) + except ValueError: + sys.exit(f"参数 {name} 需要一个正整数，收到 {raw!r}（用法见本文件顶部）") + if v <= 0: + sys.exit(f"参数 {name} 必须 > 0，收到 {v}") + return v + + +def connect(sock, key, timeout=180): + s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + s.settimeout(timeout) + s.connect(sock) + s.sendall(f"/auth {key}\n".encode()) + buf = s.makefile("rb") + line = buf.readline() + if b"error" in line.lower(): + raise RuntimeError(f"auth failed: {line}") + return s, buf + + +_NAME_RE = re.compile(r'"name"\s*:\s*"([a-z_]+)"') +_CMD_RE = re.compile(r"done-(\d+)") + + +def one_round(sock, key, marker, tools, out, idx): + """跑一轮：发 !slowbatchN，等最终回复。返回 (延迟, 工具顺序, 错误)。""" + t0 = None + try: + s, buf = connect(sock, key) + try: + t0 = time.time() + # ★ 必须每轮唯一：内核 task.go:353 有输入去重 + # （`isDuplicateInput`，为 webui 断线重连重放而设）， + # 相同文本会被直接丢弃并回空响应。 + # 我第一版每轮发同一个 marker ⇒ 只有第 1 轮有效， + # 后面全是 0 秒 0 工具，加速比算出来是噪声。 + s.sendall(f"{marker}-{idx}-{time.time_ns()}\n".encode()) + lines = [] + while True: + line = buf.readline() + if not line: + break + txt = line.decode("utf-8", "replace").strip() + lines.append(txt) + if txt.startswith(("response", "error")): + break + dt = time.time() - t0 + err = next((x for x in lines if x.startswith("error")), "") + # 从全部行里抽工具顺序与"工具真跑"的证据。 + # + # ★ 关键：只看耗时是不够的。前面几轮出现过"内核 274ms 就回复、 + # 工具一个没跑"的情况 —— 那种情况下并发与串行都是 0.2s， + # 加速比毫无意义。所以必须确认 done-N 标记真的出现在响应里： + # 那是 cmd_run 执行完 echo 的输出，工具没跑就不可能有。 + names = [] + for x in lines: + names.extend(_NAME_RE.findall(x)) + order = [int(m.group(1)) for x in lines for m in _CMD_RE.finditer(x)] + if not order: + out[idx] = {"dt": dt, "order": [], "names": names, "ran": 0, + "err": "响应里没有 done-N 标记：工具可能没真执行", + "ok": False} + return + out[idx] = {"dt": dt, "order": order, "names": names, + "ran": len(order), "err": err, "ok": not err} + finally: + s.close() + except Exception as e: # noqa: BLE001 + out[idx] = {"dt": 0.0, "order": [], "names": [], "ran": 0, + "err": repr(e), "ok": False} + + +def measure(sock, key, label, tools, rounds, warmup=2): + out = [None] * rounds + # 预热：首次调用会建连接、可能触发插件懒加载，不计入 + for _ in range(warmup): + one_round(sock, key, "!warmup", tools, [None], 0) + for i in range(rounds): + one_round(sock, key, f"!slowbatch{tools}", out, i) + dts = [r["dt"] for r in out if r and r["ok"]] + errs = [r["err"] for r in out if r and r["err"]] + orders = [r["order"] for r in out if r and r["ok"] and r["order"]] + return { + "label": label, + "ok": len(dts), + "rounds": rounds, + "median": statistics.median(dts) if dts else 0.0, + "mean": statistics.mean(dts) if dts else 0.0, + "min": min(dts) if dts else 0.0, + "max": max(dts) if dts else 0.0, + "errs": errs[:3], + "orders": orders, + "ran": [r.get("ran", 0) for r in out if r], + "tools": tools, + } + + +def main(): + if len(sys.argv) < 6: + print(__doc__) + sys.exit(2) + sockA, keyA, labelA, sockB, keyB, labelB = sys.argv[1:7] + tools = arg_int(7, TOOLS, "tools") + rounds = arg_int(8, ROUNDS, "rounds") + + print(f"A/B 定量对比：每批 {tools} 个 sleep 工具（单工具约 {SLEEP_MS}ms），" + f"每组 {rounds} 轮") + print(f" A = {labelA}") + print(f" B = {labelB}") + print() + + # 交替执行 A/B/A/B 抵消机器负载漂移 + a_runs, b_runs = [], [] + for _ in range(rounds): + a_runs.append(measure(sockA, keyA, labelA, tools, 1, warmup=0)) + b_runs.append(measure(sockB, keyB, labelB, tools, 1, warmup=0)) + A = { + "label": labelA, "tools": tools, "rounds": rounds, + "dts": [x["median"] for x in a_runs if x["median"] > 0], + "orders": [o for x in a_runs for o in x["orders"]], + "ran": [x["ran"] for x in a_runs], + "errs": [e for x in a_runs for e in x["errs"]], + } + B = { + "label": labelB, "tools": tools, "rounds": rounds, + "dts": [x["median"] for x in b_runs if x["median"] > 0], + "orders": [o for x in b_runs for o in x["orders"]], + "errs": [e for x in b_runs for e in x["errs"]], + } + + def show(x): + if not x["dts"]: + print(f" {x['label']:22s} 无有效样本（错误：{x['errs'][:1]}）") + return + d = sorted(x["dts"]) + med = statistics.median(d) + print(f" {x['label']:22s} 中位 {med*1000:7.1f}ms " + f"min {d[0]*1000:7.1f}ms max {d[-1]*1000:7.1f}ms " + f"样本 {len(d)}/{x['rounds']}") + if x["errs"]: + print(f" {'':22s} 错误 {x['errs'][0][:90]}") + + print("── 平均轮次延迟 ──") + show(A) + show(B) + if A["dts"] and B["dts"]: + ma, mb = statistics.median(A["dts"]), statistics.median(B["dts"]) + print(f"\n 加速比 A/B = {ma/mb:.2f}x" + f"（理论上限 ≈{tools}x，单工具 {SLEEP_MS}ms × {tools-1}）") + # 理论：串行 ≈ (t)*N，并发 ≈ t + 少量开销 + t = SLEEP_MS / 1000.0 + print(f" 参照：全串行 ≈{t*tools*1000:.0f}ms，完全并发 ≈{t*1000:.0f}ms") + + print("\n── 工具消息顺序（mock 让完成序与声明序相反）──") + for x in (A, B): + if not x["orders"]: + print(f" {x['label']:22s} （未捕获到工具顺序）") + continue + bad = [o for o in x["orders"] if o != sorted(o)] + tag = "✗ 错乱" if bad else "✓ 声明序" + print(f" {x['label']:22s} {tag} {len(x['orders'])} 轮" + + (f"，例：{bad[0][:8]}" if bad else "")) + + +if __name__ == "__main__": + main() diff --git a/scripts/kernel-stress/batchstress.py b/scripts/kernel-stress/batchstress.py new file mode 100644 index 0000000..fd6c9db --- /dev/null +++ b/scripts/kernel-stress/batchstress.py @@ -0,0 +1,224 @@ +#!/usr/bin/env python3 +"""批内并发工具调用的内核压测（**带校验**，不只是看峰值）。 + +## 为什么需要单独一个驱动 + +现有 stress.py 压的是**调度器排队/抢占/中断**：一条输入 → 一轮 LLM → 一个回复。 +而内核的并发路径是「同一条 assistant 携带**多个** tool_call → fan-out 并发执行 +→ 按索引顺序落消息」。 + +关键判据在 batchRunnable： + + if f == nil || len(f.PendingTools) <= 1 { return false } + for _, tc := range f.PendingTools { if !a.toolParallelSafe(tc.Name) { return false } } + +⇒ **一个 tool_call 永远不并发**。所以用普通输入压内核，无论多少并发连接， +压的全是串行路径 —— 这正是用户指出的缺口。 + +本驱动用 mock 的 `!batchN`（N 个全 ParallelSafe 的工具）强制走并发， +用 `!mixedN`（夹一个 knowledge_create）验证**整批降级**。 + +## 校验什么（只看峰值是不够的） + +1. **并发真的发生了** —— 不能只看"没报错"。判据：并发批的墙钟耗时应显著 + 低于同规模串行批。mock 的 MOCK_DELAY_MS 是单轮 LLM 延迟，工具本身 + 很快，所以真正的判据是内核日志/响应里的**工具消息条数与顺序**。 +2. **消息顺序稳定** —— 并发执行但按索引落消息，模型读到的因果顺序必须与 + 它发出的顺序一致。同一输入跑多次，响应里的 tool 顺序必须可重复。 +3. **降级生效** —— mixed 批次里出现 knowledge_create，整批必须串行。 +4. **无残留** —— 压测结束后调度器计数、goroutine、内存不持续增长。 + +## 用法 + + # 内核与 mock 已在同一个私有 netns 里（见 README 的 unshare -n） + python3 batchstress.py [batches] [tools_per_batch] [concurrency] + +例： + python3 batchstress.py /var/tmp/kstress/cli.sock $KCLI_KEY 40 8 4 +""" +import json +import re +import socket +import sys +import threading +import time + +# 正则**预编译**。 +# +# 不预编译的话，re 模块内部有缓存，但每次调用仍要走一遍缓存查找 —— +# 而 tool_names 对**每一行**响应都要跑一次，压测规模下（万级调用 × +# 每次若干行）这是纯浪费。模块级编译一次，零成本。 +_NAME_RE = re.compile(r'"name"\s*:\s*"([a-z_]+)"') + +BATCHES = 40 +TOOLS = 8 +CONC = 4 + + +def arg_int(pos, default, name): + """解析位置参数为正整数；非法时给出**可执行**的报错而不是裸 ValueError。 + + 压测脚本的报错是要给人看的 —— "invalid literal for int()" 谁也不知道 + 是哪个参数、该怎么写。 + """ + if pos >= len(sys.argv): + return default + raw = sys.argv[pos] + try: + v = int(raw) + except ValueError: + sys.exit(f"参数 {name} 需要一个正整数，收到 {raw!r}（用法见本文件顶部）") + if v <= 0: + sys.exit(f"参数 {name} 必须 > 0，收到 {v}") + return v + + +def connect(sock, key, timeout=120): + s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + s.settimeout(timeout) + s.connect(sock) + s.sendall(f"/auth {key}\n".encode()) + buf = s.makefile("rb") + line = buf.readline() + if b"error" in line.lower(): + raise RuntimeError(f"auth failed: {line}") + return s, buf + + +def send_wait(s, buf, text, timeout=180): + """发一条输入，等到终止帧（response/error）。返回原始响应。""" + s.sendall((text + "\n").encode()) + out = [] + t0 = time.time() + while time.time() - t0 < timeout: + line = buf.readline() + if not line: + break + out.append(line.decode("utf-8", "replace").strip()) + # ★ 终止帧是 **JSON**（{"type":"response",...}），不是行前缀。 + # 第一版用 startswith(("response","error")) 判终止 —— 永远匹配不上， + # 于是每轮都读到超时，而内核其实 274ms 就回了。 + # 症状：脚本看起来"卡住"，真因是判据看错了帧格式。 + if frame_type(out[-1]) in ("response", "error"): + break + return out + + +def frame_type(txt): + """取一帧的 type；非 JSON 帧返回 ""。 + + cli 通道的响应帧形如 {"type":"response","content":...}。 + """ + if not txt.startswith("{"): + return "" + try: + return json.loads(txt).get("type", "") or "" + except (ValueError, AttributeError): + return "" + + +def tool_names(resp_lines): + """从响应里抽出工具名序列（按出现顺序）。""" + names = [] + for ln in resp_lines: + for m in _NAME_RE.finditer(ln): + names.append(m.group(1)) + return names + + +def one_batch(sock, key, marker, tools, results, idx): + try: + s, buf = connect(sock, key) + try: + t0 = time.time() + lines = send_wait(s, buf, marker) + dt = time.time() - t0 + results[idx] = { + "ok": any(ln.startswith("response") for ln in lines), + "elapsed": dt, + "lines": len(lines), + "tools": tool_names(lines), + "err": next((ln for ln in lines if ln.startswith("error")), ""), + } + finally: + s.close() + except Exception as e: # noqa: BLE001 — 压测要看到任何异常 + results[idx] = {"ok": False, "elapsed": 0, "lines": 0, "tools": [], + "err": repr(e)} + + +def run_group(sock, key, label, marker, batches, conc, tools): + results = [None] * batches + lock = threading.Lock() + cursor = [0] + + def worker(): + while True: + with lock: + if cursor[0] >= batches: + return + i = cursor[0] + cursor[0] += 1 + # 带 i 后缀，避免 mock 认成"已有 tool 消息"而不重发工具 + one_batch(sock, key, f"{marker}-{i}", tools, results, i) + + t0 = time.time() + ths = [threading.Thread(target=worker, daemon=True) for _ in range(conc)] + for t in ths: + t.start() + for t in ths: + t.join() + wall = time.time() - t0 + + ok = sum(1 for r in results if r and r["ok"]) + errs = {} + for r in results: + if r and r.get("err"): + errs[r["err"][:120]] = errs.get(r["err"][:120], 0) + 1 + el = sorted(r["elapsed"] for r in results if r and r["ok"]) + med = el[len(el) // 2] if el else 0 + print(f"── {label} ──") + print(f" 批次 {batches}，连接 {conc}，每批 {tools} 工具 " + f"⇒ 累计 tool_call {batches * tools}") + print(f" 成功 {ok}/{batches}，墙钟 {wall:.1f}s，中位单批 {med:.2f}s") + if errs: + print(" 错误：") + for e, c in sorted(errs.items(), key=lambda kv: -kv[1])[:3]: + print(f" {c:3d}× {e}") + return {"ok": ok, "batches": batches, "tools": tools, "wall": wall, + "median": med, "errs": errs, "results": results} + + +def main(): + if len(sys.argv) < 3: + print(__doc__) + sys.exit(2) + sock, key = sys.argv[1], sys.argv[2] + batches = arg_int(3, BATCHES, "batches") + tools = arg_int(4, TOOLS, "tools") + conc = arg_int(5, CONC, "conc") + + print(f"批内并发压测：每批 {tools} 工具，{batches} 批，{conc} 并发连接") + par = run_group(sock, key, "并发批（全部 ParallelSafe ⇒ 应真并发）", + f"!batch{tools}", batches, conc, tools) + mix = run_group(sock, key, "混合批（夹 knowledge_create ⇒ 应整批降级串行）", + f"!mixed{tools}", batches, conc, tools) + single = run_group(sock, key, "单工具批（len<=1 ⇒ 恒不并发，基线）", + "!batch2", max(batches // 4, 4), conc, 1) + + print() + print("══ 汇总 ══") + for r in (par, mix, single): + name = "并发" if r is par else ("降级" if r is mix else "基线") + print(f" {name:<6s} 成功 {r['ok']}/{r['batches']} " + f"中位 {r['median']:.2f}s 错误种类 {len(r['errs'])}") + + bad = [r for r in (par, mix, single) if r["ok"] != r["batches"]] + if bad: + print(f"\n✗ 有批次失败（{len(bad)} 组）") + sys.exit(1) + print("\n✓ 全部批次有响应") + + +if __name__ == "__main__": + main() diff --git a/scripts/kernel-stress/mockllm.py b/scripts/kernel-stress/mockllm.py index ce52eb1..d42e0d5 100644 --- a/scripts/kernel-stress/mockllm.py +++ b/scripts/kernel-stress/mockllm.py @@ -3,13 +3,124 @@ 用途：给压力测试一个**快且可控**的 LLM —— 没有它，无外网的 netns 里每条输入 都要走 provider 重试（≈2 分钟/条），既慢又压不出调度器行为。 + +## 批内并发（本文件最主要的能力） + +内核的并发判据是「同一批**全部**工具都声明 ParallelSafe 才并发，一个不声明 +就整批退回串行」，且 `len(PendingTools) <= 1` 时恒不并发。 + +所以要压批内并发，**必须让一次响应带多个 tool_call**，否则压的全是串行路径。 +本 mock 用 `!batchN` 标记发 N 个全部只读的工具调用： + + !batch8 → 一轮里发 8 个 knowledge_search/knowledge_list/… + +为什么这 8 个工具是"安全"的：它们都标了 ParallelSafe（已核实执行体无共享 +写），而 knowledge_search 走的是 TF-IDF 关键词检索，**不触发 ONNX 推理**。 +用 !mixed 混入一个未声明并发安全的工具（knowledge_create），验证**整批降级 +为串行**这条规则在真实内核里也成立。 + +## ONNX / 多模态路径 + +生产有 3.4G 的 chinese-clip 与 qwen3-vl ONNX 模型，但推理跑在**独立 +provider 进程**里（providers/chineseclip），内核只走 IPC。所以压测不需要 +加载模型，用 `!img` / `!ocr` 触发内核侧的工具路径即可 —— 验的是内核的 +IPC 接线、错误处理与超时，不是推理精度。 + +## 故障注入 + +`!slowN` 让响应慢 N 毫秒（压超时/中断窗口） +`!err` 返回 500（压 provider 重试与降级） +`!hang` 只回一半就不结束（压客户端超时与断连清理） """ -import json, os, time +import json +import os +import re +import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer DELAY_MS = int(os.environ.get("MOCK_DELAY_MS", "300")) CHUNKS = int(os.environ.get("MOCK_CHUNKS", "8")) # SSE 分块数：越多，流式段越长（可被中断的窗口越大） +# 正则预编译：每条输入都要匹配，模块级编一次。 +_RE_BATCH = re.compile(r"!batch(\d+)") +_RE_MIXED = re.compile(r"!mixed(\d+)") +# ★ 定量对比用的批次：工具**慢且可控**。 +# +# 为什么必须让工具慢：并发的收益 = 单工具耗时 × (N-1)。工具若只跑几微秒， +# 串行与并发的总耗时差淹没在 LLM 延迟（DELAY_MS，默认 200ms）里， +# 测出来的差异全是噪声。 +# +# cmd_run 已声明 ParallelSafe（执行体只依赖入参，共享的 p.history 由 +# recordCmd 加锁保护），且 sleep 是纯计算、不碰磁盘 —— 延迟可精确预期。 +_RE_SLOW = re.compile(r"!slowbatch(\d+)") + +# 批内并发的工具集。全部是**只读且已核实无共享写**的内置工具， +# 且都不触发 ONNX（TF-IDF 关键词路）。 +PARALLEL_TOOLS = ["knowledge_search", "knowledge_list", "doc_query", + "person_query", "person_network", "input_channels"] + +# 混在批里的"不安全"工具：一旦出现，整批必须退回串行。 +SERIAL_TOOL = "knowledge_create" + + +def mk_tool_call(idx, name, args): + """造一个 tool_call。 + + ★ `index` 字段**必须给**，且要与它在数组里的位置一致。 + + 内核靠分片自带的 StreamIndex（即上游 JSON 里的 "index"）分槽累积 + arguments（process.go:347 `idx := tc.StreamIndex`）。缺 index 时所有 + 分片都落到槽 0，几个 tool_call 的 arguments 被**混拼**在一起 —— + 症状是每个工具都报「参数不是合法 JSON」，而工具一次都没真跑过。 + + ★ 这个坑很隐蔽：单 tool_call 时不设 index 也正常（只有一个槽）， + 所以老 mock 一直没暴露问题；一旦发多个就全崩。 + """ + return {"index": idx, "id": "call_batch_%d" % idx, "type": "function", + "function": {"name": name, "arguments": json.dumps(args, ensure_ascii=False)}} + + +def slow_tool_calls(n, sleep_ms): + """造 n 个 sleep 型 cmd_run —— 用于量化串行 vs 并发的差异。 + + 为什么用 sleep 而不是真跑命令： + · 延迟可精确预期（不用去猜命令要多久）； + · 不产生外部副作用（不写文件、不动网络）； + · 不会因机器负载而失真。 + 每个工具的 sleep 时长**故意错开**（递增），这样能验证落消息顺序 + 按声明序而非完成序 —— 完成的顺序是反的（大的先完成）。 + """ + tcs = [] + for i in range(n): + # 时长随索引递增 ⇒ 完成顺序与声明顺序**相反** + ms = sleep_ms + i * 5 + tcs.append(mk_tool_call(i, "cmd_run", { + "command": f"sleep {ms / 1000:.3f}; echo done-{i}", + "timeout": "30s", + })) + return tcs + + +def batch_tool_calls(n, mixed=False): + """造 n 个 tool_call；mixed=True 时夹一个未声明并发安全的工具。""" + tcs = [] + for i in range(n): + name = PARALLEL_TOOLS[i % len(PARALLEL_TOOLS)] + if name == "knowledge_search": + args = {"query": "并发压测 q%d" % i, "top_k": 3} + elif name == "doc_query": + args = {"query": "并发压测 q%d" % i, "mode": "auto"} + elif name in ("person_query", "person_network"): + args = {"name": "压测人物%d" % i} + else: + args = {} + tcs.append(mk_tool_call(i, name, args)) + if mixed: + # 放在**中间**：确保降级判据不能靠"最后一个工具"侥幸通过 + tcs.insert(len(tcs) // 2, mk_tool_call(999, SERIAL_TOOL, + {"name": "压测/批内", "content": "mixed"})) + return tcs + class H(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" @@ -26,8 +137,8 @@ class H(BaseHTTPRequestHandler): self.wfile.write(b) def do_HEAD(self): - # 内核探活用 HEAD（见 internal/network/monitor.go CheckOnce）： - # 不实现它 → BaseHTTPRequestHandler 回 501 → 判为不可达 → agent degraded/rollback。 + # 内核会拿 HEAD 探活。必须实现它 —— 不实现则 BaseHTTPRequestHandler + # 回 501 → 判为不可达 → agent degraded/rollback。 self.send_response(200) self.send_header("Content-Length", "0") self.end_headers() @@ -46,21 +157,70 @@ class H(BaseHTTPRequestHandler): text = c if isinstance(c, str) else json.dumps(c, ensure_ascii=False) break + has_tool_msg = any(m.get("role") == "tool" for m in msgs) + + # ---- 故障注入 ---- + if "!err" in text and not has_tool_msg: + self._json({"error": {"message": "mock injected failure", "type": "server_error"}}, 500) + return + if "!hang" in text and not has_tool_msg: + # 只回一半就断开：压客户端超时与连接清理 + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", "999999") + self.end_headers() + self.wfile.write(b'{"id":"mock"') + self.wfile.flush() + time.sleep(30) + return + + # ---- 多模态 / ONNX 路径（不加载模型，只验内核接线与错误处理）---- + if ("!img" in text or "!ocr" in text) and not has_tool_msg: + name = "describe_image" if "!img" in text else "ocr_image" + return self._respond(body, content=None, + tool_calls=[mk_tool_call(1, name, {"path": "/nonexistent.png"})]) + + # ---- 批内并发：!batchN / !mixedN ---- + if not has_tool_msg: + ms = _RE_SLOW.search(text) + if ms: + cnt = max(2, min(int(ms.group(1)), 32)) + sm = int(os.environ.get("MOCK_TOOL_SLEEP_MS", "200")) + return self._respond(body, content=None, + tool_calls=slow_tool_calls(cnt, sm)) + + mb = _RE_BATCH.search(text) + mm = _RE_MIXED.search(text) + if mb or mm: + cnt = int((mb or mm).group(1)) + cnt = max(2, min(cnt, 32)) # 下限 2（1 个不会并发），上限 32 防止把 mock 压成瓶颈 + return self._respond(body, content=None, + tool_calls=batch_tool_calls(cnt, mixed=mm is not None)) + # 标记 !resident ⇒ 回一个工具调用，用于在**真实内核**里驱动驻留子工具链。 - if "!resident" in text and not any(m.get("role") == "tool" for m in msgs): - tc = {"id": "call_mock_1", "type": "function", + if "!resident" in text and not has_tool_msg: + tc = {"index": 0, "id": "call_mock_1", "type": "function", "function": {"name": "resident_agents", "arguments": json.dumps({"action": "create", "id": "r1", "task_prompt": "驻留子任务：统计一下 !notify", "input_chs": "cli"}, ensure_ascii=False)}} return self._respond(body, content=None, tool_calls=[tc]) - if "!notify" in text and not any(m.get("role") == "tool" for m in msgs): - tc = {"id": "call_mock_2", "type": "function", + if "!notify" in text and not has_tool_msg: + tc = {"index": 0, "id": "call_mock_2", "type": "function", "function": {"name": "notify_parent", "arguments": json.dumps({"text": "mock 汇报：子已完成统计"}, ensure_ascii=False)}} return self._respond(body, content=None, tool_calls=[tc]) return self._respond(body, content="mock-ok:" + text[:40]) + def _chunk(self, b): + """写一个 HTTP/1.1 chunked 块。 + + ★ 我重写本文件时把原有的 _chunk 漏掉了 —— 而流式路径每一帧都要调它， + 漏掉的表现是 AttributeError，**只在 stream=true 时才炸**。 + 非流式路径照跑，所以粗看"能用"，一开流式就崩。 + """ + self.wfile.write(("%x\r\n" % len(b)).encode() + b + b"\r\n") + def _respond(self, body, content=None, tool_calls=None): if body.get("stream"): self.send_response(200) @@ -73,7 +233,38 @@ class H(BaseHTTPRequestHandler): "model": "mock", "choices": [{"index": 0, "delta": delta}]}) self._chunk(("data: " + data + "\n\n").encode()) if tool_calls: - emit({"role": "assistant", "tool_calls": tool_calls}) + # ★ 必须按 OpenAI 流式协议**分片**发：一个 chunk 一个 tool_call， + # 各自带 index；后续 chunk 只续 arguments。 + # + # 我第一版把整个数组塞进**一个** chunk，内核按"续传"语义累积 + # arguments（process.go:364 acc.argsRaw.WriteString）—— + # 结果 4 个 tool_call 的参数被**混拼**到槽 0， + # 每个工具都报"参数不是合法 JSON"，而工具一次都没真跑过。 + # + # 症状离原因很远：看起来像"内核不支持多工具调用"， + # 实际是我没按协议发。 + # OpenAI 真实语义：**每个** tool_call 都先发一片带 + # name 的首片，再发续传片。 + # + # 我第一版只给第 0 个发首片、其余直接发续传片，看起来省事， + # 但内核 flush 时按「无 name 即丢弃」处理（process.go:253 + # `flushed with EMPTY name`），于是 idx=1/2/3 三个分片 + # 全部被丢 ⇒ 只跑 1 个工具。 + # + # 症状：内核日志里 idx 分对了，却只有一个 tool_call 活下来 —— + # 看起来像"index 透传修好了但还有别的问题"。 + for tc in tool_calls: + emit({"tool_calls": [{ + "index": tc["index"], + "id": tc["id"], + "type": "function", + "function": {"name": tc["function"]["name"], + "arguments": ""}, + }]}) + emit({"tool_calls": [{ + "index": tc["index"], + "function": {"arguments": tc["function"]["arguments"]}, + }]}) time.sleep(DELAY_MS / 1000.0) if content: emit({"role": "assistant", "content": content[: max(1, len(content) // CHUNKS)]}) @@ -90,13 +281,15 @@ class H(BaseHTTPRequestHandler): msg["content"] = None time.sleep(DELAY_MS / 1000.0) self._json({"id": "mock", "object": "chat.completion", "model": "mock", - "choices": [{"index": 0, "message": msg, "finish_reason": "stop"}], - "usage": {"prompt_tokens": 10, "completion_tokens": 10, "total_tokens": 20}}) - - def _chunk(self, b): - self.wfile.write(("%x\r\n" % len(b)).encode() + b + b"\r\n") + "choices": [{"index": 0, "message": msg, "finish_reason": "tool_calls" if tool_calls else "stop"}]}) if __name__ == "__main__": - port = int(os.environ.get("MOCK_PORT", "9099")) + import sys + # 端口：先位置参数（原 mockllm.py 单独跑时用法），再 MOCK_PORT（launch.sh 用它）， + # 最后默认 9099。★ 三个来源都要留 —— 只认一个会破坏另外两个调用方。 + if len(sys.argv) > 1: + port = int(sys.argv[1]) + else: + port = int(os.environ.get("MOCK_PORT", "9099")) ThreadingHTTPServer(("127.0.0.1", port), H).serve_forever()