vent_agent_2.0/scripts/airflow_e2e.py

"""风量研判的真实端到端验证：用真实模型跑一条提示词，观察技能是否被触发与遵循。

观察点（按 skill-creator 的评测要求，同时看结果与轨迹）：
1. 技能是否被触发（主 Agent 是否委派给 airflow-review-orchestrator 子 Agent）；
2. 子 Agent 是否读了 /skills/airflow-review/SKILL.md，并按索引读了 references/stages/01-*.md；
3. 是否按契约调用了 parse_airflow_measurement_report（参数名与 scope 是否正确）；
4. 返回值里的真实读数是否被如实引用（总进风 35181 / 总回风 35713）；
5. 是否把 pending_review 与依据不足原样报出，而不是编造结论。

用法（需服务已启动且模型密钥可用）：

    python main.py &                                   # 或 bash scripts/start_server.sh start
    python scripts/airflow_e2e.py                      # 默认：测风报表 + 数据准备提示词
    AIRFLOW_E2E_FILE="布尔台矿矿井风量分配计划20241201(1).pdf"     AIRFLOW_E2E_PROMPT="请对布尔台矿做完整研判……" python scripts/airflow_e2e.py

这不是回归测试（依赖真实模型与网络），是一次需要人工判读的端到端验证；
离线回归在 tests/test_airflow_*.py。
"""
from __future__ import annotations

import json
import os
import sys
from typing import Any
from pathlib import Path

import httpx

ROOT = Path(__file__).resolve().parents[1]
BASE = os.getenv("AIRFLOW_E2E_BASE", "http://127.0.0.1:8080")


def raw_materials_dir() -> Path | None:
    """真实资料目录：位置由使用者决定，因此可探测或显式指定。

    依次看：AIRFLOW_RAW_DIR（本地便利开关）→ 仓库根 矿井风量研判资料。
    真实资料不入库，缺失时跳过。
    """
    candidates = []
    override = os.getenv("AIRFLOW_RAW_DIR", "").strip()
    if override:
        candidates.append(Path(override) if Path(override).is_absolute() else ROOT / override)
    candidates.append(ROOT / "矿井风量研判资料")
    for candidate in candidates:
        if candidate.is_dir() and (any(candidate.glob("*.pdf")) or any(candidate.glob("*.xlsx"))):
            return candidate
    return None


PROMPT = os.getenv("AIRFLOW_E2E_PROMPT") or (
    "用户上传了布尔台矿的测风报表，请先把这份报表的数据接进来做规范化，"
    "把矿井总进风、总回风、各回风井、主要通风机读出来，"
    "并说明哪些测点需要人工确认。不要做需风量比对。")
UPLOAD = os.getenv("AIRFLOW_E2E_FILE", "测风报表20260528 .xlsx")
# 置 AIRFLOW_E2E_FILE="" 则不传附件，改为让智能体自行确认资料是否齐备


def sse_events(response: httpx.Response) -> list[dict]:
    events: list[dict] = []
    for line in response.iter_lines():
        if line.startswith("data: "):
            try:
                events.append(json.loads(line[6:]))
            except json.JSONDecodeError:
                continue
    return events


def main() -> int:
    files: dict[str, Any] = {}
    directory = raw_materials_dir() if UPLOAD else None
    if not UPLOAD:
        data = {"message": PROMPT, "mode": "full"}
        return _run(data, {})
    if directory is None:
        print("未找到真实资料目录（可用 AIRFLOW_RAW_DIR 指定），跳过")
        return 0
    # 支持一次上传多份（用 2.0 的 files 多附件字段）：AIRFLOW_E2E_FILES 逗号分隔，
    # 不给就用 AIRFLOW_E2E_FILE 的单份 file 字段。
    multi = [name.strip() for name in os.getenv("AIRFLOW_E2E_FILES", "").split(",") if name.strip()]
    if multi:
        # httpx 的多值字段要传「(字段名, (文件名, 内容)) 的列表」，不是字典
        payload = [("files", (name, (directory / name).read_bytes(), "application/octet-stream"))
                   for name in multi]
        return _run({"message": PROMPT, "mode": "full"}, payload)
    report = directory / UPLOAD
    if not report.is_file():
        print("缺少真实报表，跳过")
        return 0
    files = {"file": (report.name, report.read_bytes(),
                      "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}
    data = {"message": PROMPT, "file_name": report.name, "mode": "full"}
    return _run(data, files)
def _run(data: dict, files: dict) -> int:
    with httpx.Client(timeout=float(os.getenv("AIRFLOW_E2E_TIMEOUT", "560"))) as client:
        with client.stream("POST", f"{BASE}/api/chat", data=data, files=files) as response:
            response.raise_for_status()
            events = sse_events(response)

    print(f"事件数：{len(events)}")
    types = [event.get("type") for event in events]
    print("事件类型：", {kind: types.count(kind) for kind in sorted(set(types))})

    tool_calls = [event for event in events if event.get("type") == "tool_call"]
    print(f"\n工具调用（{len(tool_calls)} 次）：")
    for event in tool_calls:
        print(f"  - {event.get('tool')}  {str(event.get('args'))[:150]}")

    text = "".join(event.get("content", "") for event in events if event.get("type") == "token")
    print(f"\n回答正文（{len(text)} 字符）：\n")
    print(text[:3000])

    print("\n---- 判读 ----")
    names = [event.get("tool") for event in tool_calls]
    print("① 委派子 Agent（task 调用）：", "task" in names)
    print("   全流程技能是否被读：", any("/skills/airflow-review" in str(e.get("args")) for e in tool_calls))
    print("② 调了报表解析工具：", "parse_airflow_measurement_report" in names)
    print("③ 回答含真实总进风 35181：", "35181" in text)
    print("④ 回答含真实总回风 35713：", "35713" in text)
    print("⑤ 提到待人工确认的测点：", ("人工" in text or "复核" in text))
    return 0


if __name__ == "__main__":
    sys.exit(main())