用户：「检查一下 AI 平台的健康状态」
        │
        ▼
Dify 新 Agent（207）
        │  三次 HTTP 工具（经 ssrf_proxy）
        ▼
FastAPI :8779（207）
        │  paramiko SSH
        ▼
GPU 机 116.62.10.183
        ├─ docker inspect vllm-qwen3.6-27b     → get_vllm_status
        ├─ nvidia-smi                           → get_gpu_info
        └─ curl 127.0.0.1:8000/metrics          → get_request_queue

① 数据采集链（真实指标从 GPU 来）
   207 FastAPI :8779 --SSH--> GPU 116.62.10.183
   ├─ docker inspect vllm-qwen3.6-27b     → 容器是否 running
   ├─ nvidia-smi                          → GPU 利用率/显存/温度
   └─ curl 127.0.0.1:8000/metrics         → vLLM 排队数、KV Cache

② 工具链
   用户 → Dify Agent → OpenAPI「vllm_health」→ :8779 → SSH GPU

③ 决策链
   system_prompt 要求 Agent 必须调齐三个工具再出报告

/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业5_vLLM健康检查Agent/作业5_00_解析自测.py
/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业5_vLLM健康检查Agent/作业5_01_部署服务.py
/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业5_vLLM健康检查Agent/作业5_02_注册OpenAPI工具.py
/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业5_vLLM健康检查Agent/作业5_03_创建Agent空壳.py
/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业5_vLLM健康检查Agent/作业5_04_查ID并挂工具.py

bash /opt/workbuddy/scripts_bash_cursor/作业5_vLLM健康检查Agent/作业5_启动_vllm_health.sh
curl -s http://127.0.0.1:8779/vllm/status
curl -s http://127.0.0.1:8779/gpu/info
curl -s http://127.0.0.1:8779/vllm/queue

