#!/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") agent_id = "" for item in (r_list.get("json") or {}).get("data") or []: if item.get("name") == C.HOMEWORK5_APP_NAME: 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 " "FROM agents a 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]: raise RuntimeError("找不到 Agent 快照: %s %s" % (out, err)) app_id, agent_id, snap_id = line[0].split("|") draft_id = ensure_draft(d, agent_id, snap_id) return app_id, agent_id, snap_id, draft_id def ensure_draft(d: Dify, agent_id: str, snap_id: str) -> str: sql = ( "SELECT id FROM agent_config_drafts WHERE agent_id='%s' LIMIT 1;" % agent_id ) out, err = d.sh( 'docker exec %s psql -U postgres -d dify -t -A -c "%s"' % (C.PG_CONTAINER, sql) ) draft_id = (out or "").strip().splitlines() draft_id = draft_id[0].strip() if draft_id else "" if draft_id: return draft_id ins = ( "INSERT INTO agent_config_drafts (" "id, tenant_id, agent_id, draft_type, base_snapshot_id, config_snapshot, " "created_by, updated_by, created_at, updated_at" ") SELECT gen_random_uuid(), s.tenant_id, s.agent_id, 'draft', s.id, " "s.config_snapshot, s.created_by, s.created_by, NOW(), NOW() " "FROM agent_config_snapshots s WHERE s.id='%s' " "RETURNING id;" % snap_id ) remote = "/tmp/hw5_ins_draft.sql" with d.sftp.open(remote, "w") as f: f.write(ins) out, err = d.sh( "docker exec -i %s psql -U postgres -d dify -t -A < %s" % (C.PG_CONTAINER, remote) ) draft_id = (out or "").strip().splitlines() draft_id = draft_id[0].strip() if draft_id else "" if not draft_id: raise RuntimeError("无法创建草稿: %s %s" % (out, err)) print("created draft", draft_id) return 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())