Repository navigation
Feature: 新增 trpc_agent_bench 压测套件 - #366
Conversation
weimch
commented
Nov 2, 2026
weimch
commented
- 参考Agno，新增压测套件及对应单测；用于测框架自身开销：import、构造、单轮 run、多轮会话、带工具的 Agent 执行和内存占用；Agent执行不会走实际的LLM调用，而是走一个Mock的llm-server
- 设计文档：docs/design/01-bench-design.md
- 使用文档：docs/mkdocs/zh/bench.md
- 补充llmagent的基础压测数据
AI Code Review审查结论通过 审查范围： 发现的问题中等
问题： 触发条件： 实际影响： mock OpenAI server 进程与 session service、agent runtime 不释放——server 是 daemon 进程，benchmark 子进程退出时会被回收，但进程存活期间的资源（内存、连接）持续占用；且错误信息中没有任何“服务未清理”的提示，难以排查。 修正方向： 将 中等
问题： 触发条件： mock 流式响应在 final 事件前中断（如客户端断连、 实际影响： 本应失败（turn 未完成）的一次 benchmark iteration 被当作成功计入统计， 修正方向： 将“缺失 final 事件”视为失败——当事件列表非空且没有任何 中等
问题： 触发条件： 工具执行失败但以非 error 事件呈现（例如 实际影响： 修正方向： 逐个断言每个 call 都能在 results 中找到 中等
问题： 触发条件： 目标 实际影响： 修正方向： 给 较低
问题： 触发条件： 在未清空 实际影响： 报告中出现过期/无对应 修正方向： 每次运行开始时清理 较低
问题： 触发条件： 任意一次 benchmark iteration 的 func 抛异常（工具执行失败、 实际影响： 该进程后续所有 修正方向： 将 |
| box: BenchSetup = await setup(script) | ||
| handle = await box.target.handle( | ||
| box.prepared, | ||
| tools_factory=tools_factory or (lambda: []), | ||
| ) | ||
| try: |
There was a problem hiding this comment.
问题： _run_turns 中 box = await setup(script) 成功启动 mock server 并完成 prepare() 之后，handle = await box.target.handle(...) 位于 try/finally 之外；tools.py:46-47 与 tools_parallel.py:56-57 也是同样模式。handle() 内部会执行 instantiate()（见 bench/llmagent/base.py:132-139），一旦构造 agent 抛异常，finally 中的 teardown(box) 不会执行。
触发条件： LlmAgentTarget.handle() 或任意自定义 target 的 handle()/instantiate() 抛异常时（如工具工厂失败、模型配置错误、agent 构造校验失败）。setup() 自身的异常路径（_setup.py:46-51）已正确关闭 server，但这里不在该保护范围内。
实际影响： mock OpenAI server 进程与 session service、agent runtime 不释放——server 是 daemon 进程，benchmark 子进程退出时会被回收，但进程存活期间的资源（内存、连接）持续占用；且错误信息中没有任何“服务未清理”的提示，难以排查。
修正方向： 将 handle 创建移入 try 内（box = await setup(script); try: handle = await ...; ... finally: await teardown(box)），保证任何构造失败都走 teardown 关闭 server 与 session service。
| turn_complete = any( | ||
| getattr(event, "turn_complete", False) or ( | ||
| callable(getattr(event, "is_final_response", None)) and event.is_final_response()) for event in event_list) | ||
| if not turn_complete and not has_non_partial_content and not partial: | ||
| raise RuntimeError("turn did not complete") |
There was a problem hiding this comment.
问题： ensure_turn_completed 的完成判定（_eval.py:229-233）在“没有 final 事件”时依赖 partial 文本兜底：if not turn_complete and not has_non_partial_content and not partial: raise。只有 partial 事件、没有任何 is_final_response()/turn_complete 事件时不会报“turn did not complete”，且 final_text 回退到 "".join(partial)。mock 的 _sse_stream 总是有最终非 partial 事件，但流被截断（连接中断、chunk 缺失）时可能只产生 partial 事件。
触发条件： mock 流式响应在 final 事件前中断（如客户端断连、_sse_stream 中途异常、chunk_delay_ms 非零时的时序问题），或自定义 benchmark 手动构造仅含 partial 事件的流。
实际影响： 本应失败（turn 未完成）的一次 benchmark iteration 被当作成功计入统计，expected_text 检查也可能被部分文本偶然匹配而漏报，压测结果把“错误路径当成了正常结果”，与设计文档声称的“确定性 + 正确性断言”目标相悖。
修正方向： 将“缺失 final 事件”视为失败——当事件列表非空且没有任何 is_final_response()/非 partial 内容事件时直接 raise，而不是用 partial 文本兜底；final_text 只取非 partial 事件文本。
|
|
||
| if expect_tool_success: | ||
| if not calls or not results: | ||
| raise RuntimeError("tool loop did not execute") | ||
| for response in results: | ||
| payload = getattr(response, "response", None) | ||
| if isinstance(payload, dict) and payload.get("error"): | ||
| raise RuntimeError("tool call failed") |
There was a problem hiding this comment.
问题： _eval.py:238-245 的 expect_tool_success 只检查“存在 call 且存在 result、result 的 response dict 无 error 键”，没有把每个 function_response.id 与对应的 function_call.id 关联校验；tools_parallel 的“3 个响应”计数（tools_parallel.py:72-75）也只数部分数量。若某个工具失败只产生 function_response 缺失或 id 不匹配的事件（如 tool_not_found 错误事件带非 partial 内容），该 turn 会被判定为工具执行成功。
触发条件： 工具执行失败但以非 error 事件呈现（例如 _tools_processor.py:752-768 生成的错误事件带 function_response 但 error_code 键在 response.error 之外的变体）、或工具未执行但 ensure_turn_completed 已被 partial 文本兜底放过。
实际影响： tools、tools_parallel、loop5/loop10 等工具类 benchmark 可能在工具循环实际失败时仍通过断言并产出“成功”的耗时/内存样本，压测数据被静默污染。
修正方向： 逐个断言每个 call 都能在 results 中找到 id 相同的 function_response，并统一通过 event.is_error()/error_code 判定工具失败，而不是只检查 response dict 的 error 键。
| def _run(command: list[str]) -> float: | ||
| started = time.perf_counter() | ||
| completed = subprocess.run(command, capture_output=True, text=True, check=False) |
There was a problem hiding this comment.
问题： import_time._run（import_time.py:20-26）的 subprocess.run 没有 timeout；main() 对每个 sample 启动 python -c "import <target>"，共 15+15 次子进程。若被测包 import 时挂起（网络请求、交互提示、死锁、或冷启动时较慢），整个 benchmark 会无限阻塞。
触发条件： 目标 import_module 的 import 副作用耗时超长或永久挂起（自定义 target 只约束了导出 TARGET，import 内容无限制）。
实际影响： run_bench 的该子进程永不退出，subprocess.run(check=True) 一直等待，整个压测 run 无界挂起，且无任何超时信息可排查；这是工具套件最恶劣的失败模式之一。
修正方向： 给 subprocess.run 增加 timeout=（如 60s 或可配置），捕获 TimeoutExpired 后 raise 明确错误并结束该 benchmark；同时可将超时值写入 BENCH_* 环境变量便于自定义。
| ) -> dict[str, Any]: | ||
| results: list[dict[str, Any]] = [] | ||
| for target in targets: | ||
| for path in sorted((results_root / target).glob("*.json")): |
There was a problem hiding this comment.
问题： run_bench._collect（run_bench.py:104-118）直接 glob 目标目录下所有 *.json 并把它们全部写进 summary.json，build_report 同样收集全部既有 JSON。两次运行间残留的旧结果（包括更早的 loop10_c2.json、memory_per_agent.json 或用户手动放置的任意 JSON）会被当作本次运行的结果收集进报告。
触发条件： 在未清空 results/<target>/ 的情况下重复运行 run_bench（设计文档明确支持“同一套脚本可重复跑”），或前半段 benchmark 失败、run_bench 异常退出，旧文件与新文件混合。
实际影响： 报告中出现过期/无对应 run_bench 配置的结果行（如同一 name 多 target 的错乱行、缺 baseline 的孤立行），对比表数值失真，压测结论被静默污染。
修正方向： 每次运行开始时清理 results_root 下本次 targets 的目录（或为每次 run 生成独立子目录/带 run-id 的目录），_collect 只收集本次写入的文件。
| tracemalloc.stop() | ||
| return max(0.0, peak - baseline) / (1024.0 * 1024.0) | ||
|
|
||
| async def _memory_async(self, baseline: float) -> float: | ||
| gc.collect() | ||
| tracemalloc.start() | ||
| await self.func() | ||
| _, peak = tracemalloc.get_traced_memory() | ||
| tracemalloc.stop() |
There was a problem hiding this comment.
问题： _eval.py:171-177（_memory_sync）与 _eval.py:181-186（_memory_async）中，tracemalloc.start() 与 tracemalloc.stop() 之间调用 self.func()/await self.func()，且 tracemalloc.stop() 不在 finally 中；memory_footprint._per_agent_footprint（memory_footprint.py:27-32）同样如此。tracemalloc.start() 是幂等的（重复调用无报错），一旦 func 抛异常，stop() 被跳过，tracing 保持开启状态。
触发条件： 任意一次 benchmark iteration 的 func 抛异常（工具执行失败、ensure_turn_completed 断言失败、网络错误等）。
实际影响： 该进程后续所有 tracemalloc.get_traced_memory() 结果继续叠加此前泄漏的分配，且若进程未立即退出（如测试或长驻 runner 中复用 PerformanceEval），后续 memory pass 的数值被污染；即使进程随即退出，也没有任何显式清理提示。
修正方向： 将 tracemalloc.stop() 放入 finally 块（或在异常路径显式 stop()），确保任何情况下 tracing 都被关闭后再向上抛异常。
ca09227 to
1b4c7b4
Compare
- 参考Agno，新增压测套件及对应单测；用于测框架自身开销：import、构造、单轮 run、多轮会话、带工具的 Agent 执行和内存占用；Agent执行不会走实际的LLM调用，而是走一个Mock的llm-server - 设计文档：docs/design/01-bench-design.md - 使用文档：docs/mkdocs/zh/bench.md - 补充llmagent的基础压测数据
1b4c7b4 to
6ea849d
Compare
AI Code Review审查结论通过 审查范围：base 594bfb0..head 1b4c7b4（52 个文件，+6378/-1），即新增 trpc_agent_bench 压测套件。已完整阅读 run_bench 子进程编排、MockOpenAIServer（fastapi/uvicorn 独立进程）、_target/_setup/_profile 协议层、_eval 测量与断言、_report 报告生成、12 个内置 benchmark、bench/llmagent 目标适配及其与 trpc_agent_sdk 的 Runner/OpenAIModel/LlmAgent/InMemorySessionService/事件流（partial/turn_complete/is_final_response）等调用链，并核对设计文档与 13 个测试文件。计划符合性：套件结构、内置压测项、mock 注入、进程隔离、统计口径（median/p95、时间内存分开测）、报告与文档均符合设计；pyproject 将 trpc_agent_bench 纳入 wheel 且 bench extra 依赖 fastapi 成立（uvicorn 在核心依赖中）。主要风险：并发 发现的问题中等
问题： 触发条件： 使用文档/README 主推的 实际影响： 修正方向： 在 中等
问题： 触发条件： 在同一 实际影响： 修正方向： 每次运行开始时按 target 清空 较低
问题： 触发条件： 用户（或 CI 环境）设置了非数字的 实际影响： 压测子进程在 修正方向： 用与 较低
问题： 触发条件： 在 Windows 上运行压测套件（fork 上下文不存在， 实际影响： 套件被打包为跨平台 wheel（pyproject 支持 3.10-3.12，无平台限制），但在 Windows 上整套压测无法运行；macOS 上 mock server 可能无法在 20 秒内上报端口， 修正方向： 通过 |
| def _name_key(name: str) -> tuple[int, int, list[Any]]: | ||
| for prefix, rank in _ORDER: | ||
| if prefix in ("turn", "loop"): | ||
| if not name.startswith(prefix): | ||
| continue | ||
| suffix = name[len(prefix):] | ||
| if suffix and suffix[0].isdigit(): | ||
| return rank, 0, _natural_key(name) | ||
| continue | ||
| if name == prefix or name.startswith(f"{prefix}_"): | ||
| return rank, 0, _natural_key(name) | ||
| return 7, 0, _natural_key(name) |
There was a problem hiding this comment.
问题： _report._name_key（_report.py 147-158 行）对带并发后缀的结果名分组失效：turn1_c8、turn5_c8 这类名字的 suffix 是 _c8，suffix[0] 为 _ 而不是数字，命中 suffix and suffix[0].isdigit() 的分支不会成立，最终落入默认 rank 7（"自定义 benchmark"）。
触发条件： 使用文档/README 主推的 --concurrency 1,8,16,32 运行 turn1/turn5/turn10、loop1/loop5/loop10 后生成报告，表中会同时出现基础名结果和 _cN 后缀结果。
实际影响： turn1_c8、loop5_c16 等并发结果被排在表格末尾的自定义区段，而不是紧挨对应的 turn1/loop5 行；报告 Notes 承诺的 "turn1/turn5/turn10 和 loop1/loop5/loop10 分别放在一起" 被破坏，并发对比表难以对照阅读，压测主场景的报告排版与实际不一致。
修正方向： 在 _name_key 中先将 _c<数字> 后缀剥离后再做前缀匹配（例如对 name 先 re.sub(r'_c\d+$', '', name) 再按现有逻辑分组），或对 turn/loop 前缀额外接受 _cN 形式的后缀，使并发变体与基础项同组排序。
| def _collect( | ||
| results_root: Path, | ||
| targets: list[str], | ||
| baseline: str | None, | ||
| ) -> dict[str, Any]: | ||
| results: list[dict[str, Any]] = [] | ||
| for target in targets: | ||
| for path in sorted((results_root / target).glob("*.json")): | ||
| results.append(json.loads(path.read_text(encoding="utf-8"))) | ||
| return { | ||
| "results_dir": str(results_root), | ||
| "targets": targets, | ||
| "baseline": baseline, | ||
| "results": results, | ||
| } |
There was a problem hiding this comment.
问题： run_bench 每次运行前不清理 results_root/<target>/ 目录，而 _collect（run_bench.py 104-118 行）与 build_report 的 _load（_report.py 86-97 行）都以 glob("*.json") 全量读取目标目录下所有结果文件。
触发条件： 在同一 --results-dir 下先跑完整 12 项或 --concurrency 1,8，之后再跑只含 turn1 或不同并发集合的运行，旧运行遗留的 turn5.json、turn10_c8.json 等文件不会被删除。
实际影响： summary.json、report.md、report.html 会把上一次运行（可能对应不同代码版本或不同并发配置）的结果当作本次运行数据混合展示，甚至出现同一 benchmark 新旧两行、比较表基线错配，压测数字失真且难以察觉。
修正方向： 每次运行开始时按 target 清空 results_root/<target>（或先记录本次将要写入的结果名，收集/出报告时只纳入本次生成的文件），并在写 summary.json 时标注运行时间供核对。
|
|
||
| DESCRIPTION = ("批量构造 agent 并一直持有，不释放。" | ||
| "memory = 每个常驻 agent 占用的内存，不是构造过程中的瞬时峰值。") | ||
| AGENTS_PER_SAMPLE = max(1, int(os.environ.get("BENCH_MEMORY_AGENTS", "1000"))) |
There was a problem hiding this comment.
问题： memory_footprint.py 第 21 行直接 int(os.environ.get("BENCH_MEMORY_AGENTS", "1000"))，环境变量值未做类型校验；该变量在文档/README 中也没有任何说明。
触发条件： 用户（或 CI 环境）设置了非数字的 BENCH_MEMORY_AGENTS（例如空字符串、auto），运行 memory_footprint 压测项时模块导入阶段即抛 ValueError。
实际影响： 压测子进程在 _runner 加载模块时直接崩溃，run_bench 因 check=True 整体失败退出，且报错只是裸的 int() 异常，用户无法判断是哪个配置出了问题。
修正方向： 用与 _positive_int/_parse_concurrency 一致的校验方式解析该变量，解析失败时给出明确错误信息（或回退默认值并告警），并在文档中登记该环境变量的含义。
| self._connection: Any = None | ||
|
|
||
| def start(self) -> str: | ||
| context = multiprocessing.get_context("fork") |
There was a problem hiding this comment.
问题： MockOpenAIServer.start()（_model.py 355-375 行）使用 multiprocessing.get_context("fork") 硬编码 fork 启动方式。
触发条件： 在 Windows 上运行压测套件（fork 上下文不存在，get_context("fork") 直接抛 ValueError）；在 macOS 上从已导入 fastapi/openai 等线程化库的进程中 fork，存在子进程挂起风险。
实际影响： 套件被打包为跨平台 wheel（pyproject 支持 3.10-3.12，无平台限制），但在 Windows 上整套压测无法运行；macOS 上 mock server 可能无法在 20 秒内上报端口，start() 抛 "mock OpenAI server did not report a port"。
修正方向： 通过 multiprocessing.get_context("spawn")（或默认上下文）启动子进程，并把 spec/model_conf 以参数或序列化方式传入，避免依赖 fork 语义。
AI Code Review审查结论通过 审查范围： 计划符合性：套件按设计文档落地——Mock 模型驱动真实 agent loop、按 (target, benchmark, concurrency) 子进程隔离、warmup→timed→memory 分离、median/p95 统计、多 target 对比报告。经代码级验证：mock 的 SSE 线格式与 SDK 流式消费（ 主要风险：(1) 会话型 benchmark 的内存指标被跨迭代累积、从未释放的会话驻留污染（MODERATE）；(2) mock 的 测试充分性：13 个测试文件覆盖编排参数、请求定位、统计、报告、mock 线协议与配置校验，但无真实 agent 对 mock 服务的集成测试。 门禁结论：未发现 SEVERE 级别问题，变更与计划相符， 发现的问题中等
问题： 会话型内置 benchmark（ 触发条件： 以 实际影响： 报告的 修正方向： 让每个采样度量独立过程：在每次迭代完成后删除本次会话（通过 较低
问题： 触发条件： 自定义 benchmark 或错误路径测试期望通过脚本化的 实际影响： 错误注入只能依赖脚本越界（ 修正方向： 在 较低
问题： 依赖 SDK 真实执行路径的 benchmark（ 触发条件： SDK 侧行为变化（事件部分标志、 实际影响： 修正方向： 新增端到端集成测试：真实构造 |
| async def one_session(index: int) -> None: | ||
| user_id = _session_user_id(base_user_id, index) | ||
| session_id = new_session_id() | ||
| for turn in script: | ||
| events = await drain( | ||
| handle.run_turn( | ||
| app_name=app_name, | ||
| user_id=user_id, | ||
| session_id=session_id, | ||
| message="Say ok.", | ||
| ), ) | ||
| ensure_turn_completed( | ||
| events, | ||
| expected_text=_expected_text(turn), | ||
| expect_tool_success=_has_function(turn), | ||
| ) | ||
|
|
||
| async def arun_conversation() -> None: | ||
| await asyncio.gather(*(one_session(index) for index in range(current))) |
There was a problem hiding this comment.
问题： 会话型内置 benchmark（turn1/5/10、loop1/5/10、tools_parallel）的每个采样迭代都会通过 new_session_id() 创建一个全新的会话 id，而 bench/llmagent/base.py 创建的 InMemorySessionService 默认 TTL 为 24 小时、close() 只停止后台清理任务并不清空 _sessions；tools.py 的 tools benchmark 每轮采样同样新建会话。整个内存采样阶段（PerformanceEval._memory_async 执行 --iterations 次）中，所有迭代产生的会话与完整事件历史（各轮 SSE 文本、工具调用记录及 get_session 中 filter_events 的深拷贝）持续驻留，后一次采样的 tracemalloc 峰值必然包含此前全部会话累积触发的分配。
触发条件： 以 --iterations N（N>1）运行任一上述 benchmark：第 1 次采样的峰值只含本次会话的分配，第 N 次采样前服务里已保留 N-1 个会话，且每次 run_turn 都要对服务内全部事件记录做深拷贝，峰值随 N 放大。
实际影响： 报告的 memory_usages 与 median_memory_usage 随 --iterations 变化且系统性偏大，baseline 与 target 之间的内存对比失真；残留会话还持续占用真实堆，令 timed pass 的真实内存同样随迭代增长，违背指标描述中"本次采样中 tracemalloc 记录的峰值"的独立采样语义。
修正方向： 让每个采样度量独立过程：在每次迭代完成后删除本次会话（通过 AgentHandle 暴露 delete_session 或等效清理接口），或在每个内存样本前重建 session service（重新执行 setup()/teardown()），使每个样本只包含自身的分配峰值。
| if model_reply.error: | ||
| raise HTTPException(status_code=500, detail=model_reply.error) |
There was a problem hiding this comment.
问题： chat() 中的 if model_reply.error: 分支是死代码：ModelReply.error（_model.py:56）没有任何赋值路径，reply_from_script 只构造文本或工具调用两类回复，因此该分支永不触发，Mock 服务的"脚本化错误注入"能力实际不存在。
触发条件： 自定义 benchmark 或错误路径测试期望通过脚本化的 ModelReply.error 触发 HTTP 500（进而验证 SDK 错误事件与 ensure_turn_completed 的报错路径）时，该分支不会执行。
实际影响： 错误注入只能依赖脚本越界（IndexError → 500），无法确定性构造模型错误响应，SDK 侧错误事件的端到端验证缺少可控手段。
修正方向： 在 _reply_spec/reply_from_script 中支持错误构造（例如允许 TextBlock 携带 error 或新增错误块类型），使 ModelReply.error 可被赋值并覆盖该 500 分支；或删除该分支并在文档中明确错误注入方式。
| def make_tools() -> List[BaseTool]: | ||
| return [FunctionTool(add_numbers)] | ||
|
|
||
|
|
||
| async def _run() -> None: | ||
| box = await setup(TURN_SCRIPT) | ||
| handle = await box.target.handle(box.prepared, tools_factory=make_tools) | ||
|
|
||
| async def arun_agent() -> None: | ||
| events = await drain( | ||
| handle.run_turn( | ||
| app_name=box.conf.app_name, |
There was a problem hiding this comment.
问题： 依赖 SDK 真实执行路径的 benchmark（tools、tools_parallel、loop*）没有任何端到端测试：现有单测用 SimpleNamespace 伪造事件直接调用 ensure_turn_completed，或用独立的最小客户端验证 mock 的 SSE 线格式，从未用真实的 LlmAgent + OpenAIModel 连接 MockOpenAIServer 跑完一轮完整会话。
触发条件： SDK 侧行为变化（事件部分标志、turn_complete/is_final_response 语义、工具调用 id 匹配、_merge_parallel_function_response_events 合并逻辑）回归时，现有测试无法在合入前捕获。
实际影响： tools_parallel 依赖"3 个 FunctionBlock 合并为单个含 3 个 function_response part 的事件"这一 SDK 内部行为（responses < 3 断言），ensure_turn_completed 则依赖"SDK 从不置 turn_complete，只能靠 is_final_response() 判定完成"这一隐含约定；一旦 SDK 变更，套件可能只在真实压测中才暴露失败，且 mock 服务 500 → 错误事件 → RuntimeError 的失败路径同样未经测试。
修正方向： 新增端到端集成测试：真实构造 LlmAgent/OpenAIModel（base_url 指向 mock server）执行 tools/tools_parallel 脚本，断言工具链执行、返回事件数及 ensure_turn_completed 通过；并补测 HTTP 500 场景下错误事件被正确识别并抛出的路径。