/usr/bin/python3 /opt/workbuddy/scripts_cursor/作业5_vLLM健康检查Agent/作业5_05_对话验收.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""作业 5 · 从 GPU 机拉取 vLLM / nvidia-smi / 排队指标。"""
from __future__ import annotations

from typing import Callable, Optional

import _config as C

Runner = Callable[[str], str]


def gpu_exec(cmd: str, runner: Optional[Runner] = None, timeout: int = 40) -> str:
    if runner is not None:
        return runner(cmd)
    import paramiko

    c = paramiko.SSHClient()
    c.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    c.connect(C.GPU_HOST, username=C.GPU_USER, password=C.GPU_PASS, timeout=20)
    try:
        _i, o, e = c.exec_command(cmd, timeout=timeout)
        out = o.read().decode("utf-8", "replace")
        err = e.read().decode("utf-8", "replace")
        if not out.strip() and err.strip():
            raise RuntimeError(err.strip())
        return out
    finally:
        c.close()


def parse_docker_status(raw: str) -> dict:
    line = (raw or "").strip().splitlines()[0] if (raw or "").strip() else ""
    parts = line.split("|")
    if len(parts) < 3:
        raise ValueError("unexpected docker inspect: %r" % raw)
    name = parts[0].lstrip("/")
    running = parts[1].strip().lower() == "true"
    return {
        "container": name,
        "running": running,
        "status": parts[2].strip(),
        "started_at": parts[3].strip() if len(parts) > 3 else "",
        "image": parts[4].strip() if len(parts) > 4 else "",
        "healthy": running and parts[2].strip() == "running",
    }


def parse_nvidia_smi(raw: str) -> dict:
    gpus = []
    for line in (raw or "").splitlines():
        line = line.strip()
        if not line:
            continue
        parts = [p.strip() for p in line.split(",")]
        if len(parts) < 6:
            continue
        mem_used = int(parts[4].replace("MiB", "").strip() or "0")
        mem_total = int(parts[5].replace("MiB", "").strip() or "0")
        gpus.append(
            {
                "index": int(parts[0]),
                "name": parts[1],
                "temperature_c": int(parts[2]),
                "utilization": parts[3],
                "memory_used_mib": mem_used,
                "memory_total_mib": mem_total,
                "memory": "%s/%s MiB" % (mem_used, mem_total),
            }
        )
    if not gpus:
        raise ValueError("unexpected nvidia-smi: %r" % raw)
    return {"host": C.GPU_HOST, "gpus": gpus}


def _metric_value(text: str, exact_name: str) -> Optional[float]:
    for line in (text or "").splitlines():
        s = line.strip()
        if not s or s.startswith("#"):
            continue
        name = s.split("{", 1)[0].split(" ", 1)[0]
        if name != exact_name:
            continue
        try:
            return float(s.rsplit(None, 1)[-1])
        except ValueError:
            continue
    return None


def parse_vllm_queue(raw: str) -> dict:
    waiting = _metric_value(raw, "vllm:num_requests_waiting")
    running = _metric_value(raw, "vllm:num_requests_running")
    kv = _metric_value(raw, "vllm:kv_cache_usage_perc")
    return {
        "waiting": 0.0 if waiting is None else waiting,
        "running": 0.0 if running is None else running,
        "kv_cache_usage_perc": kv,
        "source": "vLLM /metrics",
    }


def get_vllm_status(runner: Optional[Runner] = None) -> dict:
    cmd = (
        "docker inspect -f '{{.Name}}|{{.State.Running}}|{{.State.Status}}|"
        "{{.State.StartedAt}}|{{.Config.Image}}' %s"
        % C.GPU_VLLM_CONTAINER
    )
    raw = gpu_exec(cmd, runner=runner)
    out = parse_docker_status(raw)
    out["host"] = C.GPU_HOST
    return out


def get_gpu_info(runner: Optional[Runner] = None) -> dict:
    cmd = (
        "nvidia-smi --query-gpu=index,name,temperature.gpu,utilization.gpu,"
        "memory.used,memory.total --format=csv,noheader"
    )
    raw = gpu_exec(cmd, runner=runner)
    return parse_nvidia_smi(raw)


def get_request_queue(runner: Optional[Runner] = None) -> dict:
    cmd = (
        "curl -s -m 10 -H 'Authorization: Bearer %s' http://127.0.0.1:8000/metrics"
        % C.VLLM_KEY
    )
    raw = gpu_exec(cmd, runner=runner)
    out = parse_vllm_queue(raw)
    out["host"] = C.GPU_HOST
    out["model"] = C.QWEN_MODEL
    return out

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""作业 5 · vLLM 健康检查 HTTP 工具（FastAPI）。

207 上启动：
  bash /opt/workbuddy/scripts_bash_cursor/作业5_vLLM健康检查Agent/作业5_启动_vllm_health.sh
"""
from __future__ import annotations

import _bootstrap  # noqa: F401
from fastapi import FastAPI

from 作业5_vllm_health import get_gpu_info, get_request_queue, get_vllm_status

app = FastAPI(
    title="vLLM Health API",
    version="1.0.0",
    description="作业 5：检查 GPU 机 vLLM 容器、GPU 指标、请求排队",
)


@app.get("/vllm/status", operation_id="get_vllm_status", summary="检查 vLLM 容器是否运行")
def api_vllm_status() -> dict:
    return get_vllm_status()


@app.get("/gpu/info", operation_id="get_gpu_info", summary="获取 GPU 利用率、显存、温度")
def api_gpu_info() -> dict:
    return get_gpu_info()


@app.get("/vllm/queue", operation_id="get_request_queue", summary="获取当前排队请求数")
def api_queue() -> dict:
    return get_request_queue()


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


if __name__ == "__main__":
    import uvicorn
    import _config as C

    uvicorn.run("作业5_vllm_health_api:app", host="0.0.0.0", port=C.VLLM_HEALTH_PORT, reload=False)

#!/usr/bin/env bash
# 作业 5 · 在 207 上启动 vLLM 健康检查 FastAPI
set -euo pipefail

DIR="/opt/workbuddy/scripts_cursor/作业5_vLLM健康检查Agent"
PORT="${VLLM_HEALTH_PORT:-8779}"
PIDFILE="/tmp/vllm_health_api.pid"
LOG="/tmp/vllm_health_api.log"

