harness_zbt/backend/app/commands/evaluate_agents.py

"""使用真实 HTTP、真实大模型和 LangSmith Trace 评测两种 Agent 人格。"""

import argparse
import json
import sys
import time
from contextlib import suppress
from pathlib import Path
from typing import Any
from uuid import NAMESPACE_URL, UUID, uuid5

import httpx
from langsmith import Client

from app.core.config import Settings

BACKEND_ROOT = Path(__file__).resolve().parents[2]


def load_cases(suite: str) -> list[dict[str, Any]]:
    suffix = "smoke" if suite == "smoke" else "full"
    cases: list[dict[str, Any]] = []
    for persona in ("customer", "operation"):
        path = BACKEND_ROOT / "evals" / f"{persona}-{suffix}.jsonl"
        cases.extend(
            json.loads(line)
            for line in path.read_text(encoding="utf-8").splitlines()
            if line.strip()
        )
    return cases


def login(client: httpx.Client, persona: str, admin_password: str) -> str:
    if persona == "customer":
        response = client.post(
            "/h5/auth/login",
            json={"mobile": "18800000001", "code": "147258"},
        )
    else:
        response = client.post(
            "/admin/auth/login",
            json={"username": "admin", "password": admin_password},
        )
    response.raise_for_status()
    return str(response.json()["data"]["tokens"]["access_token"])


def run_case(
    client: httpx.Client,
    case: dict[str, Any],
    admin_password: str,
) -> dict[str, Any]:
    persona = str(case["persona"])
    token = login(client, persona, admin_password)
    prefix = "" if persona == "customer" else "/admin"
    headers = {"Authorization": f"Bearer {token}"}
    thread_response = client.post(
        f"{prefix}/agent/threads",
        headers=headers,
        json={"title": f"评测-{case['id']}"},
    )
    thread_response.raise_for_status()
    thread_id = thread_response.json()["data"]["id"]
    response = client.post(
        f"{prefix}/agent/threads/{thread_id}/messages",
        headers=headers,
        json={
            "content": {"type": "text", "text": case["input"]},
            "client_message_id": f"eval-{case['id']}",
        },
        timeout=120,
    )
    response.raise_for_status()
    data = response.json()["data"]
    run_id = str(data["run_id"])
    for _ in range(5):
        if data.get("trace_url"):
            break
        time.sleep(1)
        detail = client.get(
            f"{prefix}/agent/runs/{run_id}",
            headers=headers,
        )
        detail.raise_for_status()
        data["trace_url"] = detail.json()["data"].get("trace_url")

    text = str(data["assistant_message"]["text"])
    tools = set(data.get("invoked_tools", []))
    expected_tools = set(case["expected_tools"])
    expected_terms = [str(term) for term in case["expected_terms"]]
    scores = {
        "response_nonempty": bool(text.strip()),
        "tool_coverage": expected_tools.issubset(tools),
        "term_coverage": all(term in text for term in expected_terms),
        "trace_linked": bool(data.get("trace_id")),
    }
    return {
        "id": case["id"],
        "persona": persona,
        "run_id": run_id,
        "trace_id": data.get("trace_id"),
        "trace_url": data.get("trace_url"),
        "invoked_tools": sorted(tools),
        "scores": scores,
        "passed": all(scores.values()),
        "answer": text,
    }


def sync_langsmith(
    settings: Settings,
    suite: str,
    cases: list[dict[str, Any]],
    results: list[dict[str, Any]],
) -> None:
    if not settings.langsmith_api_key:
        return
    client = Client(
        api_url=settings.langsmith_endpoint,
        api_key=settings.langsmith_api_key,
    )
    dataset_name = f"智保通-S1-{suite}"
    if not client.has_dataset(dataset_name=dataset_name):
        client.create_dataset(
            dataset_name,
            description="智保通第一阶段客户与运营 Agent 真实模型评测集",
            metadata={"project": "智保通", "stage": 1, "suite": suite},
        )
    result_by_id = {result["id"]: result for result in results}
    for case in cases:
        example_id = uuid5(NAMESPACE_URL, f"{dataset_name}:{case['id']}")
        # 示例 ID 固定；重复执行时保留第一次创建的基准样本。
        with suppress(Exception):
            client.create_example(
                example_id=example_id,
                dataset_name=dataset_name,
                inputs={"persona": case["persona"], "text": case["input"]},
                outputs={
                    "expected_tools": case["expected_tools"],
                    "expected_terms": case["expected_terms"],
                },
                metadata={"suite": suite, "case_id": case["id"]},
            )
        result = result_by_id[str(case["id"])]
        trace_id = result.get("trace_id")
        if not trace_id:
            continue
        for key, score in result["scores"].items():
            client.create_feedback(
                run_id=UUID(str(trace_id)),
                key=f"eval.{key}",
                score=bool(score),
                comment=f"智保通 S1 {suite} 自动评测",
            )


def main() -> int:
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8")
    parser = argparse.ArgumentParser()
    parser.add_argument("--suite", choices=("smoke", "full"), default="smoke")
    parser.add_argument("--base-url", default="http://127.0.0.1:8000/api/v1")
    parser.add_argument("--admin-password", default="zaq1XSW@")
    arguments = parser.parse_args()
    settings = Settings()
    cases = load_cases(arguments.suite)
    with httpx.Client(base_url=arguments.base_url, timeout=120) as client:
        results = [run_case(client, case, arguments.admin_password) for case in cases]
    sync_langsmith(settings, arguments.suite, cases, results)
    print(json.dumps({"suite": arguments.suite, "results": results}, ensure_ascii=False, indent=2))
    return 0 if all(result["passed"] for result in results) else 1


if __name__ == "__main__":
    raise SystemExit(main())