99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
"""
|
|||
|
|
场景模板库：从现有知你 Agent 复制出可复用的「模板 Agent」（草稿），便于主控台/运营选用。
|
|||
|
|
|
|||
|
|
默认创建三个模板（已存在则跳过）：
|
|||
|
|
- 模板-客服标准（来自 知你客服14号）
|
|||
|
|
- 模板-研发多步（来自 知你客服15号）
|
|||
|
|
- 模板-Loop多段（来自 知你客服16号）
|
|||
|
|
|
|||
|
|
用法:
|
|||
|
|
cd backend && .\\venv\\Scripts\\python.exe scripts/bootstrap_scene_templates.py
|
|||
|
|
|
|||
|
|
环境变量: PLATFORM_BASE_URL, PLATFORM_USERNAME, PLATFORM_PASSWORD
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|||
|
|
|
|||
|
|
import requests
|
|||
|
|
|
|||
|
|
BASE = os.getenv("PLATFORM_BASE_URL", "http://127.0.0.1:8037").rstrip("/")
|
|||
|
|
USER = os.getenv("PLATFORM_USERNAME", "admin")
|
|||
|
|
PWD = os.getenv("PLATFORM_PASSWORD", "123456")
|
|||
|
|
|
|||
|
|
TEMPLATES: List[Tuple[str, str, str]] = [
|
|||
|
|
("模板-客服标准", "知你客服14号", "场景模板：标准客服 + 全量工具基线"),
|
|||
|
|
("模板-研发多步", "知你客服15号", "场景模板：单节点多轮工具 / 可持续执行"),
|
|||
|
|
("模板-Loop多段", "知你客服16号", "场景模板：Loop 多段 LLM + 合并输出"),
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _login_headers() -> Dict[str, str]:
|
|||
|
|
r = requests.post(
|
|||
|
|
f"{BASE}/api/v1/auth/login",
|
|||
|
|
data={"username": USER, "password": PWD},
|
|||
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|||
|
|
timeout=15,
|
|||
|
|
)
|
|||
|
|
r.raise_for_status()
|
|||
|
|
token = r.json().get("access_token")
|
|||
|
|
if not token:
|
|||
|
|
raise RuntimeError("无 access_token")
|
|||
|
|
return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _find_agent_id(h: Dict[str, str], name: str) -> Optional[str]:
|
|||
|
|
r = requests.get(f"{BASE}/api/v1/agents", params={"search": name, "limit": 100}, headers=h, timeout=30)
|
|||
|
|
if r.status_code != 200:
|
|||
|
|
return None
|
|||
|
|
for a in r.json() or []:
|
|||
|
|
if a.get("name") == name:
|
|||
|
|
return a.get("id")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main() -> int:
|
|||
|
|
h = _login_headers()
|
|||
|
|
created: List[Dict[str, Any]] = []
|
|||
|
|
skipped: List[str] = []
|
|||
|
|
|
|||
|
|
for tpl_name, source_name, tpl_desc in TEMPLATES:
|
|||
|
|
if _find_agent_id(h, tpl_name):
|
|||
|
|
skipped.append(tpl_name)
|
|||
|
|
continue
|
|||
|
|
src_id = _find_agent_id(h, source_name)
|
|||
|
|
if not src_id:
|
|||
|
|
print(f"跳过 {tpl_name}：未找到源 Agent {source_name}", file=sys.stderr)
|
|||
|
|
continue
|
|||
|
|
dup = requests.post(
|
|||
|
|
f"{BASE}/api/v1/agents/{src_id}/duplicate",
|
|||
|
|
headers=h,
|
|||
|
|
json={"name": tpl_name},
|
|||
|
|
timeout=120,
|
|||
|
|
)
|
|||
|
|
if dup.status_code != 201:
|
|||
|
|
print(f"复制失败 {tpl_name}:", dup.status_code, dup.text[:500], file=sys.stderr)
|
|||
|
|
continue
|
|||
|
|
new_id = dup.json()["id"]
|
|||
|
|
up = requests.put(
|
|||
|
|
f"{BASE}/api/v1/agents/{new_id}",
|
|||
|
|
headers=h,
|
|||
|
|
json={"description": tpl_desc},
|
|||
|
|
timeout=60,
|
|||
|
|
)
|
|||
|
|
if up.status_code != 200:
|
|||
|
|
print(f"更新描述失败 {tpl_name}:", up.text[:400], file=sys.stderr)
|
|||
|
|
created.append({"id": new_id, "name": tpl_name, "source": source_name})
|
|||
|
|
|
|||
|
|
print("已跳过(已存在):", ", ".join(skipped) if skipped else "(无)")
|
|||
|
|
print("新建:", json.dumps(created, ensure_ascii=False))
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
raise SystemExit(main())
|