pip3 install fastapi uvicorn paramiko -q

if [[ -f "$PIDFILE" ]] && kill -0 "$(cat "$PIDFILE")" 2>/dev/null; then
  echo "vllm_health_api already running pid=$(cat "$PIDFILE")"
  curl -sf "http://127.0.0.1:${PORT}/health" && echo
  exit 0
fi

cd "$DIR"
nohup python3 -m uvicorn 作业5_vllm_health_api:app \
  --host 0.0.0.0 --port "$PORT" >>"$LOG" 2>&1 &
echo $! >"$PIDFILE"
sleep 2
curl -sf "http://127.0.0.1:${PORT}/health" && echo
echo "started on :${PORT}, log=$LOG"

openapi: 3.0.0
info:
  title: vLLM Health API
  version: 1.0.0
  description: 作业 5 · 检查 GPU 机 vLLM 容器、GPU 指标、请求排队
servers:
  - url: http://192.168.110.207:8779
paths:
  /vllm/status:
    get:
      operationId: get_vllm_status
      summary: 检查 vLLM 容器是否运行
      description: 通过 SSH 到 GPU 机执行 docker inspect，返回容器是否 running
      responses:
        "200":
          description: 成功
          content:
            application/json:
              schema:
                type: object
                properties:
                  host:
                    type: string
                  container:
                    type: string
                  running:
                    type: boolean
                  status:
                    type: string
                  started_at:
                    type: string
                  image:
                    type: string
                  healthy:
                    type: boolean
  /gpu/info:
    get:
      operationId: get_gpu_info
      summary: 获取 GPU 利用率、显存、温度
      description: 在 GPU 机执行 nvidia-smi 查询
      responses:
        "200":
          description: 成功
          content:
            application/json:
              schema:
                type: object
                properties:
                  host:
                    type: string
                  gpus:
                    type: array
                    items:
                      type: object
  /vllm/queue:
    get:
      operationId: get_request_queue
      summary: 获取当前排队请求数
      description: 读取 vLLM /metrics 中的 num_requests_waiting
      responses:
        "200":
          description: 成功
          content:
            application/json:
              schema:
                type: object
                properties:
                  host:
                    type: string
                  model:
                    type: string
                  waiting:
                    type: number
                  running:
                    type: number
                  kv_cache_usage_perc:
                    type: number
                  source:
                    type: string

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""作业 5 · 指标解析自测（不连 GPU，先红后绿）。"""
from __future__ import annotations

import sys

import _bootstrap  # noqa: F401
from 作业5_vllm_health import (
    parse_docker_status,
    parse_nvidia_smi,
    parse_vllm_queue,
    get_gpu_info,
    get_request_queue,
    get_vllm_status,
)


def test_parse_docker_status() -> None:
    raw = "/vllm-qwen3.6-27b|true|running|2026-08-18T01:00:00Z|vllm/vllm-openai:latest"
    got = parse_docker_status(raw)
    assert got["running"] is True
    assert got["status"] == "running"
    assert got["container"] == "vllm-qwen3.6-27b"


def test_parse_nvidia_smi() -> None:
    raw = "0, NVIDIA H20, 37, 0 %, 83868 MiB, 97871 MiB"
    got = parse_nvidia_smi(raw)
    assert got["gpus"][0]["name"] == "NVIDIA H20"
    assert got["gpus"][0]["temperature_c"] == 37
    assert got["gpus"][0]["utilization"] == "0 %"
    assert got["gpus"][0]["memory_used_mib"] == 83868
    assert got["gpus"][0]["memory_total_mib"] == 97871


def test_parse_vllm_queue() -> None:
    raw = """
