Move workflow evaluations to standalone runner

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.

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

evals/cases/direct/initial.yaml

@@ -0,0 +1,7 @@
+name: direct_initial_stage
+clientMode: direct
+
+steps:
+  - input: hi
+    expect:
+      stageCode: "1001"

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

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

requirements.txt

@@ -20,3 +20,4 @@ pandas
 requests
 sqlalchemy
 pymysql
+PyYAML>=6.0

test/golden/README.md

@@ -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.

test/golden/cases.json

@@ -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"
-        }
-      }
-    ]
-  }
-]

test/golden/test_stage_code_golden.py

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