diff --git a/.github/workflows/ci-benchmarks.yml b/.github/workflows/ci-benchmarks.yml index 507b69e..1f62a1d 100644 --- a/.github/workflows/ci-benchmarks.yml +++ b/.github/workflows/ci-benchmarks.yml @@ -15,6 +15,22 @@ on: description: "Experiments to run (comma-separated Q-IDs or 'all')" required: false default: "all" + backend: + description: "Runtime backend" + required: false + default: "sage" + type: choice + options: + - sage + - ray + seed: + description: "Random seed" + required: false + default: "42" + parallelism: + description: "Single-node operator parallelism" + required: false + default: "2" schedule: - cron: "0 2 * * *" @@ -23,7 +39,7 @@ jobs: runs-on: ubuntu-latest timeout-minutes: 90 env: - HF_ENDPOINT: https://hf-mirror.com + HF_ENDPOINT: https://huggingface.co SAGELLM_PORT: 8888 SAGELLM_EMBED_PORT: 8890 steps: @@ -42,8 +58,11 @@ jobs: - name: Install Python dependencies run: | python -m pip install --upgrade pip - python -m pip install "isagellm>=0.5.1.9" - python -m pip install -e . + if [ "${{ github.event.inputs.backend || 'sage' }}" = "ray" ]; then + python -m pip install -e ".[ray-baseline]" + else + python -m pip install -e . + fi - name: Start sagellm full stack (CPU + embedding, port ${{ env.SAGELLM_PORT }}) run: | @@ -63,6 +82,11 @@ jobs: if curl -sf "http://localhost:${SAGELLM_PORT}/health" > /dev/null 2>&1; then echo "Gateway healthy after ~$((i * 5))s"; break fi + if ! kill -0 "$SAGELLM_PID" 2>/dev/null; then + echo "::error::sagellm exited before the gateway became healthy" + cat /tmp/sagellm.log + exit 1 + fi [ "$i" -eq 72 ] && { cat /tmp/sagellm.log; exit 1; } sleep 5 done @@ -86,7 +110,7 @@ jobs: | python -c "import sys,json; d=json.load(sys.stdin); print('LLM ok')" curl -sf -X POST "http://localhost:${SAGELLM_PORT}/v1/embeddings" \ -H "Content-Type: application/json" \ - -d '{"model":"BAAI/bge-small-zh-v1.5","input":["hello"]}' \ + -d '{"model":"sentence-transformers/all-MiniLM-L6-v2","input":["hello"]}' \ | python -c "import sys,json; d=json.load(sys.stdin); print('Embed ok, dim='+str(len(d['data'][0]['embedding'])))" - name: Run Q1-Q8 benchmarks @@ -94,12 +118,23 @@ jobs: QUICK_FLAG="" [ "${{ github.event.inputs.quick || 'true' }}" = "true" ] && QUICK_FLAG="--quick" EXPERIMENTS="${{ github.event.inputs.experiments || 'all' }}" + COMMON_ARGS=( + --backend "${{ github.event.inputs.backend || 'sage' }}" + --seed "${{ github.event.inputs.seed || '42' }}" + --nodes 1 + --parallelism "${{ github.event.inputs.parallelism || '2' }}" + --continue-on-error + --output-dir results/ci_run + ) if [ "$EXPERIMENTS" = "all" ]; then - python __main__.py --all $QUICK_FLAG --output-dir results/ci_run + python __main__.py --all $QUICK_FLAG "${COMMON_ARGS[@]}" else IFS=',' read -ra EXP_LIST <<< "$EXPERIMENTS" for EXP in "${EXP_LIST[@]}"; do - python __main__.py --experiment "$(echo "$EXP" | tr -d ' ')" $QUICK_FLAG --output-dir results/ci_run + python __main__.py \ + --experiment "$(echo "$EXP" | tr -d ' ')" \ + $QUICK_FLAG \ + "${COMMON_ARGS[@]}" done fi @@ -111,7 +146,10 @@ jobs: uses: actions/upload-artifact@v4 with: name: benchmark-results-${{ github.run_id }} - path: results/ci_run + path: | + results/ci_run + hf_data/benchmark_results.json + hf_data/benchmark_summary.json if-no-files-found: warn - name: Upload sagellm logs (on failure) @@ -120,3 +158,11 @@ jobs: with: name: sagellm-logs-${{ github.run_id }} path: /tmp/sagellm.log + + - name: Stop sagellm + if: always() + run: | + if [ -n "${SAGELLM_PID:-}" ]; then + kill "$SAGELLM_PID" 2>/dev/null || true + wait "$SAGELLM_PID" 2>/dev/null || true + fi diff --git a/.github/workflows/upload-to-hf.yml b/.github/workflows/upload-to-hf.yml index 52facd4..805b127 100644 --- a/.github/workflows/upload-to-hf.yml +++ b/.github/workflows/upload-to-hf.yml @@ -1,20 +1,21 @@ name: Upload to Hugging Face -# 当 hf_data/ 下有 JSON 文件变动时自动触发（仅 main-dev 分支） +# 当 main 分支的 hf_data/ 下有 JSON 文件变动时自动触发 # 用户工作流： # 1. python scripts/aggregate_for_hf.py （本地聚合） -# 2. git add hf_data/ && git commit && git push origin main-dev （触发此 workflow） +# 2. git add hf_data/ && git commit && git push origin main （触发此 workflow） # 3. Actions 自动：并发安全合并 → 上传 HF → 清理 hf_data/ -# -# 注意：main 分支不触发此 workflow，benchmark results 不应直接推送到 main。 on: push: branches: - - main-dev + - main paths: - "hf_data/**/*.json" +permissions: + contents: write + jobs: upload-to-hf: runs-on: ubuntu-latest @@ -44,7 +45,7 @@ jobs: - name: Upload to Hugging Face env: HF_TOKEN: ${{ secrets.HF_TOKEN }} - HF_ENDPOINT: https://hf-mirror.com + HF_ENDPOINT: https://huggingface.co run: | python scripts/upload_to_hf.py diff --git a/README.md b/README.md index 365a3d0..59dfda1 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,10 @@ The `quickstart.sh` script will automatically: ## ⚡ One-Click Full Pipeline +For the Q1–Q8 runner contract, output schema, CI inputs, aggregation flow, and +failure troubleshooting, see the +[Q1–Q8 single-node benchmark guide](docs/q-matrix-benchmark.md). + Run the end-to-end benchmark pipeline in one command: ```bash diff --git a/__main__.py b/__main__.py index 223ef9a..27e52b0 100644 --- a/__main__.py +++ b/__main__.py @@ -15,9 +15,13 @@ from __future__ import annotations import argparse +import json import sys import uuid +from contextlib import redirect_stderr, redirect_stdout +from datetime import datetime, timezone from pathlib import Path +from typing import TextIO WORKLOAD_CATALOG = { "Q1": { @@ -65,6 +69,22 @@ VALID_EXPERIMENTS = tuple(WORKLOAD_CATALOG.keys()) +class _Tee: + """Write workload output to both the terminal and its per-run log.""" + + def __init__(self, *streams: TextIO): + self.streams = streams + + def write(self, data: str) -> int: + for stream in self.streams: + stream.write(data) + return len(data) + + def flush(self) -> None: + for stream in self.streams: + stream.flush() + + def _workload_label(exp_q: str) -> str: meta = WORKLOAD_CATALOG[exp_q] return f"{exp_q} ({meta['name']}, {meta['entry']})" @@ -86,6 +106,23 @@ def _resolve_default_config_path(base_dir: Path, canonical_q: str) -> Path | Non return None +def _write_matrix_status(output_dir: Path, attempts: list[dict[str, object]]) -> None: + """Persist a machine-readable status summary, including partial runs.""" + output_dir.mkdir(parents=True, exist_ok=True) + payload = { + "schema_version": 1, + "updated_at": datetime.now(timezone.utc).isoformat(), + "total": len(attempts), + "passed": sum(item["status"] == "passed" for item in attempts), + "failed": sum(item["status"] == "failed" for item in attempts), + "attempts": attempts, + } + target = output_dir / "matrix_status.json" + temporary = target.with_suffix(".json.tmp") + temporary.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") + temporary.replace(target) + + def main() -> int: # Import here so --help is fast even without heavy deps installed. from experiments.common.cli_args import ( @@ -119,28 +156,37 @@ def main() -> int: # ── Workload selection (suite-level) ──────────────────────────────────── selection_grp = parser.add_argument_group("workload selection") - selection_grp.add_argument( + selection = selection_grp.add_mutually_exclusive_group(required=True) + selection.add_argument( "--experiment", "-e", type=str, help="Workload to run (Q1..Q8).", ) - selection_grp.add_argument( - "--all", "-a", action="store_true", help="Run all workloads in catalog." - ) + selection.add_argument("--all", "-a", action="store_true", help="Run all workloads in catalog.") selection_grp.add_argument( "--config", "-c", type=str, help="Path to a custom config YAML file." ) # ── Standardised benchmark flags (shared across all workloads) ────────── add_common_benchmark_args(parser, include_quick=True, include_dry_run=True) + failure_grp = parser.add_mutually_exclusive_group() + failure_grp.add_argument( + "--continue-on-error", + dest="fail_fast", + action="store_false", + help="Run the remaining workloads after a failure (default).", + ) + failure_grp.add_argument( + "--fail-fast", + dest="fail_fast", + action="store_true", + help="Stop the matrix immediately after the first failed repetition.", + ) + parser.set_defaults(fail_fast=False) args = parser.parse_args() - if not args.experiment and not args.all: - parser.print_help() - return 1 - validate_benchmark_args(args) # Import here to avoid slow startup for --help @@ -179,6 +225,9 @@ def main() -> int: output_dir.mkdir(parents=True, exist_ok=True) results: dict[str, object] = {} + attempts: list[dict[str, object]] = [] + failure_count = 0 + stop_matrix = False for exp_q in experiments_to_run: run_cfg = build_run_config(args, workload=exp_q) @@ -236,22 +285,63 @@ def main() -> int: experiment.nodes = int(args.nodes) experiment.parallelism = int(args.parallelism) experiment.run_id = f"{exp_q.lower()}-{args.backend}-{rep}-{uuid.uuid4().hex[:8]}" + setup_started = False try: - experiment.setup() - result = experiment.run() - experiment.teardown() + rep_output.mkdir(parents=True, exist_ok=True) + with (rep_output / "run.log").open("a", encoding="utf-8") as log_handle: + with redirect_stdout(_Tee(sys.stdout, log_handle)): + with redirect_stderr(_Tee(sys.stderr, log_handle)): + setup_started = True + experiment.setup() + result = experiment.run() + setup_started = False + experiment.teardown() results.setdefault(exp_q, []).append(result) # type: ignore[union-attr] + attempts.append( + { + "workload": exp_q.lower(), + "repeat": rep, + "run_id": experiment.run_id, + "status": "passed", + "output_dir": str(rep_output), + } + ) + _write_matrix_status(output_dir, attempts) print( f" Workload {_workload_label(exp_q)}{rep_label} completed. " f"Results saved to {rep_output}" ) except Exception as exc: # noqa: BLE001 + if setup_started: + try: + experiment.teardown() + except Exception as teardown_exc: # noqa: BLE001 + exc = RuntimeError(f"{exc}; teardown also failed: {teardown_exc}") print(f"Error running workload {_workload_label(exp_q)}{rep_label}: {exc}") + with (rep_output / "run.log").open("a", encoding="utf-8") as log_handle: + log_handle.write(f"ERROR: {exc}\n") if args.verbose: import traceback traceback.print_exc() results.setdefault(exp_q, []).append({"error": str(exc)}) # type: ignore[union-attr] + failure_count += 1 + attempts.append( + { + "workload": exp_q.lower(), + "repeat": rep, + "run_id": experiment.run_id, + "status": "failed", + "error": str(exc), + "output_dir": str(rep_output), + } + ) + _write_matrix_status(output_dir, attempts) + if args.fail_fast: + stop_matrix = True + break + if stop_matrix: + break if not args.dry_run: print(f"\n{'=' * 60}") @@ -270,7 +360,7 @@ def main() -> int: print(f" {_workload_label(exp_q)}: COMPLETED") print(f"\nResults saved to: {output_dir.absolute()}") - return 0 + return 1 if failure_count else 0 if __name__ == "__main__": # pragma: no cover diff --git a/docs/q-matrix-benchmark.md b/docs/q-matrix-benchmark.md new file mode 100644 index 0000000..0809eeb --- /dev/null +++ b/docs/q-matrix-benchmark.md @@ -0,0 +1,111 @@ +# Q1–Q8 single-node benchmark matrix + +The Q1–Q8 catalog is a TPC-inspired workload matrix for SAGE system behavior. +It is not an implementation of TPC-H or TPC-C. Each workload writes the same +machine-readable metrics schema so that results can be compared and published +without workload-specific conversion. + +## Run the matrix + +Install the benchmark package and start the LLM and embedding endpoints required +by the selected workloads. Then run: + +```bash +python __main__.py \ + --all \ + --backend sage \ + --nodes 1 \ + --parallelism 2 \ + --seed 42 \ + --repeat 1 \ + --output-dir results/q-matrix +``` + +Use `--quick` for a reduced smoke run. The default failure policy finishes the +remaining workloads and exits non-zero if any repetition failed. Use +`--fail-fast` to stop at the first failure. `matrix_status.json` records every +attempt, including partial runs. + +Run one workload with the same contract: + +```bash +python __main__.py \ + --experiment q3 \ + --backend sage \ + --nodes 1 \ + --parallelism 2 \ + --seed 42 \ + --quick \ + --output-dir results/q3-smoke +``` + +## Output layout + +```text +results/q-matrix/ +├── matrix_status.json +├── q1/ +│ ├── config.json +│ ├── results.json +│ ├── unified_results.jsonl +│ └── unified_results.csv +└── ... q2 through q8 +``` + +Q-matrix records use lower-case workload identifiers (`q1` through `q8`). +Every unified record includes the backend, run ID, seed, node count, +parallelism, configuration hash, metrics, experiment name, component versions, +model identifiers, and a system profile. + +## Aggregate and publish + +Create the publication bundle locally: + +```bash +python scripts/aggregate_for_hf.py +``` + +This scans `results/**/unified_results.jsonl`, rejects malformed local records, +deduplicates records by benchmark configuration, and writes: + +```text +hf_data/ +├── benchmark_results.json +└── benchmark_summary.json +``` + +The summary contains record counts by workload, backend, and seed. Publication +to `intellistream/sage-benchmark-results` is performed by the protected +`upload-to-hf.yml` workflow after an `hf_data/**/*.json` change reaches +`main`. Do not commit tokens or place an HF token in a benchmark result. + +## GitHub Actions + +Open **Actions → Benchmark CI → Run workflow**. The workflow accepts: + +- an experiment list (`all` or comma-separated Q IDs); +- backend; +- seed; +- single-node parallelism; +- quick or full scale. + +It archives the partial status and result files even when a workload fails. +A green workflow is required before treating the matrix or HF publication path +as operational. + +## Troubleshooting + +- **Gateway never becomes healthy:** inspect the uploaded `sagellm` log. Confirm + that the model can be downloaded from the configured Hugging Face endpoint + and that the selected sagellm version supports the workflow's serve flags. +- **Embedding smoke test fails:** confirm the registered model and the model in + the `/v1/embeddings` request are both + `sentence-transformers/all-MiniLM-L6-v2`. +- **Matrix exits non-zero:** inspect `matrix_status.json` and the corresponding + `q*/` directory. A partial artifact is diagnostic evidence, not a successful + benchmark matrix. +- **Aggregation refuses a file:** repair the named malformed JSONL record. The + publication path intentionally fails closed instead of silently dropping bad + input. +- **Ray backend is unavailable:** install the documented Ray baseline extra + before selecting `--backend ray`. diff --git a/experiments/base_experiment.py b/experiments/base_experiment.py index 1843dfd..de5527f 100644 --- a/experiments/base_experiment.py +++ b/experiments/base_experiment.py @@ -29,6 +29,7 @@ UnifiedMetricsRecord, compute_backend_hash, compute_config_hash, + normalize_workload_name, ) from experiments.common.result_writer import ( append_jsonl_record, @@ -297,7 +298,7 @@ def _save_unified_outputs(self, result: ExperimentResult) -> None: backend = getattr(self, "backend", "sage") nodes = int(getattr(self, "nodes", 1)) parallelism = int(getattr(self, "parallelism", 2)) - workload = self.config.experiment_section + workload = normalize_workload_name(self.config.experiment_section) run_id = getattr(self, "run_id", "") or f"{workload}-{uuid.uuid4().hex[:12]}" config_payload = { @@ -339,6 +340,10 @@ def _save_unified_outputs(self, result: ExperimentResult) -> None: backend_hash=compute_backend_hash(backend), metadata={ "experiment_name": result.experiment_name, + "experiment_section": self.config.experiment_section, + "seed": self.config.workload.seed, + "nodes": nodes, + "parallelism": parallelism, "sage_version": resolved_sage_version, "sagellm_version": resolved_sagellm_version, "benchmark_version": BENCHMARK_VERSION, diff --git a/experiments/common/__init__.py b/experiments/common/__init__.py index 1cc89c1..38b8ec4 100644 --- a/experiments/common/__init__.py +++ b/experiments/common/__init__.py @@ -35,6 +35,7 @@ QueryComplexityLevel, TaskState, ) + try: from .operators import ( # Adaptive-RAG operators diff --git a/experiments/common/metrics_schema.py b/experiments/common/metrics_schema.py index 87f9e7a..e365e1b 100644 --- a/experiments/common/metrics_schema.py +++ b/experiments/common/metrics_schema.py @@ -10,7 +10,7 @@ import hashlib import json from dataclasses import dataclass, field -from datetime import UTC, datetime +from datetime import datetime, timezone from typing import Any REQUIRED_FIELDS: tuple[str, ...] = ( @@ -29,10 +29,24 @@ "timestamp", ) +Q_MATRIX_WORKLOADS: frozenset[str] = frozenset(f"q{index}" for index in range(1, 9)) + + +def normalize_workload_name(value: Any) -> Any: + """Return the canonical lower-case identifier for Q1--Q8 workloads. + + Historical records used both ``Q1`` and ``q1``. Normalising at the schema + boundary keeps newly written records stable while leaving unrelated + workload names (for example ``scheduler_comparison``) unchanged. + """ + if isinstance(value, str) and value.lower() in Q_MATRIX_WORKLOADS: + return value.lower() + return value + def utc_timestamp() -> str: """Return an ISO-8601 UTC timestamp string.""" - return datetime.now(UTC).isoformat() + return datetime.now(timezone.utc).isoformat() def compute_config_hash(config: dict[str, Any]) -> str: @@ -74,7 +88,7 @@ def to_dict(self) -> dict[str, Any]: """Convert record to a JSON-serialisable dict with fixed key set.""" payload: dict[str, Any] = { "backend": self.backend, - "workload": self.workload, + "workload": normalize_workload_name(self.workload), "run_id": self.run_id, "seed": int(self.seed), "nodes": int(self.nodes), @@ -99,6 +113,7 @@ def normalize_metrics_record(record: dict[str, Any]) -> dict[str, Any]: Missing required fields are set to ``None``. """ normalized = {field: record.get(field, None) for field in REQUIRED_FIELDS} + normalized["workload"] = normalize_workload_name(normalized["workload"]) normalized["config_hash"] = record.get("config_hash") normalized["backend_hash"] = record.get("backend_hash") normalized["metadata"] = record.get("metadata", {}) diff --git a/experiments/common/operators.py b/experiments/common/operators.py index 92e247f..313e33c 100644 --- a/experiments/common/operators.py +++ b/experiments/common/operators.py @@ -26,11 +26,11 @@ from .models import TaskState try: - from .models import TaskState from .inference import create_unified_inference_client, response_to_text + from .models import TaskState except ImportError: - from models import TaskState from inference import create_unified_inference_client, response_to_text + from models import TaskState # 示例查询池 - 包含 ZERO/SINGLE/MULTI 三种复杂度 diff --git a/experiments/common/pipeline.py b/experiments/common/pipeline.py index 62d5cdb..24e7c55 100644 --- a/experiments/common/pipeline.py +++ b/experiments/common/pipeline.py @@ -19,7 +19,13 @@ import time from typing import TYPE_CHECKING, Any -from sage.runtime import BaseService, FIFOScheduler, FluttyEnvironment, LoadAwareScheduler, LocalEnvironment +from sage.runtime import ( + BaseService, + FIFOScheduler, + FluttyEnvironment, + LoadAwareScheduler, + LocalEnvironment, +) if TYPE_CHECKING: from sage.runtime import FluttyEnvironment, LocalEnvironment @@ -61,7 +67,7 @@ def register_embedding_service( - env: LocalEnvironment | FlownetEnvironment, + env: LocalEnvironment | FluttyEnvironment, base_url: str, model: str, ) -> bool: @@ -133,7 +139,7 @@ def close(self): def register_vector_db_service( - env: LocalEnvironment | FlownetEnvironment, + env: LocalEnvironment | FluttyEnvironment, embedding_base_url: str, embedding_model: str, knowledge_base: list[dict[str, Any]] | None = None, @@ -246,7 +252,9 @@ def _ensure_initialized(self): ] print(f"[VectorDB] Loaded {len(vectors)} documents") - def search(self, query_vec, k: int = 5, top_k: int | None = None) -> list[tuple[float, dict]]: + def search( + self, query_vec, k: int = 5, top_k: int | None = None + ) -> list[tuple[float, dict]]: """Search for similar documents""" self._ensure_initialized() import numpy as np @@ -266,10 +274,7 @@ def search(self, query_vec, k: int = 5, top_k: int | None = None) -> list[tuple[ vector_norms = np.linalg.norm(vectors, axis=1) + 1e-8 scores = (vectors @ query) / (vector_norms * query_norm) ranked_indices = np.argsort(scores)[::-1][:limit] - return [ - (float(scores[idx]), metadata_list[idx]) - for idx in ranked_indices - ] + return [(float(scores[idx]), metadata_list[idx]) for idx in ranked_indices] def process(self, query_vec, k: int = 5) -> list[tuple[float, dict]]: """Default RPC method - alias for search""" @@ -318,7 +323,7 @@ def add_batch(self, vectors, metadata_list): def register_llm_service( - env: LocalEnvironment | FlownetEnvironment, + env: LocalEnvironment | FluttyEnvironment, base_url: str, model: str, max_tokens: int = 256, @@ -409,7 +414,7 @@ def close(self): def register_fiqa_vdb_service( - env: LocalEnvironment | FlownetEnvironment, + env: LocalEnvironment | FluttyEnvironment, embedding_base_url: str, embedding_model: str, data_dir: str = FIQA_DATA_DIR, @@ -601,7 +606,7 @@ def process(self, query: str, top_k: int = 5) -> list[dict]: def register_all_services( - env: LocalEnvironment | FlownetEnvironment, + env: LocalEnvironment | FluttyEnvironment, config: BenchmarkConfig, knowledge_base: list[dict[str, Any]] | None = None, vdb_node_ip: str | None = None, @@ -722,7 +727,9 @@ def _create_scheduler(self): return LoadAwareScheduler( platform=platform, max_concurrent=self.config.parallelism * 100, - strategy=self.config.scheduler_strategy if scheduler_type == "load_aware" else "balanced", + strategy=self.config.scheduler_strategy + if scheduler_type == "load_aware" + else "balanced", ) def _create_environment(self, name: str) -> LocalEnvironment | FluttyEnvironment: diff --git a/experiments/distributed_workloads/workload4/clustering.py b/experiments/distributed_workloads/workload4/clustering.py index 0b1dc1c..615add5 100644 --- a/experiments/distributed_workloads/workload4/clustering.py +++ b/experiments/distributed_workloads/workload4/clustering.py @@ -9,10 +9,10 @@ from typing import Any import numpy as np -from sklearn.cluster import DBSCAN -from sklearn.metrics.pairwise import cosine_similarity from sage.foundation import MapFunction from sage.runtime import StopSignal +from sklearn.cluster import DBSCAN +from sklearn.metrics.pairwise import cosine_similarity from .models import ClusteringResult, GraphMemoryResult, VDBRetrievalResult diff --git a/experiments/distributed_workloads/workload4/graph_memory.py b/experiments/distributed_workloads/workload4/graph_memory.py index 34bd35d..8cb71ae 100644 --- a/experiments/distributed_workloads/workload4/graph_memory.py +++ b/experiments/distributed_workloads/workload4/graph_memory.py @@ -14,6 +14,7 @@ import networkx as nx import numpy as np from sage.foundation import MapFunction +from sage.runtime import StopSignal try: from .models import GraphMemoryResult, JoinedEvent diff --git a/experiments/distributed_workloads/workload4/pipeline.py b/experiments/distributed_workloads/workload4/pipeline.py index 28416cf..68db6d2 100644 --- a/experiments/distributed_workloads/workload4/pipeline.py +++ b/experiments/distributed_workloads/workload4/pipeline.py @@ -80,7 +80,7 @@ def register_embedding_service( - env: LocalEnvironment | FlownetEnvironment, + env: LocalEnvironment | FluttyEnvironment, config: Workload4Config, ) -> bool: """ @@ -107,7 +107,7 @@ def register_embedding_service( def register_vdb_services( - env: LocalEnvironment | FlownetEnvironment, + env: LocalEnvironment | FluttyEnvironment, config: Workload4Config, ) -> dict[str, bool]: """ @@ -149,7 +149,7 @@ def register_vdb_services( def register_graph_memory_service( - env: LocalEnvironment | FlownetEnvironment, + env: LocalEnvironment | FluttyEnvironment, config: Workload4Config, ) -> bool: """ @@ -177,7 +177,7 @@ def register_graph_memory_service( def register_llm_service( - env: LocalEnvironment | FlownetEnvironment, + env: LocalEnvironment | FluttyEnvironment, config: Workload4Config, ) -> bool: """ @@ -206,7 +206,7 @@ def register_llm_service( def register_all_services( - env: LocalEnvironment | FlownetEnvironment, + env: LocalEnvironment | FluttyEnvironment, config: Workload4Config, ) -> dict[str, bool]: """ diff --git a/experiments/pipelines/adaptive_rag/branch_pipeline.py b/experiments/pipelines/adaptive_rag/branch_pipeline.py index fbbb67a..4e09caf 100644 --- a/experiments/pipelines/adaptive_rag/branch_pipeline.py +++ b/experiments/pipelines/adaptive_rag/branch_pipeline.py @@ -36,7 +36,8 @@ SinkFunction, SourceFunction, ) -from sage.runtime import FluttyEnvironment as FlownetEnvironment, LocalEnvironment +from sage.runtime import FluttyEnvironment as FlownetEnvironment +from sage.runtime import LocalEnvironment # 支持直接运行和模块运行两种方式 try: diff --git a/experiments/pipelines/adaptive_rag/functions.py b/experiments/pipelines/adaptive_rag/functions.py index c7aaa0e..24559b3 100644 --- a/experiments/pipelines/adaptive_rag/functions.py +++ b/experiments/pipelines/adaptive_rag/functions.py @@ -14,9 +14,10 @@ from dataclasses import dataclass, field from typing import Any -from experiments.common.inference import create_unified_inference_client, response_to_text from sage.foundation import MapFunction +from experiments.common.inference import create_unified_inference_client, response_to_text + # 支持直接运行和模块运行两种方式 try: from .classifier import ( diff --git a/experiments/pipelines/adaptive_rag/modular_pipeline.py b/experiments/pipelines/adaptive_rag/modular_pipeline.py index 483d32d..9b46ad5 100644 --- a/experiments/pipelines/adaptive_rag/modular_pipeline.py +++ b/experiments/pipelines/adaptive_rag/modular_pipeline.py @@ -48,7 +48,8 @@ SinkFunction, SourceFunction, ) -from sage.runtime import FluttyEnvironment as FlownetEnvironment, LocalEnvironment +from sage.runtime import FluttyEnvironment as FlownetEnvironment +from sage.runtime import LocalEnvironment # 支持直接运行和模块运行 try: diff --git a/experiments/pipelines/adaptive_rag/sage_dataflow_pipeline.py b/experiments/pipelines/adaptive_rag/sage_dataflow_pipeline.py index 5b2f516..d7435e1 100644 --- a/experiments/pipelines/adaptive_rag/sage_dataflow_pipeline.py +++ b/experiments/pipelines/adaptive_rag/sage_dataflow_pipeline.py @@ -42,7 +42,6 @@ from pathlib import Path from typing import Any -from experiments.common.inference import create_unified_inference_client # ============================================================================ # SAGE 核心导入 # ============================================================================ @@ -55,6 +54,8 @@ ) from sage.runtime import LocalEnvironment +from experiments.common.inference import create_unified_inference_client + # 本地分类器导入（支持脚本直接运行和模块运行） if __package__ in (None, ""): current_dir = Path(__file__).resolve().parent diff --git a/experiments/pipelines/pipeline_c_vector_join.py b/experiments/pipelines/pipeline_c_vector_join.py index 90b93ca..e1d3e0c 100644 --- a/experiments/pipelines/pipeline_c_vector_join.py +++ b/experiments/pipelines/pipeline_c_vector_join.py @@ -38,8 +38,8 @@ import httpx from sage.foundation import ( FilterFunction, - SagePorts, MapFunction, + SagePorts, SinkFunction, SourceFunction, ) diff --git a/experiments/tool_use_agent/operators.py b/experiments/tool_use_agent/operators.py index d7b6d91..c5ed3b3 100644 --- a/experiments/tool_use_agent/operators.py +++ b/experiments/tool_use_agent/operators.py @@ -20,10 +20,11 @@ import time from typing import TYPE_CHECKING -from experiments.common.inference import create_unified_inference_client, response_to_text from sage.foundation import MapFunction, SinkFunction, SourceFunction from sage.runtime import StopSignal +from experiments.common.inference import create_unified_inference_client, response_to_text + if TYPE_CHECKING: from .agent_tools import ToolRegistry diff --git a/pyproject.toml b/pyproject.toml index e933ca6..ddd3a65 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -63,6 +63,9 @@ full = [ # Vector DB backends "pymilvus[model]>=2.4.0", ] +ray-baseline = [ + "ray>=2.9.0", +] dev = [ # dev includes all ML/DB backends — no need for pip install -e .[dev,full] "isage-benchmark[full]", diff --git a/scripts/aggregate_for_hf.py b/scripts/aggregate_for_hf.py index 8b2779a..2d7d59b 100644 --- a/scripts/aggregate_for_hf.py +++ b/scripts/aggregate_for_hf.py @@ -22,11 +22,44 @@ import json import urllib.request +from collections import Counter from pathlib import Path # HF 配置 HF_REPO = "intellistream/sage-benchmark-results" HF_BRANCH = "main" +Q_MATRIX_WORKLOADS = frozenset(f"q{index}" for index in range(1, 9)) +REQUIRED_IDENTITY_FIELDS = ( + "backend", + "workload", + "run_id", + "seed", + "nodes", + "parallelism", + "config_hash", +) + + +def normalize_workload_name(value: object) -> object: + if isinstance(value, str) and value.lower() in Q_MATRIX_WORKLOADS: + return value.lower() + return value + + +def validate_local_record(record: object, *, source: str) -> dict: + """Validate and canonicalise one locally generated record. + + Remote historical records remain readable for backwards compatibility, but + malformed new records must not silently enter the publication pipeline. + """ + if not isinstance(record, dict): + raise ValueError(f"{source}: expected a JSON object") + missing = [field for field in REQUIRED_IDENTITY_FIELDS if record.get(field) is None] + if missing: + raise ValueError(f"{source}: missing required fields: {', '.join(missing)}") + normalized = dict(record) + normalized["workload"] = normalize_workload_name(normalized["workload"]) + return normalized def download_from_hf(filename: str) -> list[dict]: @@ -86,15 +119,19 @@ def load_local_results(results_dir: Path) -> list[dict]: """递归加载 results/ 目录下的所有 unified_results.jsonl 文件。""" all_records: list[dict] = [] - for jsonl_file in results_dir.rglob("unified_results.jsonl"): + errors: list[str] = [] + for jsonl_file in sorted(results_dir.rglob("unified_results.jsonl")): try: skipped = 0 with jsonl_file.open("r", encoding="utf-8") as fh: - for line in fh: + for line_number, line in enumerate(fh, start=1): stripped = line.strip() if not stripped: continue - record = json.loads(stripped) + record = validate_local_record( + json.loads(stripped), + source=f"{jsonl_file}:{line_number}", + ) if not _is_valid_record(record): skipped += 1 continue @@ -106,9 +143,12 @@ def load_local_results(results_dir: Path) -> list[dict]: print(f" ✓ 加载: {label}") except Exception as e: print(f" ✗ 加载失败: {jsonl_file} - {e}") - except Exception as e: - print(f" ✗ 加载失败: {jsonl_file} - {e}") + errors.append(str(e)) + if errors: + raise ValueError( + f"拒绝聚合：发现 {len(errors)} 个格式错误的本地结果文件。请修复上述错误后重试。" + ) return all_records @@ -116,7 +156,7 @@ def get_config_key(entry: dict) -> str: """生成配置唯一标识 key（用于去重）。""" parts = [ str(entry.get("backend", "")), - str(entry.get("workload", "")), + str(normalize_workload_name(entry.get("workload", ""))), str(entry.get("seed", "")), str(entry.get("nodes", "")), str(entry.get("parallelism", "")), @@ -169,6 +209,19 @@ def merge_results(existing: list[dict], new_results: list[dict]) -> list[dict]: return list(merged.values()) +def build_coverage_summary(records: list[dict]) -> dict: + """Build deterministic coverage counts for logs and publication.""" + by_workload = Counter(str(normalize_workload_name(row.get("workload", ""))) for row in records) + by_backend = Counter(str(row.get("backend", "")) for row in records) + by_seed = Counter(str(row.get("seed", "")) for row in records) + return { + "total_records": len(records), + "by_workload": dict(sorted(by_workload.items())), + "by_backend": dict(sorted(by_backend.items())), + "by_seed": dict(sorted(by_seed.items())), + } + + def main() -> None: print("=" * 70) print("📦 SAGE Benchmark - 本地聚合工具") @@ -210,6 +263,17 @@ def main() -> None: json.dump(merged, fh, indent=2, ensure_ascii=False) print(f" ✓ {output_file.name} ({len(merged)} 条)") + coverage = build_coverage_summary(merged) + summary_file = hf_output_dir / "benchmark_summary.json" + summary_file.write_text( + json.dumps(coverage, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + print(" 📊 覆盖统计:") + for dimension in ("by_workload", "by_backend", "by_seed"): + print(f" {dimension}: {coverage[dimension]}") + print(f" ✓ {summary_file.name}") + print("\n" + "=" * 70) print("✅ 聚合完成！") print("=" * 70) diff --git a/scripts/merge_and_upload.py b/scripts/merge_and_upload.py index da41a9d..5419ea2 100644 --- a/scripts/merge_and_upload.py +++ b/scripts/merge_and_upload.py @@ -15,11 +15,19 @@ import json import urllib.request +from collections import Counter from pathlib import Path # HF 配置 HF_REPO = "intellistream/sage-benchmark-results" HF_BRANCH = "main" +Q_MATRIX_WORKLOADS = frozenset(f"q{index}" for index in range(1, 9)) + + +def normalize_workload_name(value: object) -> object: + if isinstance(value, str) and value.lower() in Q_MATRIX_WORKLOADS: + return value.lower() + return value def download_from_hf(filename: str) -> list[dict]: @@ -56,7 +64,7 @@ def get_config_key(entry: dict) -> str: """生成配置唯一标识 key。""" parts = [ str(entry.get("backend", "")), - str(entry.get("workload", "")), + str(normalize_workload_name(entry.get("workload", ""))), str(entry.get("seed", "")), str(entry.get("nodes", "")), str(entry.get("parallelism", "")), @@ -120,6 +128,18 @@ def smart_merge(hf_latest: list[dict], user_data: list[dict]) -> list[dict]: return list(merged.values()) +def build_coverage_summary(records: list[dict]) -> dict: + by_workload = Counter(str(normalize_workload_name(row.get("workload", ""))) for row in records) + by_backend = Counter(str(row.get("backend", "")) for row in records) + by_seed = Counter(str(row.get("seed", "")) for row in records) + return { + "total_records": len(records), + "by_workload": dict(sorted(by_workload.items())), + "by_backend": dict(sorted(by_backend.items())), + "by_seed": dict(sorted(by_seed.items())), + } + + def main() -> None: print("=" * 60) print("🔀 并发安全合并（GitHub Actions）") @@ -158,6 +178,12 @@ def main() -> None: encoding="utf-8", ) print(f" ✓ {user_file} ({len(merged)} 条)") + summary_file = hf_data_dir / "benchmark_summary.json" + summary_file.write_text( + json.dumps(build_coverage_summary(merged), indent=2, ensure_ascii=False), + encoding="utf-8", + ) + print(f" ✓ {summary_file}") print("\n✅ 并发安全合并完成！") print("💡 下一步: 运行 upload_to_hf.py 上传到 Hugging Face") diff --git a/scripts/upload_to_hf.py b/scripts/upload_to_hf.py index bc924ab..1c36055 100644 --- a/scripts/upload_to_hf.py +++ b/scripts/upload_to_hf.py @@ -100,6 +100,7 @@ def main() -> None: # 要上传的文件 files_to_upload = [ HF_DATA_DIR / "benchmark_results.json", + HF_DATA_DIR / "benchmark_summary.json", ] if not HF_DATA_DIR.exists(): diff --git a/tests/test_hf_aggregation.py b/tests/test_hf_aggregation.py new file mode 100644 index 0000000..340ed5b --- /dev/null +++ b/tests/test_hf_aggregation.py @@ -0,0 +1,52 @@ +"""Tests for Q-matrix aggregation and publication guards.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from scripts.aggregate_for_hf import ( + build_coverage_summary, + get_config_key, + load_local_results, +) + + +def _valid_record(workload: str = "Q1") -> dict: + return { + "backend": "sage", + "workload": workload, + "run_id": "run-1", + "seed": 42, + "nodes": 1, + "parallelism": 2, + "config_hash": "abc", + "throughput": 1.0, + "latency_p99": 2.0, + "success_rate": 1.0, + } + + +def test_loader_normalizes_q_workloads_and_coverage(tmp_path: Path) -> None: + result_file = tmp_path / "q1" / "unified_results.jsonl" + result_file.parent.mkdir() + result_file.write_text(json.dumps(_valid_record()) + "\n", encoding="utf-8") + + records = load_local_results(tmp_path) + + assert records[0]["workload"] == "q1" + assert build_coverage_summary(records)["by_workload"] == {"q1": 1} + assert get_config_key(_valid_record("Q1")) == get_config_key(_valid_record("q1")) + + +def test_loader_fails_closed_on_malformed_local_record(tmp_path: Path) -> None: + result_file = tmp_path / "q1" / "unified_results.jsonl" + result_file.parent.mkdir() + malformed = _valid_record() + del malformed["config_hash"] + result_file.write_text(json.dumps(malformed) + "\n", encoding="utf-8") + + with pytest.raises(ValueError, match="拒绝聚合"): + load_local_results(tmp_path) diff --git a/tests/test_metrics_schema.py b/tests/test_metrics_schema.py index d795fd5..ee91c59 100644 --- a/tests/test_metrics_schema.py +++ b/tests/test_metrics_schema.py @@ -42,6 +42,7 @@ def test_unified_record_contains_all_required_fields(): assert record["latency_p50"] is None assert record["latency_p95"] is None assert record["latency_p99"] is None + assert record["workload"] == "q1" def test_normalize_metrics_record_fills_missing_with_none(): @@ -50,6 +51,7 @@ def test_normalize_metrics_record_fills_missing_with_none(): assert key in normalized assert normalized["run_id"] is None assert normalized["throughput"] is None + assert normalized["workload"] == "q2" def test_jsonl_and_csv_writer_roundtrip(tmp_path: Path): diff --git a/tests/test_q_matrix_runner.py b/tests/test_q_matrix_runner.py new file mode 100644 index 0000000..1be0112 --- /dev/null +++ b/tests/test_q_matrix_runner.py @@ -0,0 +1,157 @@ +"""Behavioral tests for the Q1--Q8 matrix orchestrator.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +import types +from pathlib import Path +from types import SimpleNamespace + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +def _load_matrix_module(): + spec = importlib.util.spec_from_file_location( + "sage_benchmark_matrix_main", + REPO_ROOT / "__main__.py", + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class _FakeConfigLoader: + def load(self, _path: str): + return self.get_default_config("Q1") + + def get_default_config(self, experiment: str): + return SimpleNamespace( + experiment_section=experiment, + workload=SimpleNamespace(seed=42), + hardware=SimpleNamespace(cpu_nodes=0), + ) + + def apply_quick_mode(self, config): + return config + + +def _install_fake_workloads(monkeypatch: pytest.MonkeyPatch, calls: list[str]) -> None: + experiments_package = types.ModuleType("experiments") + experiments_package.__path__ = [] + monkeypatch.setitem(sys.modules, "experiments", experiments_package) + + common_package = types.ModuleType("experiments.common") + common_package.__path__ = [] + monkeypatch.setitem(sys.modules, "experiments.common", common_package) + cli_module = types.ModuleType("experiments.common.cli_args") + + def add_common_benchmark_args(parser, **_kwargs): + parser.add_argument("--backend", default="sage") + parser.add_argument("--nodes", type=int, default=1) + parser.add_argument("--parallelism", type=int, default=2) + parser.add_argument("--repeat", type=int, default=1) + parser.add_argument("--seed", type=int, default=42) + parser.add_argument("--output-dir", default="results") + parser.add_argument("--verbose", action="store_true") + parser.add_argument("--quick", action="store_true") + parser.add_argument("--dry-run", action="store_true") + + cli_module.add_common_benchmark_args = add_common_benchmark_args + cli_module.validate_benchmark_args = lambda _args: None + cli_module.build_run_config = lambda args, **extra: { + "backend": args.backend, + "nodes": args.nodes, + "parallelism": args.parallelism, + "repeat": args.repeat, + "seed": args.seed, + **extra, + } + monkeypatch.setitem(sys.modules, "experiments.common.cli_args", cli_module) + + config_package = types.ModuleType("config") + config_package.__path__ = [] + monkeypatch.setitem(sys.modules, "config", config_package) + config_module = types.ModuleType("config.config_loader") + config_module.ConfigLoader = _FakeConfigLoader + monkeypatch.setitem(sys.modules, "config.config_loader", config_module) + + classes = { + "experiments.q1_pipelinechain": ("E2EPipelineExperiment", "Q1"), + "experiments.q2_controlmix": ("ControlPlaneExperiment", "Q2"), + "experiments.q3_noisyneighbor": ("IsolationExperiment", "Q3"), + "experiments.q4_scalefrontier": ("ScalabilityExperiment", "Q4"), + "experiments.q5_heteroresilience": ("HeterogeneityExperiment", "Q5"), + "experiments.q6_bursttown": ("BurstTownExperiment", "Q6"), + "experiments.q7_reconfigdrill": ("ReconfigDrillExperiment", "Q7"), + "experiments.q8_recoverysoak": ("RecoverySoakExperiment", "Q8"), + } + + def make_experiment(workload: str): + class FakeExperiment: + def __init__(self, config, output_dir, verbose=False): + self.config = config + self.output_dir = output_dir + + def setup(self): + calls.append(f"{workload}:setup") + + def run(self): + calls.append(f"{workload}:run") + if workload == "Q2": + raise RuntimeError("intentional Q2 failure") + return {"workload": workload} + + def teardown(self): + calls.append(f"{workload}:teardown") + + return FakeExperiment + + for module_name, (class_name, workload) in classes.items(): + module = types.ModuleType(module_name) + setattr(module, class_name, make_experiment(workload)) + monkeypatch.setitem(sys.modules, module_name, module) + + +@pytest.mark.parametrize( + ("failure_flag", "expected_attempts", "last_workload"), + [ + ("--continue-on-error", 8, "q8"), + ("--fail-fast", 2, "q2"), + ], +) +def test_matrix_failure_policy_sets_nonzero_exit_and_writes_status( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + failure_flag: str, + expected_attempts: int, + last_workload: str, +) -> None: + matrix = _load_matrix_module() + calls: list[str] = [] + _install_fake_workloads(monkeypatch, calls) + output_dir = tmp_path / "matrix" + monkeypatch.setattr( + sys, + "argv", + [ + "__main__.py", + "--all", + failure_flag, + "--output-dir", + str(output_dir), + ], + ) + + assert matrix.main() == 1 + + status = json.loads((output_dir / "matrix_status.json").read_text(encoding="utf-8")) + assert status["failed"] == 1 + assert status["total"] == expected_attempts + assert status["attempts"][-1]["workload"] == last_workload + assert "Q2:teardown" in calls + assert "intentional Q2 failure" in (output_dir / "q2" / "run.log").read_text(encoding="utf-8")