# HELP vllm:num_requests_waiting ...
vllm:num_requests_running{engine="0",model_name="qwen3.6-27b"} 1.0
vllm:num_requests_waiting{engine="0",model_name="qwen3.6-27b"} 3.0
vllm:num_requests_waiting_by_reason{engine="0",model_name="qwen3.6-27b",reason="capacity"} 2.0
vllm:kv_cache_usage_perc{engine="0",model_name="qwen3.6-27b"} 0.42
"""
    got = parse_vllm_queue(raw)
    assert got["waiting"] == 3.0
    assert got["running"] == 1.0
    assert got["kv_cache_usage_perc"] == 0.42


def test_tools_use_injected_runner() -> None:
    def runner(cmd: str) -> str:
        if "inspect" in cmd:
            return "/vllm-qwen3.6-27b|true|running|t|img"
        if "nvidia-smi" in cmd:
            return "0, NVIDIA H20, 37, 0 %, 100 MiB, 200 MiB"
        if "metrics" in cmd:
            return 'vllm:num_requests_waiting{engine="0"} 5.0\nvllm:num_requests_running{engine="0"} 1.0\n'
        raise AssertionError(cmd)

    st = get_vllm_status(runner=runner)
    gpu = get_gpu_info(runner=runner)
    q = get_request_queue(runner=runner)
    assert st["running"] is True
    assert gpu["gpus"][0]["name"] == "NVIDIA H20"
    assert q["waiting"] == 5.0


def main() -> int:
    test_parse_docker_status()
    test_parse_nvidia_smi()
    test_parse_vllm_queue()
    test_tools_use_injected_runner()
    print("[PASS] 作业5 解析自测")
    return 0


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""作业 5 · 步骤 1：在 207 启动健康检查 API，并实打 GPU 机。"""
from __future__ import annotations

import sys
from pathlib import Path

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify

LOCAL_SH = (
    Path(__file__).resolve().parents[2]
    / "scripts_bash_cursor"
    / "作业5_vLLM健康检查Agent"
    / "作业5_启动_vllm_health.sh"
)


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        d.sftp.put(str(LOCAL_SH), "/tmp/作业5_启动_vllm_health.sh")
        out, err = d.sh("bash /tmp/作业5_启动_vllm_health.sh")
        print(out or err)
        checks = [
            ("%s/health" % C.VLLM_HEALTH_BASE, "ok"),
            ("%s/vllm/status" % C.VLLM_HEALTH_BASE, C.GPU_VLLM_CONTAINER),
            ("%s/gpu/info" % C.VLLM_HEALTH_BASE, "H20"),
            ("%s/vllm/queue" % C.VLLM_HEALTH_BASE, "waiting"),
        ]
        for url, needle in checks:
            out, err = d.sh("curl -sf -m 25 '%s'" % url)
            text = out or err or ""
            print(url, text[:500])
            if needle not in text:
                print("[FAIL] 未包含 %s" % needle)
                return 1
        print("[OK] 健康检查 API 已在 %s 运行，且已打到 GPU 机" % C.VLLM_HEALTH_BASE)
        return 0
    finally:
        d.close()


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""作业 5 · 步骤 2：在 Dify 注册 OpenAPI 工具（vLLM 健康检查）。"""
from __future__ import annotations

import sys
from pathlib import Path

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify, pp

SCHEMA_PATH = (
    Path(__file__).resolve().parents[2]
    / "patches_cursor"
    / "作业5_vLLM健康检查Agent"
    / "作业5_vllm_health_openapi.yaml"
)


def main() -> int:
    schema = SCHEMA_PATH.read_text(encoding="utf-8")
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        existing = d.get("/workspaces/current/tools/api")
        names = []
        for item in existing.get("json") or []:
            names.append(item.get("name") or item.get("provider"))
        print("existing api tools:", names)
        if C.VLLM_HEALTH_PROVIDER in names:
            r_del = d.post(
                "/workspaces/current/tool-provider/api/delete",
                body={"provider": C.VLLM_HEALTH_PROVIDER},
            )
            pp("delete old", r_del)

        body = {
            "provider": C.VLLM_HEALTH_PROVIDER,
            "credentials": {"auth_type": "none"},
            "schema_type": "openapi",
            "schema": schema,
            "icon": {"background": "#E0F2FE", "content": "🩺"},
            "privacy_policy": "",
            "custom_disclaimer": "",
            "labels": ["ops", "homework5", "aiops"],
        }
        r = d.post("/workspaces/current/tool-provider/api/add", body=body)
        pp("add api tool", r)
        if r.get("code") not in ("200", "201"):
            return 1
        r2 = d.get(
            "/workspaces/current/tool-provider/api/tools?provider=%s"
            % C.VLLM_HEALTH_PROVIDER
        )
        pp("list tools", r2, limit=4000)
        return 0
    finally:
        d.close()


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""作业 5 · 步骤 3：创建 vLLM 健康检查 Agent 空壳。"""
from __future__ import annotations

