jianf/HomeAgent
1
0
Fork 0
mirror of https://gitcode.com/JianFeeeee/HomeAgent.git synced 2026-10-14 19:43:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(stress): 补批内并发压测与 A/B 对比脚本，并给 mock 加多 tool_call 能力

Browse Source 
## 缺口

现有 kernel-stress 只压**调度器**（多连接排队 + L4 中断 + 驻留子），mock 每轮
只发**一个** tool_call。而内核的并发判据是：

    if f == nil || len(f.PendingTools) <= 1 { return false }

⇒ **一个 tool_call 永远不并发**。所以现有压测压的全是串行路径，批内并发
一条都没走过。

## mockllm.py：让 mock 能发多个 tool_call

- `!batchN`   N 个全部 ParallelSafe 的只读工具 ⇒ 强制走 stepToolBatch
- `!mixedN`   夹一个 knowledge_create（未声明并发安全）⇒ 验证整批降级
- `!slowbatchN` N 个 sleep 型 cmd_run（单工具约 200ms，时长递增）
  ⇒ 工具慢才能让并发的收益从噪声里显出来；时长递增使**完成序与声明序相反**，
  可据此判定"是否按声明序落消息"
- `!serialbatchN` 在 !slowbatch 基础上插入一个非并发安全工具 ⇒ 强制整批串行，
  作为并发对照的另一半
- `!img` / `!ocr` 触发多模态工具路径（不加载模型，只验内核 IPC 接线与错误处理）
- `!err` / `!hang` 故障注入

### 三个协议细节（踩过才知道，都写在注释里）

1. **必须带 `index`** —— 内核按 `StreamIndex`（上游 JSON 的 index）分槽累积
   arguments。缺 index 时所有分片落到槽 0，几个 tool_call 的参数被**混拼**，
   症状是每个工具都报"参数不是合法 JSON"而工具一次没真跑过。
   单 tool_call 时不设 index 也正常，所以老 mock 一直没暴露。
2. **必须按协议分片** —— 一个 chunk 一个 tool_call，各自带 index；
   后续 chunk 只续 arguments。整数组塞进一个 chunk 会被内核按"续传"语义累积。
3. **每个工具都要发"带 name 的首片"** —— 我第一版只给第 0 个发首片、其余直接
   发续传片，看起来省事，但内核 flush 时按"无 name 即丢弃"处理，
   于是 idx=1/2/3 全被丢，只跑 1 个工具。

## batchstress.py：批内并发压测（带校验）

只看峰值是不够的 —— 校验：工具是否真跑（响应里应有结果标记）、
消息顺序是否稳定（并发执行但按索引落消息）、mixed 批是否整批降级。

## abtest.py：串行 vs 并发的定量对比

同一套内核上用 !slowbatch / !serialbatch 两组对照，交替执行抵消机器负载漂移，
取中位数（长尾会污染均值）。

★ 判据里加了"工具是否真执行"这一项：**只看耗时是不够的** —— 出现过
"内核 274ms 就回复、工具一个没跑"的情况，那种情况下并发与串行都是 0.2s，
加速比毫无意义。

## 实测（隔离 netns 实例，mock + 内核同网段，5 轮中位）

    N    并发        串行         加速     上限
    2    0.481s    0.685s      1.42x    2
    4    0.491s    1.114s      2.27x    4
    8    0.513s    2.037s      3.97x    8
    12   0.531s    3.039s      5.72x    12

并发批耗时几乎不随 N 增长，串行批严格线性。

★ 另一个踩过的坑（abtest 脚本自己的）：内核有输入去重
（task.go:353 `isDuplicateInput`，为 webui 断线重连重放而设），相同文本会被
丢弃并回空响应。第一版每轮发同一个 marker ⇒ 只有第 1 轮有效，后面全是
0 秒 0 工具。修法是每轮加 `time.time_ns()` 唯一后缀。
This commit is contained in:
JianFeeeee
2026-09-27 18:07:58 +08:00
parent 825bac3920
commit 71700f1739
3 changed files with 646 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

221
scripts/kernel-stress/mockllm.py
View File

@ -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()