diff --git a/docs/architecture.md b/docs/architecture.md index 50f4cf5091..e9b32c3335 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -244,6 +244,8 @@ Use the master entry to answer what should be benchmarked. Use runner config for Derived concurrency points, eval selection, topology defaults, names, and runner-derived facts belong in `generate_sweep_configs.py`. Workflows should forward matrix fields, not reimplement generator policy in expressions or shell. +The `full-sweep` and `test-config` commands share fixed-sequence and AgentX row builders. Command-specific selection remains in the callers; the AgentX builder owns worker defaults, offload budgets, experiment names, node counts, and validation. It validates topology and offload budgets before filtering concurrency, preserves point and runner ordering, and filters AgentX bounds without inventing a capped concurrency point. + ### Trigger selection is separate from configuration `perf-changelog.yaml` selects work and records why. It does not redefine a master entry. This makes the configuration catalog reusable while keeping a reviewable history of what each sweep intended to run. diff --git a/docs/architecture_zh.md b/docs/architecture_zh.md index 00753dd23a..856d326ef5 100644 --- a/docs/architecture_zh.md +++ b/docs/architecture_zh.md @@ -244,6 +244,8 @@ bash ./runners/launch_${RUNNER_NAME%%_*}.sh 派生并发点、评测选择、拓扑默认值、名称和运行器派生信息属于 `generate_sweep_configs.py`。工作流应转发矩阵字段，而不应在表达式或 Shell 中重新实现生成器策略。 +`full-sweep` 和 `test-config` 命令共用固定序列与 AgentX 的矩阵行构建逻辑。各命令的选择规则仍由调用方负责；AgentX 构建逻辑负责 worker 默认值、卸载预算、实验名称、节点数和验证。它在过滤并发度前验证拓扑与卸载预算，保留并发点和运行器的原有顺序，并仅按上下限过滤 AgentX 并发点，不会额外生成截断到上限的并发点。 + ### 触发选择与配置彼此独立 `perf-changelog.yaml` 选择工作并记录原因。它不会重新定义主条目。这样既可以复用配置目录，又能保留可供审查的历史记录，说明每次扫描打算运行什么。 diff --git a/utils/matrix_logic/generate_sweep_configs.py b/utils/matrix_logic/generate_sweep_configs.py index 2b864226f4..33a02da290 100644 --- a/utils/matrix_logic/generate_sweep_configs.py +++ b/utils/matrix_logic/generate_sweep_configs.py @@ -846,6 +846,123 @@ def _fixed_sequence_entries( return entries +def _agentic_entries( + config: dict, + benchmark: dict, + scenario: dict, + runners: list[str], + runner_data: dict, + *, + step_size: int = 2, + min_conc: int | None = None, + max_conc: int | None = None, + conc_filter: list[int] | None = None, +) -> list[dict]: + """Expand one AgentX deployment for either generator command. + + Resolve topology and the offload budget before filtering concurrency so an + invalid deployment still fails when its points are filtered out. Agentic + bounds filter existing points; they never introduce a capped point. + """ + is_multinode = config.get(Fields.MULTINODE.value, False) + disagg = config.get(Fields.DISAGG.value, False) + model_code = config[Fields.MODEL_PREFIX.value] + if is_multinode: + prefill, decode = multinode_worker_pair(benchmark, disagg) + kv_offloading = benchmark.get(Fields.KV_OFFLOADING.value, "none") + else: + tp = benchmark[Fields.TP.value] + pp = benchmark.get(Fields.PP.value, 1) + dcp_size = benchmark.get(Fields.DCP_SIZE.value, 1) + pcp_size = benchmark.get(Fields.PCP_SIZE.value, 1) + ep = benchmark.get(Fields.EP.value) + dp_attn = benchmark.get(Fields.DP_ATTN.value) + kv_offloading = benchmark[Fields.KV_OFFLOADING.value] + spec_decoding = benchmark.get(Fields.SPEC_DECODING.value, "none") + kv_offload_backend = benchmark.get(Fields.KV_OFFLOAD_BACKEND.value) + total_cpu_dram_gb = agentic_dram_offload_gb( + scenario, benchmark, config[Fields.RUNNER.value], runner_data) + + conc_values = benchmark.get(Fields.CONC_LIST.value) + if not conc_values: + conc_values = _concurrency_range( + benchmark[Fields.CONC_START.value], benchmark[Fields.CONC_END.value], step_size) + if min_conc is not None: + conc_values = [c for c in conc_values if c >= min_conc] + if max_conc is not None: + conc_values = [c for c in conc_values if c <= max_conc] + if conc_filter: + conc_values = [c for c in conc_values if c in conc_filter] + if not conc_values: + return [] + + # Multi-node batches are runner-major; single-node points are conc-major. + if is_multinode: + offload_suffix = ( + f"_{agentic_kv_offload_suffix(kv_offloading, kv_offload_backend)}" + if kv_offloading != "none" else "" + ) + points = ( + (runner, batch) for runner in runners + for batch in chunk_multinode_agentic_concurrencies(conc_values) + ) + else: + points = ((runner, conc) for conc in conc_values for runner in runners) + + entries = [] + for runner, conc in points: + entry = { + Fields.IMAGE.value: config[Fields.IMAGE.value], + Fields.MODEL.value: config[Fields.MODEL.value], + Fields.MODEL_PREFIX.value: model_code, + Fields.PRECISION.value: config[Fields.PRECISION.value], + Fields.FRAMEWORK.value: config[Fields.FRAMEWORK.value], + Fields.RUNNER.value: runner, + } + if is_multinode: + entry.update({ + Fields.SPEC_DECODING.value: spec_decoding, + Fields.PREFILL.value: prefill, + Fields.DECODE.value: decode, + Fields.CONC.value: conc, + }) + exp_name = multinode_agentic_exp_name( + model_code, prefill, decode, conc, offload_suffix) + else: + entry.update({ + Fields.TP.value: tp, + Fields.PP.value: pp, + Fields.DCP_SIZE.value: dcp_size, + Fields.PCP_SIZE.value: pcp_size, + Fields.EP.value: ep if ep is not None else 1, + Fields.DP_ATTN.value: dp_attn if dp_attn is not None else False, + Fields.SPEC_DECODING.value: spec_decoding, + Fields.CONC.value: conc, + }) + exp_name = ( + f"{model_code}_tp{tp}_conc{conc}_" + f"{agentic_kv_offload_suffix(kv_offloading, kv_offload_backend)}" + + (f"_spec-{spec_decoding}" if spec_decoding != "none" else "") + ) + entry.update({ + Fields.KV_OFFLOADING.value: kv_offloading, + Fields.TOTAL_CPU_DRAM_GB.value: total_cpu_dram_gb, + Fields.DURATION.value: DEFAULT_AGENTIC_DURATION_SECONDS, + Fields.EXP_NAME.value: exp_name, + }) + if is_multinode: + entry[Fields.DISAGG.value] = disagg + entry[Fields.SCENARIO_TYPE.value] = "agentic-coding" + if kv_offload_backend is not None: + entry[Fields.KV_OFFLOAD_BACKEND.value] = kv_offload_backend + entry.update(component_metadata(benchmark, config)) + if is_multinode: + add_multinode_node_count( + entry, runner_data, benchmark.get(Fields.NUM_NODES.value)) + entries.append(validate_agentic_matrix_entry(entry)) + return entries + + def generate_full_sweep(args, all_config_data, runner_data): """Generate full sweep configurations with optional filtering. @@ -903,18 +1020,11 @@ def generate_full_sweep(args, all_config_data, runner_data): # Check if this is a multinode config is_multinode = val.get(Fields.MULTINODE.value, False) - # Get disagg value, defaulting to False if not specified - disagg = val.get(Fields.DISAGG.value, False) scenarios = val[Fields.SCENARIOS.value] scenario_filter = set(args.scenario_type) if getattr(args, 'scenario_type', None) else None seq_len_configs = scenarios.get(Fields.FIXED_SEQ_LEN.value, []) if (scenario_filter is None or 'fixed-seq-len' in scenario_filter) else [] - image = val[Fields.IMAGE.value] - model = val[Fields.MODEL.value] - precision = val[Fields.PRECISION.value] - framework = val[Fields.FRAMEWORK.value] runner = val[Fields.RUNNER.value] - model_code = val[Fields.MODEL_PREFIX.value] # Compute filtered runner nodes for this config if filter is specified runner_nodes_to_use = None @@ -1061,121 +1171,12 @@ def generate_full_sweep(args, all_config_data, runner_data): if not is_multinode and not args.single_node: continue - for agentic_config in agentic_configs: - bmk_space = agentic_config[Fields.SEARCH_SPACE.value] - duration = DEFAULT_AGENTIC_DURATION_SECONDS - - for bmk in bmk_space: - if is_multinode: - prefill, decode = multinode_worker_pair(bmk, disagg) - spec_decoding = bmk.get(Fields.SPEC_DECODING.value, "none") - kv_offloading = bmk.get(Fields.KV_OFFLOADING.value, "none") - kv_offload_backend = bmk.get(Fields.KV_OFFLOAD_BACKEND.value) - else: - tp = bmk[Fields.TP.value] - pp = bmk.get(Fields.PP.value, 1) - dcp_size = bmk.get(Fields.DCP_SIZE.value, 1) - pcp_size = bmk.get(Fields.PCP_SIZE.value, 1) - ep = bmk.get(Fields.EP.value) - dp_attn = bmk.get(Fields.DP_ATTN.value) - spec_decoding = bmk.get(Fields.SPEC_DECODING.value, "none") - kv_offloading = bmk[Fields.KV_OFFLOADING.value] - kv_offload_backend = bmk.get(Fields.KV_OFFLOAD_BACKEND.value) - total_cpu_dram_gb = agentic_dram_offload_gb( - agentic_config, bmk, runner, runner_data) - - # Get concurrency values - conc_list = bmk.get(Fields.CONC_LIST.value) - if conc_list: - conc_values = conc_list - else: - conc_start = bmk[Fields.CONC_START.value] - conc_end = bmk[Fields.CONC_END.value] - conc_values = _concurrency_range(conc_start, conc_end, args.step_size) - - # Apply conc filters - if args.min_conc is not None: - conc_values = [c for c in conc_values if c >= args.min_conc] - if args.max_conc is not None: - conc_values = [c for c in conc_values if c <= args.max_conc] - if not conc_values: - continue - - runners_for_entry = runner_nodes_to_use if runner_nodes_to_use else [runner] - - if is_multinode: - # Preserve historical exp-names for the default (no offload) - # case; only append a suffix when KV offloading is active. - offload_suffix = ( - f"_{agentic_kv_offload_suffix(kv_offloading, kv_offload_backend)}" - if kv_offloading != "none" - else "" - ) - for runner_value in runners_for_entry: - for conc_batch in chunk_multinode_agentic_concurrencies(conc_values): - entry = { - Fields.IMAGE.value: image, - Fields.MODEL.value: model, - Fields.MODEL_PREFIX.value: model_code, - Fields.PRECISION.value: precision, - Fields.FRAMEWORK.value: framework, - Fields.RUNNER.value: runner_value, - Fields.SPEC_DECODING.value: spec_decoding, - Fields.PREFILL.value: prefill, - Fields.DECODE.value: decode, - Fields.CONC.value: conc_batch, - Fields.KV_OFFLOADING.value: kv_offloading, - Fields.TOTAL_CPU_DRAM_GB.value: total_cpu_dram_gb, - Fields.DURATION.value: duration, - Fields.EXP_NAME.value: multinode_agentic_exp_name( - model_code, prefill, decode, conc_batch, offload_suffix - ), - Fields.DISAGG.value: disagg, - Fields.SCENARIO_TYPE.value: "agentic-coding", - } - if kv_offload_backend is not None: - entry[Fields.KV_OFFLOAD_BACKEND.value] = kv_offload_backend - entry.update(component_metadata(bmk, val)) - add_multinode_node_count( - entry, - runner_data, - bmk.get(Fields.NUM_NODES.value), - ) - validate_agentic_matrix_entry(entry) - matrix_values.append(entry) - else: - for conc in conc_values: - for runner_value in runners_for_entry: - entry = { - Fields.IMAGE.value: image, - Fields.MODEL.value: model, - Fields.MODEL_PREFIX.value: model_code, - Fields.PRECISION.value: precision, - Fields.FRAMEWORK.value: framework, - Fields.RUNNER.value: runner_value, - Fields.TP.value: tp, - Fields.PP.value: pp, - Fields.DCP_SIZE.value: dcp_size, - Fields.PCP_SIZE.value: pcp_size, - Fields.EP.value: ep if ep is not None else 1, - Fields.DP_ATTN.value: dp_attn if dp_attn is not None else False, - Fields.SPEC_DECODING.value: spec_decoding, - Fields.CONC.value: conc, - Fields.KV_OFFLOADING.value: kv_offloading, - Fields.TOTAL_CPU_DRAM_GB.value: total_cpu_dram_gb, - Fields.DURATION.value: duration, - Fields.EXP_NAME.value: ( - f"{model_code}_tp{tp}_conc{conc}_" - f"{agentic_kv_offload_suffix(kv_offloading, kv_offload_backend)}" - + (f"_spec-{spec_decoding}" if spec_decoding != "none" else "") - ), - Fields.SCENARIO_TYPE.value: "agentic-coding", - } - if kv_offload_backend is not None: - entry[Fields.KV_OFFLOAD_BACKEND.value] = kv_offload_backend - entry.update(component_metadata(bmk, val)) - validate_agentic_matrix_entry(entry) - matrix_values.append(entry) + for scenario in agentic_configs: + for benchmark in scenario[Fields.SEARCH_SPACE.value]: + matrix_values.extend(_agentic_entries( + val, benchmark, scenario, runner_nodes_to_use or [runner], runner_data, + step_size=args.step_size, min_conc=args.min_conc, max_conc=args.max_conc, + )) return matrix_values @@ -1211,19 +1212,12 @@ def generate_test_config_sweep(args, all_config_data, runner_data=None): for key in resolved_keys: val = all_config_data[key] - is_multinode = val.get(Fields.MULTINODE.value, False) - image = val[Fields.IMAGE.value] - model = val[Fields.MODEL.value] - model_code = val[Fields.MODEL_PREFIX.value] - precision = val[Fields.PRECISION.value] - framework = val[Fields.FRAMEWORK.value] runner = val[Fields.RUNNER.value] runners_for_entry = _runner_values_for_filter( runner, runner_data, getattr(args, 'runner_node_filter', None)) if not runners_for_entry: continue - disagg = val.get(Fields.DISAGG.value, False) # Build seq-len filter if --seq-lens was provided seq_lens_filter = None @@ -1256,113 +1250,12 @@ def generate_test_config_sweep(args, all_config_data, runner_data=None): # ---- Agentic-coding scenarios ---- agentic_configs = val[Fields.SCENARIOS.value].get(Fields.AGENTIC_CODING.value, []) if (scenario_filter is None or 'agentic-coding' in scenario_filter) else [] - for agentic_config in agentic_configs: - duration = DEFAULT_AGENTIC_DURATION_SECONDS - bmk_space = agentic_config[Fields.SEARCH_SPACE.value] - - for bmk in bmk_space: - if is_multinode: - prefill, decode = multinode_worker_pair(bmk, disagg) - spec_decoding = bmk.get(Fields.SPEC_DECODING.value, "none") - kv_offloading = bmk.get(Fields.KV_OFFLOADING.value, "none") - kv_offload_backend = bmk.get(Fields.KV_OFFLOAD_BACKEND.value) - else: - tp = bmk[Fields.TP.value] - pp = bmk.get(Fields.PP.value, 1) - dcp_size = bmk.get(Fields.DCP_SIZE.value, 1) - pcp_size = bmk.get(Fields.PCP_SIZE.value, 1) - ep = bmk.get(Fields.EP.value) - dp_attn = bmk.get(Fields.DP_ATTN.value) - spec_decoding = bmk.get(Fields.SPEC_DECODING.value, "none") - kv_offloading = bmk[Fields.KV_OFFLOADING.value] - kv_offload_backend = bmk.get(Fields.KV_OFFLOAD_BACKEND.value) - total_cpu_dram_gb = agentic_dram_offload_gb( - agentic_config, bmk, runner, runner_data) - - conc_list = bmk.get(Fields.CONC_LIST.value) - if conc_list: - conc_values = conc_list - else: - conc_start = bmk[Fields.CONC_START.value] - conc_end = bmk[Fields.CONC_END.value] - conc_values = _concurrency_range(conc_start, conc_end, 2) - - if getattr(args, 'conc', None): - conc_values = [c for c in conc_values if c in args.conc] - if not conc_values: - continue - - if is_multinode: - # Preserve historical exp-names for the default (no offload) - # case; only append a suffix when KV offloading is active. - offload_suffix = ( - f"_{agentic_kv_offload_suffix(kv_offloading, kv_offload_backend)}" - if kv_offloading != "none" - else "" - ) - for runner_value in runners_for_entry: - for conc_batch in chunk_multinode_agentic_concurrencies(conc_values): - entry = { - Fields.IMAGE.value: image, - Fields.MODEL.value: model, - Fields.MODEL_PREFIX.value: model_code, - Fields.PRECISION.value: precision, - Fields.FRAMEWORK.value: framework, - Fields.RUNNER.value: runner_value, - Fields.SPEC_DECODING.value: spec_decoding, - Fields.PREFILL.value: prefill, - Fields.DECODE.value: decode, - Fields.CONC.value: conc_batch, - Fields.KV_OFFLOADING.value: kv_offloading, - Fields.TOTAL_CPU_DRAM_GB.value: total_cpu_dram_gb, - Fields.DURATION.value: duration, - Fields.EXP_NAME.value: multinode_agentic_exp_name( - model_code, prefill, decode, conc_batch, offload_suffix - ), - Fields.DISAGG.value: disagg, - Fields.SCENARIO_TYPE.value: "agentic-coding", - } - if kv_offload_backend is not None: - entry[Fields.KV_OFFLOAD_BACKEND.value] = kv_offload_backend - entry.update(component_metadata(bmk, val)) - add_multinode_node_count( - entry, - runner_data, - bmk.get(Fields.NUM_NODES.value), - ) - matrix_values.append(validate_agentic_matrix_entry(entry)) - else: - for conc in conc_values: - for runner_value in runners_for_entry: - entry = { - Fields.IMAGE.value: image, - Fields.MODEL.value: model, - Fields.MODEL_PREFIX.value: model_code, - Fields.PRECISION.value: precision, - Fields.FRAMEWORK.value: framework, - Fields.RUNNER.value: runner_value, - Fields.TP.value: tp, - Fields.PP.value: pp, - Fields.DCP_SIZE.value: dcp_size, - Fields.PCP_SIZE.value: pcp_size, - Fields.EP.value: ep if ep is not None else 1, - Fields.DP_ATTN.value: dp_attn if dp_attn is not None else False, - Fields.SPEC_DECODING.value: spec_decoding, - Fields.CONC.value: conc, - Fields.KV_OFFLOADING.value: kv_offloading, - Fields.TOTAL_CPU_DRAM_GB.value: total_cpu_dram_gb, - Fields.DURATION.value: duration, - Fields.EXP_NAME.value: ( - f"{model_code}_tp{tp}_conc{conc}_" - f"{agentic_kv_offload_suffix(kv_offloading, kv_offload_backend)}" - + (f"_spec-{spec_decoding}" if spec_decoding != "none" else "") - ), - Fields.SCENARIO_TYPE.value: "agentic-coding", - } - if kv_offload_backend is not None: - entry[Fields.KV_OFFLOAD_BACKEND.value] = kv_offload_backend - entry.update(component_metadata(bmk, val)) - matrix_values.append(validate_agentic_matrix_entry(entry)) + for scenario in agentic_configs: + for benchmark in scenario[Fields.SEARCH_SPACE.value]: + matrix_values.extend(_agentic_entries( + val, benchmark, scenario, runners_for_entry, runner_data, + conc_filter=getattr(args, 'conc', None), + )) return matrix_values diff --git a/utils/matrix_logic/test_generate_sweep_configs.py b/utils/matrix_logic/test_generate_sweep_configs.py index 819b09c554..08f3447b8a 100644 --- a/utils/matrix_logic/test_generate_sweep_configs.py +++ b/utils/matrix_logic/test_generate_sweep_configs.py @@ -2309,8 +2309,101 @@ def test_runner_node_filter_no_match_skips_config(self, sample_multinode_config, assert result == [] - def test_runner_node_filter_expands_agentic_config_runner(self, sample_runner_config): - """Agentic test-config entries should support concrete runner targeting.""" + +@pytest.fixture(params=["full-sweep", "test-config"]) +def agentic_mode(request): + return request.param + + +@pytest.fixture +def generate_agentic_sweep(agentic_mode, full_sweep_args_single_node): + def generate(config, runner_data, **filters): + args = copy.copy(full_sweep_args_single_node) + vars(args).update( + config_keys=list(config), conc=None, multi_node=True, + scenario_type=["agentic-coding"], + ) + vars(args).update(filters) + generate = generate_full_sweep if agentic_mode == "full-sweep" else generate_test_config_sweep + return generate(args, config, runner_data) + return generate + + +@pytest.fixture(params=["single", "aggregated", "disaggregated"]) +def agentic_config(request, sample_single_node_config): + config = copy.deepcopy(sample_single_node_config) + entry = next(iter(config.values())) + entry.update(runner="cluster:b300-nv", multinode=request.param != "single") + if request.param == "single": + benchmark = {"tp": 4, "kv-offloading": "none"} + elif request.param == "aggregated": + benchmark = {"num-nodes": 2, "worker": {"num-worker": 2, "tp": 8, "ep": 1, "dp-attn": False}} + else: + entry.update(disagg=True, **{"kv-p2p-transfer": "nixl"}) + benchmark = { + "prefill": {"num-worker": 1, "tp": 8, "ep": 1, "dp-attn": False}, + "decode": {"num-worker": 1, "tp": 8, "ep": 1, "dp-attn": False}, + } + entry["scenarios"] = {"agentic-coding": [{"search-space": [benchmark]}]} + return config, benchmark + + +class TestAgenticGeneration: + def test_point_order_and_input_preservation( + self, agentic_config, sample_runner_config, generate_agentic_sweep, + ): + config, benchmark = agentic_config + benchmark["conc-list"] = [32, 8, 32] + original = copy.deepcopy(config) + entries = generate_agentic_sweep(config, sample_runner_config, runner_node_filter="b300-nv_") + if next(iter(config.values()))["multinode"]: + expected = [ + ("b300-nv_0", [32]), ("b300-nv_0", [8]), ("b300-nv_0", [32]), + ("b300-nv_1", [32]), ("b300-nv_1", [8]), ("b300-nv_1", [32]), + ] + else: + expected = [ + ("b300-nv_0", 32), ("b300-nv_1", 32), + ("b300-nv_0", 8), ("b300-nv_1", 8), + ("b300-nv_0", 32), ("b300-nv_1", 32), + ] + assert [(e["runner"], e["conc"]) for e in entries] == expected + assert config == original + + @pytest.mark.parametrize(("full_filters", "exact_filters", "expected"), [ + ({}, {}, [3, 6, 10]), + ({"min_conc": 5, "max_conc": 9}, {"conc": [6, 9]}, [6]), + ({"max_conc": 2}, {"conc": [2]}, []), + ({"min_conc": 11}, {"conc": [11]}, []), + ]) + def test_range_boundaries( + self, agentic_config, sample_runner_config, generate_agentic_sweep, + agentic_mode, full_filters, exact_filters, expected, + ): + config, benchmark = agentic_config + benchmark.update({"conc-start": 3, "conc-end": 10}) + filters = full_filters if agentic_mode == "full-sweep" else exact_filters + entries = generate_agentic_sweep(config, sample_runner_config, **filters) + points = [entry["conc"] for entry in entries] + assert points == ([[c] for c in expected] if next(iter(config.values()))["multinode"] else expected) + + def test_step_size_and_parallelism_caps_keep_command_semantics( + self, agentic_config, sample_runner_config, generate_agentic_sweep, agentic_mode, + ): + config, benchmark = agentic_config + benchmark.update({"conc-start": 3, "conc-end": 10}) + # Agentic rows ignore the fixed-sequence TP/EP caps. Only full-sweep + # takes a custom range step; test-config always doubles concurrency. + entries = generate_agentic_sweep( + config, sample_runner_config, step_size=3, max_tp=1, max_ep=0, + ) + expected = [3, 9, 10] if agentic_mode == "full-sweep" else [3, 6, 10] + assert [e["conc"] for e in entries] == ( + [[c] for c in expected] if next(iter(config.values()))["multinode"] else expected + ) + + def test_runner_node_filter_expands_agentic_config_runner(self, sample_runner_config, generate_agentic_sweep): + """Agentic entries support concrete runner targeting through both commands.""" config = { "qwen-agentic-hicache": { "image": "sglang-rocm", @@ -2338,15 +2431,8 @@ def test_runner_node_filter_expands_agentic_config_runner(self, sample_runner_co }, } } - args = argparse.Namespace( - config_keys=["qwen-agentic-hicache"], - seq_lens=None, - conc=None, - scenario_type=["agentic-coding"], - runner_node_filter="b300-nv_1", - ) - result = generate_test_config_sweep(args, config, sample_runner_config) + result = generate_agentic_sweep(config, sample_runner_config, runner_node_filter="b300-nv_1") assert len(result) == 1 assert result[0]["runner"] == "b300-nv_1" @@ -2354,7 +2440,7 @@ def test_runner_node_filter_expands_agentic_config_runner(self, sample_runner_co assert result[0]["total-cpu-dram-gb"] == 2399 assert result[0]["duration"] == 3600 - def test_agentic_node_dram_uses_explicit_gpu_count(self, sample_runner_config): + def test_agentic_node_dram_uses_explicit_gpu_count(self, sample_runner_config, generate_agentic_sweep): config = { "dsv4-b300-agentic": { "image": "vllm/vllm-openai:v0.23.0", @@ -2402,15 +2488,8 @@ def test_agentic_node_dram_uses_explicit_gpu_count(self, sample_runner_config): }, }, } - args = argparse.Namespace( - config_keys=["dsv4-b300-agentic"], - seq_lens=None, - conc=None, - scenario_type=["agentic-coding"], - runner_node_filter=None, - ) - result = generate_test_config_sweep(args, config, sample_runner_config) + result = generate_agentic_sweep(config, sample_runner_config) budgets = { (entry["pp"], entry["dcp-size"], entry["pcp-size"]): entry["total-cpu-dram-gb"] @@ -2424,7 +2503,8 @@ def test_agentic_node_dram_uses_explicit_gpu_count(self, sample_runner_config): } assert all(entry["duration"] == 3600 for entry in result) - def test_agentic_node_dram_rejects_tp_above_runner_gpus(self, sample_runner_config): + @pytest.mark.parametrize("filters", [{}, {"min_conc": 999, "conc": [999]}]) + def test_agentic_node_dram_rejects_tp_above_runner_gpus(self, sample_runner_config, generate_agentic_sweep, filters): config = { "dsv4-b300-agentic": { "image": "vllm/vllm-openai:v0.23.0", @@ -2451,19 +2531,12 @@ def test_agentic_node_dram_rejects_tp_above_runner_gpus(self, sample_runner_conf } runner_config = copy.deepcopy(sample_runner_config) runner_config["hardware"]["cluster:b300-nv"]["gpus-per-node"] = 2 - args = argparse.Namespace( - config_keys=["dsv4-b300-agentic"], - seq_lens=None, - conc=None, - scenario_type=["agentic-coding"], - runner_node_filter=None, - ) with pytest.raises(ValueError, match="exceeds gpus-per-node"): - generate_test_config_sweep(args, config, runner_config) + generate_agentic_sweep(config, runner_config, **filters) def test_multinode_agentic_groups_concurrencies_per_search_entry( - self, sample_runner_config + self, sample_runner_config, generate_agentic_sweep ): """One server allocation should run exactly one concurrency (one task per conc).""" config = { @@ -2492,15 +2565,8 @@ def test_multinode_agentic_groups_concurrencies_per_search_entry( }, } } - args = argparse.Namespace( - config_keys=["dsv4-agentic-2p1d"], - seq_lens=None, - conc=[16, 32, 64, 128, 256], - scenario_type=["agentic-coding"], - runner_node_filter=None, - ) - result = generate_test_config_sweep(args, config, sample_runner_config) + result = generate_agentic_sweep(config, sample_runner_config) assert len(result) == 5 assert [entry["conc"] for entry in result] == [[16], [32], [64], [128], [256]] @@ -2519,7 +2585,7 @@ def test_multinode_agentic_groups_concurrencies_per_search_entry( assert result[0]["decode"]["pcp-size"] == 1 assert {entry["node-count"] for entry in result} == {9} - def test_multinode_agentic_preserves_kv_offload_fields(self, sample_runner_config): + def test_multinode_agentic_preserves_kv_offload_fields(self, sample_runner_config, generate_agentic_sweep): config = { "dsv4-agentic-hicache": { "image": "sglang-rocm", @@ -2545,15 +2611,8 @@ def test_multinode_agentic_preserves_kv_offload_fields(self, sample_runner_confi }, }, } - args = argparse.Namespace( - config_keys=["dsv4-agentic-hicache"], - seq_lens=None, - conc=None, - scenario_type=["agentic-coding"], - runner_node_filter=None, - ) - result = generate_test_config_sweep(args, config, sample_runner_config) + result = generate_agentic_sweep(config, sample_runner_config) assert len(result) == 1 assert result[0]["kv-offloading"] == "dram" @@ -2565,7 +2624,7 @@ def test_multinode_agentic_preserves_kv_offload_fields(self, sample_runner_confi assert result[0]["total-cpu-dram-gb"] == 2399 def test_multinode_agentic_budget_ignores_decode_topology( - self, sample_runner_config + self, sample_runner_config, generate_agentic_sweep ): """Only prefill offloads today, so decode's topology does not shrink it.""" config = { @@ -2594,22 +2653,15 @@ def test_multinode_agentic_budget_ignores_decode_topology( }, }, } - args = argparse.Namespace( - config_keys=["dsv4-agentic-hicache-asym"], - seq_lens=None, - conc=None, - scenario_type=["agentic-coding"], - runner_node_filter=None, - ) - result = generate_test_config_sweep(args, config, sample_runner_config) + result = generate_agentic_sweep(config, sample_runner_config) assert len(result) == 1 # prefill 8/8 -> full budget, regardless of decode tp=4. assert result[0]["total-cpu-dram-gb"] == 2399 def test_multinode_agentic_rejects_node_misaligned_prefill( - self, sample_runner_config + self, sample_runner_config, generate_agentic_sweep ): """A prefill worker whose GPU footprint does not tile the node is rejected.""" config = { @@ -2638,16 +2690,9 @@ def test_multinode_agentic_rejects_node_misaligned_prefill( }, }, } - args = argparse.Namespace( - config_keys=["dsv4-agentic-hicache-misaligned"], - seq_lens=None, - conc=None, - scenario_type=["agentic-coding"], - runner_node_filter=None, - ) with pytest.raises(ValueError, match="does not divide"): - generate_test_config_sweep(args, config, sample_runner_config) + generate_agentic_sweep(config, sample_runner_config) # =============================================================================