import sys

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify, pp


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        r_list = d.get("/agent?page=1&limit=50")
        data = (r_list.get("json") or {}).get("data") or []
        for item in data:
            if item.get("name") == C.HOMEWORK5_APP_NAME:
                print(
                    "already exists app_id=",
                    item.get("id"),
                    "agent_id=",
                    item.get("agent_id") or item.get("id"),
                )
                print("下一步：运行 作业5_04_查ID并挂工具.py")
                return 0

        body = {
            "name": C.HOMEWORK5_APP_NAME,
            "description": "本周实战作业：AIOps 最小闭环 · 检查 vLLM 容器 / GPU / 排队",
            "icon_type": "emoji",
            "icon": "🩺",
            "icon_background": "#E0F2FE",
            "role": "ops",
        }
        r = d.post("/agent", body)
        pp("create agent", r)
        j = r.get("json") or {}
        print("app_id", j.get("id"))
        print("agent_id", j.get("agent_id") or j.get("id"))
        if r.get("code") not in ("200", "201"):
            return 1
        print("下一步：运行 作业5_04_查ID并挂工具.py")
        return 0
    finally:
        d.close()


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""作业 5 · 步骤 4：查 ID，挂 3 个健康检查工具，写提示词。"""
from __future__ import annotations

import json
import re
import sys
from pathlib import Path

import _bootstrap  # noqa: F401
import _agent_cfg as AC
import _config as C
from dify_api import Dify, pp

PREFERRED = ("get_vllm_status", "get_gpu_info", "get_request_queue")

SYSTEM_PROMPT = """你是企业内部 AI 平台运维助手。当用户询问 AI 平台健康状态、vLLM、GPU、排队、是否卡顿时，
必须依次调用以下三个工具，再根据工具返回结果用中文输出结构化报告，不要编造指标：

1. get_vllm_status：检查 vLLM 容器是否运行
2. get_gpu_info：获取 GPU 利用率、显存、温度
3. get_request_queue：获取当前排队请求数（以及 running / KV Cache 占用）

报告格式：
## AI 平台健康报告
- 结论：健康 / 告警 / 异常（一句话）
- vLLM 容器：{container}，状态 {status}，running={running}
- GPU：名称、利用率、显存、温度
- 请求队列：waiting / running / kv_cache_usage_perc
- 建议：若 waiting>0 或 GPU 利用率长期 100% 或容器未运行，给出下一步

