0608161c82
补齐平台模板与场景 DSL、预算控制、执行看板和企业场景脚本，增强 Windows 启动/迁移与前端代理和聊天会话记忆，修复执行创建阶段 500 与异步链路排障体验。 Made-with: Cursor
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())
|