File size: 5,080 Bytes
6bcddd0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | import argparse
import json
import os
import sys
from pathlib import Path
from urllib.parse import urlparse
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
def require_modal():
try:
import modal # noqa: F401
except Exception as exc:
raise SystemExit(
"Modal Python package is not installed. Install with `python -m pip install -r requirements.txt`, "
"then run `modal setup` to log in."
) from exc
def check_remote_methods(service: str = "all") -> int:
require_modal()
from modal_apps.modal_image import CharacterImage, app as image_app
from modal_apps.modal_llm import PersonaLLM, app as llm_app
from modal_apps.modal_tts import CharacterTTS, app as tts_app
all_checks = [
("llm", llm_app, lambda: PersonaLLM().health.remote()),
("tts", tts_app, lambda: CharacterTTS().health.remote()),
("image", image_app, lambda: CharacterImage().health.remote()),
]
checks = [check for check in all_checks if service in {"all", check[0]}]
ok = True
for name, modal_app, fn in checks:
try:
print(f"[{name}] checking health...")
with modal_app.run():
result = fn()
print(json.dumps(result, ensure_ascii=False, indent=2))
except Exception as exc:
ok = False
print(f"[{name}] failed: {exc}", file=sys.stderr)
return 0 if ok else 1
def check_deployed_endpoints() -> int:
try:
import httpx
except Exception as exc:
raise SystemExit("httpx is not installed. Run `python -m pip install -r requirements.txt`.") from exc
llm_url = os.environ.get("VC_MODAL_LLM_URL")
tts_url = os.environ.get("VC_MODAL_TTS_URL")
image_url = os.environ.get("VC_MODAL_IMAGE_URL")
ok = True
if llm_url:
print("[llm:endpoint] checking SSE...")
with httpx.stream(
"POST",
llm_url,
json={"text": "你好，请简单回复一句。", "character": {"display_name": "星萤"}, "max_new_tokens": 40},
timeout=120,
) as response:
response.raise_for_status()
seen = 0
for line in response.iter_lines():
if line:
print(line)
seen += 1
if seen >= 5:
break
else:
print("[llm:endpoint] skipped; set VC_MODAL_LLM_URL")
if tts_url:
print("[tts:endpoint] checking audio...")
response = httpx.post(
_tts_endpoint_url(tts_url),
json={"text": "你好，我在听。", "voice_id": "default", "emotion": "neutral"},
timeout=120,
trust_env=False,
)
response.raise_for_status()
out = Path("modal_tts_check.wav")
out.write_bytes(response.content)
print(f"wrote {out} ({len(response.content)} bytes)")
else:
print("[tts:endpoint] skipped; set VC_MODAL_TTS_URL")
if image_url:
print("[image:endpoint] checking png...")
response = httpx.post(
image_url,
json={"prompt": "original anime virtual character portrait, gentle sci-fi style", "steps": 4, "seed": 7},
timeout=180,
)
response.raise_for_status()
out = Path("modal_image_check.png")
out.write_bytes(response.content)
print(f"wrote {out} ({len(response.content)} bytes)")
else:
print("[image:endpoint] skipped; set VC_MODAL_IMAGE_URL")
return 0 if ok else 1
def _health_targets(url: str) -> list[str]:
base = url.rstrip("/")
tail = base.rsplit("/", 1)[-1]
service_base = base.rsplit("/", 1)[0] if tail in {"tts", "persona_events"} else base
return list(dict.fromkeys([service_base + "/health", service_base + "/health_http", base + "/health"]))
def _tts_endpoint_url(url: str) -> str:
base = url.rstrip("/")
parsed = urlparse(base)
if not parsed.path or parsed.path == "/":
return base
if parsed.path.rstrip("/").rsplit("/", 1)[-1] == "tts":
return base
return base + "/tts"
def main() -> int:
parser = argparse.ArgumentParser(description="Check Modal model connectivity for Virtual Characters.")
parser.add_argument(
"--mode",
choices=["remote-methods", "endpoints"],
default="remote-methods",
help="remote-methods checks Modal class methods; endpoints checks deployed web URLs.",
)
parser.add_argument(
"--service",
choices=["all", "llm", "tts", "image"],
default="all",
help="Limit remote-methods checks to one service.",
)
args = parser.parse_args()
if args.mode == "remote-methods":
return check_remote_methods(args.service)
if args.service != "all":
raise SystemExit("--service is only supported with --mode remote-methods")
return check_deployed_endpoints()
if __name__ == "__main__":
raise SystemExit(main())
|