工具失败时如实说明失败原因。"""


def find_ids(d: Dify) -> tuple[str, str, str, str]:
    r_list = d.get("/agent?page=1&limit=50")
    app_id = agent_id = ""
    for item in (r_list.get("json") or {}).get("data") or []:
        if item.get("name") == C.HOMEWORK5_APP_NAME:
            app_id = item.get("id") or ""
            agent_id = item.get("agent_id") or item.get("id") or ""
            break
    if not agent_id:
        raise RuntimeError("列表里找不到应用：%s" % C.HOMEWORK5_APP_NAME)

    sql = (
        "SELECT a.app_id, a.id, a.active_config_snapshot_id, d.id "
        "FROM agents a JOIN agent_config_drafts d ON d.agent_id=a.id "
        "WHERE a.id='%s' LIMIT 1;" % agent_id
    )
    out, err = d.sh(
        'docker exec %s psql -U postgres -d dify -t -A -F "|" -c "%s"'
        % (C.PG_CONTAINER, sql)
    )
    line = (out or "").strip().splitlines()
    if not line or "|" not in line[0]:
        sql2 = (
            "SELECT a.app_id, a.id, a.active_config_snapshot_id, d.id "
            "FROM agents a JOIN agent_config_drafts d ON d.agent_id=a.id "
            "WHERE a.app_id='%s' LIMIT 1;" % app_id
        )
        out, err = d.sh(
            'docker exec %s psql -U postgres -d dify -t -A -F "|" -c "%s"'
            % (C.PG_CONTAINER, sql2)
        )
        line = (out or "").strip().splitlines()
    if not line or "|" not in line[0]:
        raise RuntimeError("找不到快照/草稿: %s %s" % (out, err))
    app_id, agent_id, snap_id, draft_id = line[0].split("|")
    return app_id, agent_id, snap_id, draft_id


def resolve_tool_entries(d: Dify) -> list[dict]:
    out, err = d.sh(
        "docker exec %s psql -U postgres -d dify -t -A -c "
        "\"SELECT id FROM tool_api_providers WHERE name='%s' LIMIT 1;\""
        % (C.PG_CONTAINER, C.VLLM_HEALTH_PROVIDER)
    )
    provider_uuid = (out or "").strip()
    if not provider_uuid:
        raise RuntimeError(
            "找不到 tool_api_providers.name=%s: %s" % (C.VLLM_HEALTH_PROVIDER, err)
        )

    r = d.get(
        "/workspaces/current/tool-provider/api/tools?provider=%s"
        % C.VLLM_HEALTH_PROVIDER
    )
    pp("api tools meta", r, limit=4000)
    data = r.get("json") or []
    if isinstance(data, dict):
        data = data.get("data") or data.get("tools") or [data]
    if not data:
        raise RuntimeError("未找到 provider=%s 的工具列表" % C.VLLM_HEALTH_PROVIDER)

    by_name = {}
    for tool in data:
        name = tool.get("name") or tool.get("tool_name") or ""
        by_name[name] = {
            "enabled": True,
            "provider_type": "api",
            "provider_id": provider_uuid,
            "tool_name": name,
            "credential_type": "unauthorized",
            "credential_ref": None,
            "runtime_parameters": {},
        }

    entries = []
    for name in PREFERRED:
        if name not in by_name:
            raise RuntimeError("缺少工具 %s，实际=%s" % (name, list(by_name)))
        entries.append(by_name[name])
    return entries


def patch_cfg(cfg: dict, tool_entries: list[dict]) -> dict:
    prompt = cfg.setdefault("prompt", {})
    if prompt is None:
        cfg["prompt"] = {}
        prompt = cfg["prompt"]
    prompt["system_prompt"] = SYSTEM_PROMPT

    tools = cfg.get("tools")
    if not isinstance(tools, dict):
        tools = {}
        cfg["tools"] = tools
    tools["cli_tools"] = tools.get("cli_tools") or []
    tools["dify_tools"] = tool_entries

    model = cfg.get("model")
    if not isinstance(model, dict):
        model = {}
        cfg["model"] = model
    model["model"] = C.QWEN_MODEL
    model["plugin_id"] = "langgenius/openai"
    model["model_provider"] = C.OPENAI_PROVIDER
    model["credential_ref"] = None
    ms = model.get("model_settings")
    if not isinstance(ms, dict):
        ms = {}
        model["model_settings"] = ms
    ms["max_tokens"] = ms.get("max_tokens") or 4096
    for k in (
        "stop",
        "top_p",
        "temperature",
        "response_format",
        "presence_penalty",
        "frequency_penalty",
    ):
        ms.setdefault(k, None)
    return cfg


def write_ids(app_id: str, agent_id: str, snap_id: str, draft_id: str) -> None:
    mapping = {
        "HOMEWORK5_APP_ID": app_id,
        "HOMEWORK5_AGENT_ID": agent_id,
        "HOMEWORK5_SNAP_ID": snap_id,
        "HOMEWORK5_DRAFT_ID": draft_id,
    }
    paths = [
        Path(__file__).resolve().parent / "_config.py",
        Path(__file__).resolve().parents[1] / "_config.py",
    ]
    for path in paths:
        if not path.is_file():
            continue
        text = path.read_text(encoding="utf-8")
        for key, val in mapping.items():
            text, n = re.subn(
                r'%s = "[^"]*"' % key,
                '%s = "%s"' % (key, val),
                text,
                count=1,
            )
            if n != 1:
                raise RuntimeError("未能更新 %s 中的 %s" % (path, key))
        path.write_text(text, encoding="utf-8")
        print("updated", path)


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        app_id, agent_id, snap_id, draft_id = find_ids(d)
        print("app_id", app_id)
        print("agent_id", agent_id)
        print("snap_id", snap_id)
        print("draft_id", draft_id)

        tool_entries = resolve_tool_entries(d)
        print("tool_entries", json.dumps(tool_entries, ensure_ascii=False))

        snap = AC.load_cfg(d, "agent_config_snapshots", snap_id)
        draft = AC.load_cfg(d, "agent_config_drafts", draft_id)
        AC.save_cfg(d, "agent_config_snapshots", snap_id, patch_cfg(snap, tool_entries), "hw5s")
        AC.save_cfg(d, "agent_config_drafts", draft_id, patch_cfg(draft, tool_entries), "hw5d")

        write_ids(app_id, agent_id, snap_id, draft_id)
        print("\n已写入 _config.py：")
        print('HOMEWORK5_APP_ID = "%s"' % app_id)
        print('HOMEWORK5_AGENT_ID = "%s"' % agent_id)
        print('HOMEWORK5_SNAP_ID = "%s"' % snap_id)
        print('HOMEWORK5_DRAFT_ID = "%s"' % draft_id)
        print("\n[WARN] 改完不要点「构建/发布」；强刷后新开对话验收。")
        return 0
    finally:
        d.close()


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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""作业 5 · 步骤 5：问「检查一下 AI 平台的健康状态」，应调用三工具并出报告。"""
from __future__ import annotations

