From e6437ab3327808b0d690f4f953d3a14b41d86705 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Wed, 29 Jul 2026 10:35:45 +0800 Subject: [PATCH] Move workflow evaluations to standalone runner --- evals/README.md | 48 +++++ evals/cases/browser_addon/initial.yaml | 7 + evals/cases/direct/initial.yaml | 7 + evals/run.py | 248 +++++++++++++++++++++++++ pytest.ini | 1 - requirements.txt | 1 + test/golden/README.md | 32 ---- test/golden/cases.json | 26 --- test/golden/test_stage_code_golden.py | 112 ----------- 9 files changed, 311 insertions(+), 171 deletions(-) create mode 100644 evals/README.md create mode 100644 evals/cases/browser_addon/initial.yaml create mode 100644 evals/cases/direct/initial.yaml create mode 100644 evals/run.py delete mode 100644 test/golden/README.md delete mode 100644 test/golden/cases.json delete mode 100644 test/golden/test_stage_code_golden.py diff --git a/evals/README.md b/evals/README.md new file mode 100644 index 0000000..50fe18b --- /dev/null +++ b/evals/README.md @@ -0,0 +1,48 @@ +# Workflow evaluations + +The evaluation runner calls a live FastAPI service backed by FastGPT. For now, +each conversation step checks only the SSE `stage_code.nextStageCode` value. + +## Case format + +Add YAML files anywhere under `evals/cases/`: + +```yaml +name: browser_addon_example +clientMode: browser_addon + +steps: + - input: hi + expect: + stageCode: browser_addon.1001 + + - input: 继续 + expect: + stageCode: browser_addon.1002 +``` + +Steps in one file share the same generated `sessionId`. Different cases always +use different session IDs. + +## Run + +Start the API, then execute: + +```bash +python evals/run.py +``` + +Use another API address: + +```bash +python evals/run.py --base-url http://192.168.1.10:8000 +``` + +Run selected cases: + +```bash +python evals/run.py --case browser_addon +``` + +The process exits with code `1` when an expectation fails and code `2` when +case configuration is invalid. diff --git a/evals/cases/browser_addon/initial.yaml b/evals/cases/browser_addon/initial.yaml new file mode 100644 index 0000000..034adfd --- /dev/null +++ b/evals/cases/browser_addon/initial.yaml @@ -0,0 +1,7 @@ +name: browser_addon_initial_stage +clientMode: browser_addon + +steps: + - input: hi + expect: + stageCode: browser_addon.1001 diff --git a/evals/cases/direct/initial.yaml b/evals/cases/direct/initial.yaml new file mode 100644 index 0000000..6758917 --- /dev/null +++ b/evals/cases/direct/initial.yaml @@ -0,0 +1,7 @@ +name: direct_initial_stage +clientMode: direct + +steps: + - input: hi + expect: + stageCode: "1001" diff --git a/evals/run.py b/evals/run.py new file mode 100644 index 0000000..7f59826 --- /dev/null +++ b/evals/run.py @@ -0,0 +1,248 @@ +#!/usr/bin/env python3 +"""Run live FastGPT workflow evaluations against the FastAPI service.""" + +import argparse +import json +import os +import re +import sys +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from uuid import uuid4 + +import httpx +import yaml + + +EVALS_DIR = Path(__file__).resolve().parent +DEFAULT_CASES_DIR = EVALS_DIR / "cases" + + +@dataclass +class ChatResult: + stage_codes: list[str] + errors: list[dict] + + +def timestamp() -> str: + return datetime.now(timezone.utc).isoformat() + + +def load_cases(cases_dir: Path, name_filters: list[str]) -> list[dict]: + cases = [] + for path in sorted(cases_dir.rglob("*.yaml")): + case = yaml.safe_load(path.read_text(encoding="utf-8")) + if not isinstance(case, dict): + raise ValueError(f"{path}: case must be a YAML object") + + name = case.get("name") + client_mode = case.get("clientMode") + steps = case.get("steps") + if not isinstance(name, str) or not name: + raise ValueError(f"{path}: name is required") + if client_mode not in {"direct", "browser_addon"}: + raise ValueError(f"{path}: unsupported clientMode={client_mode!r}") + if not isinstance(steps, list) or not steps: + raise ValueError(f"{path}: at least one step is required") + + for index, step in enumerate(steps, start=1): + expected_code = ( + step.get("expect", {}).get("stageCode") + if isinstance(step, dict) + else None + ) + if not isinstance(step, dict) or not isinstance(step.get("input"), str): + raise ValueError(f"{path}: step {index} input must be a string") + if not isinstance(expected_code, str) or not expected_code: + raise ValueError( + f"{path}: step {index} expect.stageCode is required" + ) + + if name_filters and not any(value in name for value in name_filters): + continue + + case["_path"] = str(path.relative_to(EVALS_DIR)) + cases.append(case) + + if not cases: + raise ValueError(f"No evaluation cases found under {cases_dir}") + return cases + + +def parse_sse(response: httpx.Response) -> ChatResult: + stage_codes: list[str] = [] + errors: list[dict] = [] + event_name = "" + data_lines: list[str] = [] + + def consume_event() -> None: + nonlocal event_name, data_lines + if data_lines: + payload = json.loads("\n".join(data_lines)) + if event_name == "stage_code": + stage_codes.append(payload["nextStageCode"]) + elif event_name == "error": + errors.append(payload) + event_name = "" + data_lines = [] + + for line in response.iter_lines(): + if not line: + consume_event() + elif line.startswith("event:"): + event_name = line.removeprefix("event:").strip() + elif line.startswith("data:"): + data_lines.append(line.removeprefix("data:").strip()) + + consume_event() + return ChatResult(stage_codes=stage_codes, errors=errors) + + +def run_chat_step( + client: httpx.Client, + *, + session_id: str, + client_mode: str, + text: str, +) -> ChatResult: + with client.stream( + "POST", + "/chat", + params={"stream": "true"}, + json={ + "sessionId": session_id, + "timeStamp": timestamp(), + "text": text, + "clientMode": client_mode, + "needFormUpdate": False, + "useTextChunk": False, + }, + ) as response: + response.raise_for_status() + return parse_sse(response) + + +def cleanup_session(client: httpx.Client, session_id: str) -> str | None: + try: + response = client.request( + "DELETE", + "/delete_session", + json={"sessionId": session_id, "timeStamp": timestamp()}, + ) + response.raise_for_status() + payload = response.json() + if payload.get("code") != "200": + return f"cleanup response={payload}" + except (httpx.HTTPError, ValueError) as exc: + return f"cleanup error={exc}" + return None + + +def session_id_for(case_name: str) -> str: + slug = re.sub(r"[^a-zA-Z0-9_-]+", "-", case_name).strip("-") + return f"eval-{slug[:24]}-{uuid4().hex[:12]}" + + +def run_case(client: httpx.Client, case: dict) -> tuple[bool, list[str]]: + session_id = session_id_for(case["name"]) + messages = [] + passed = True + + try: + for step_number, step in enumerate(case["steps"], start=1): + expected = step["expect"]["stageCode"] + try: + result = run_chat_step( + client, + session_id=session_id, + client_mode=case["clientMode"], + text=step["input"], + ) + except (httpx.HTTPError, json.JSONDecodeError, KeyError) as exc: + messages.append(f"step {step_number}: request error: {exc}") + passed = False + break + + if result.stage_codes != [expected]: + messages.append( + f"step {step_number}: expected={[expected]} " + f"actual={result.stage_codes} sse_errors={result.errors}" + ) + passed = False + break + + messages.append(f"step {step_number}: stageCode={expected}") + finally: + cleanup_warning = cleanup_session(client, session_id) + if cleanup_warning: + messages.append(f"warning: {cleanup_warning}") + + return passed, messages + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Run live workflow evaluations and assert stage codes." + ) + parser.add_argument( + "--base-url", + default=os.getenv("EVAL_BASE_URL", "http://127.0.0.1:8000"), + help="FastAPI base URL (default: EVAL_BASE_URL or http://127.0.0.1:8000)", + ) + parser.add_argument( + "--cases-dir", + type=Path, + default=DEFAULT_CASES_DIR, + help="Directory containing YAML cases", + ) + parser.add_argument( + "--case", + action="append", + default=[], + help="Run cases whose names contain this value; repeatable", + ) + parser.add_argument("--timeout", type=float, default=90.0) + parser.add_argument( + "--insecure", + action="store_true", + help="Disable TLS certificate verification", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + try: + cases = load_cases(args.cases_dir, args.case) + except (OSError, ValueError, yaml.YAMLError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 2 + + passed_count = 0 + failed_count = 0 + with httpx.Client( + base_url=args.base_url.rstrip("/"), + timeout=args.timeout, + verify=not args.insecure, + ) as client: + for case in cases: + passed, messages = run_case(client, case) + status = "PASS" if passed else "FAIL" + print(f"{status} {case['name']} ({case['_path']})") + for message in messages: + print(f" {message}") + if passed: + passed_count += 1 + else: + failed_count += 1 + + print( + f"\nSummary: total={len(cases)} " + f"passed={passed_count} failed={failed_count}" + ) + return 1 if failed_count else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pytest.ini b/pytest.ini index 0c543ce..b7a5232 100644 --- a/pytest.ini +++ b/pytest.ini @@ -2,4 +2,3 @@ pythonpath = . markers = integration: marks tests that need access to external services - golden_live: runs golden conversation cases against a live API and FastGPT diff --git a/requirements.txt b/requirements.txt index 3baac68..62fbfeb 100644 --- a/requirements.txt +++ b/requirements.txt @@ -20,3 +20,4 @@ pandas requests sqlalchemy pymysql +PyYAML>=6.0 diff --git a/test/golden/README.md b/test/golden/README.md deleted file mode 100644 index fbe4d4c..0000000 --- a/test/golden/README.md +++ /dev/null @@ -1,32 +0,0 @@ -# Stage-code golden tests - -These tests call the running FastAPI service and its configured FastGPT -workflow. Each conversation step asserts only the `nextStageCode` carried by -the SSE `stage_code` event. - -Edit `cases.json` to add or update conversation cases: - -```json -{ - "name": "example", - "clientMode": "browser_addon", - "steps": [ - { - "text": "继续", - "expect": { - "stageCode": "browser_addon.1002" - } - } - ] -} -``` - -Start the API, then run: - -```bash -RUN_GOLDEN=1 \ -GOLDEN_BASE_URL=http://127.0.0.1:8000 \ -pytest -m golden_live test/golden -v -``` - -Without `RUN_GOLDEN=1`, the cases are skipped and do not make network calls. diff --git a/test/golden/cases.json b/test/golden/cases.json deleted file mode 100644 index 29fd25d..0000000 --- a/test/golden/cases.json +++ /dev/null @@ -1,26 +0,0 @@ -[ - { - "name": "direct_initial_stage", - "clientMode": "direct", - "steps": [ - { - "text": "hi", - "expect": { - "stageCode": "1001" - } - } - ] - }, - { - "name": "browser_addon_initial_stage", - "clientMode": "browser_addon", - "steps": [ - { - "text": "hi", - "expect": { - "stageCode": "browser_addon.1001" - } - } - ] - } -] diff --git a/test/golden/test_stage_code_golden.py b/test/golden/test_stage_code_golden.py deleted file mode 100644 index e0a4675..0000000 --- a/test/golden/test_stage_code_golden.py +++ /dev/null @@ -1,112 +0,0 @@ -import json -import os -from datetime import datetime, timezone -from pathlib import Path -from uuid import uuid4 - -import httpx -import pytest - - -CASES_PATH = Path(__file__).with_name("cases.json") -RUN_GOLDEN = os.getenv("RUN_GOLDEN") == "1" -BASE_URL = os.getenv("GOLDEN_BASE_URL", "http://127.0.0.1:8000").rstrip("/") - - -def load_cases() -> list[dict]: - return json.loads(CASES_PATH.read_text(encoding="utf-8")) - - -def parse_stage_codes(response: httpx.Response) -> list[str]: - stage_codes: list[str] = [] - event_name = "" - data_lines: list[str] = [] - - def consume_event() -> None: - nonlocal event_name, data_lines - if event_name == "stage_code" and data_lines: - payload = json.loads("\n".join(data_lines)) - stage_codes.append(payload["nextStageCode"]) - event_name = "" - data_lines = [] - - for line in response.iter_lines(): - if not line: - consume_event() - elif line.startswith("event:"): - event_name = line.removeprefix("event:").strip() - elif line.startswith("data:"): - data_lines.append(line.removeprefix("data:").strip()) - - consume_event() - return stage_codes - - -def run_chat_step( - client: httpx.Client, - *, - session_id: str, - client_mode: str, - text: str, -) -> list[str]: - payload = { - "sessionId": session_id, - "timeStamp": datetime.now(timezone.utc).isoformat(), - "text": text, - "clientMode": client_mode, - "needFormUpdate": False, - "useTextChunk": False, - } - - with client.stream( - "POST", - "/chat", - params={"stream": "true"}, - json=payload, - ) as response: - response.raise_for_status() - return parse_stage_codes(response) - - -def cleanup_session(client: httpx.Client, session_id: str) -> None: - try: - client.request( - "DELETE", - "/delete_session", - json={ - "sessionId": session_id, - "timeStamp": datetime.now(timezone.utc).isoformat(), - }, - ) - except httpx.HTTPError: - # A unique session ID is used for every run, so cleanup failure must not - # hide the stage-code assertion that the golden test is responsible for. - pass - - -@pytest.mark.golden_live -@pytest.mark.skipif( - not RUN_GOLDEN, - reason="Set RUN_GOLDEN=1 to run live golden cases", -) -@pytest.mark.parametrize("case", load_cases(), ids=lambda case: case["name"]) -def test_expected_stage_codes(case): - session_id = f"golden-{case['name']}-{uuid4().hex}" - - with httpx.Client(base_url=BASE_URL, timeout=90.0) as client: - try: - for step_number, step in enumerate(case["steps"], start=1): - actual_codes = run_chat_step( - client, - session_id=session_id, - client_mode=case["clientMode"], - text=step["text"], - ) - expected_code = step["expect"]["stageCode"] - - assert actual_codes == [expected_code], ( - f"case={case['name']} step={step_number} " - f"expected={[expected_code]} actual={actual_codes}" - ) - finally: - cleanup_session(client, session_id)