import sys

import _bootstrap  # noqa: F401
import _config as C
from dify_api import Dify, pp


def resolve_agent_id(d: Dify) -> str:
    aid = getattr(C, "HOMEWORK5_AGENT_ID", "") or ""
    if aid:
        return aid
    r_list = d.get("/agent?page=1&limit=50")
    for item in (r_list.get("json") or {}).get("data") or []:
        if item.get("name") == C.HOMEWORK5_APP_NAME:
            return item.get("agent_id") or item.get("id") or ""
    raise RuntimeError("找不到 Agent: %s" % C.HOMEWORK5_APP_NAME)


def main() -> int:
    d = Dify(C.DIFY_EMAIL, C.DIFY_PASSWORD)
    try:
        d.login()
        agent_id = resolve_agent_id(d)
        print("agent_id", agent_id)
        r = d.post(
            "/agent/%s/chat-messages" % agent_id,
            body={
                "inputs": {},
                "query": C.HOMEWORK5_PROBE_QUERY,
                "response_mode": "streaming",
                "files": [],
                "conversation_id": "",
            },
            extra="-m 480 --max-time 480",
            t=510,
        )
        pp("chat", r, limit=8000)
        text = r.get("text") or ""
        need = ("get_vllm_status", "get_gpu_info", "get_request_queue")
        hits = [m for m in need if m in text]
        extra = [m for m in ("H20", "vllm-qwen", "waiting", "running") if m in text]
        print("\n匹配到的工具名:", hits)
        print("匹配到的指标词:", extra)
        if r.get("code") not in ("200",) or len(hits) < 3:
            print("\n[FAIL] 未看到三个工具都被调用。")
            return 1
        if len(extra) < 2:
            print("\n[FAIL] 工具调用了，但回复里缺少 GPU/队列关键信息。")
            return 1
        print("\n[PASS] Agent 已调用三工具并输出健康报告。")
        return 0
    finally:
        d.close()


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