Explorar el Código

feat(scripts): install_builtin_packs.py — 一键安装内置 research.* pack

支持离线模式(直接写入 agentpacks_dir,推荐)和在线模式(需 server 运行)。
自动跳过已安装的 pack。

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
kenny67nju hace 3 meses
padre
commit
432306c445
Se han modificado 1 ficheros con 186 adiciones y 0 borrados
  1. 186 0
      scripts/install_builtin_packs.py

+ 186 - 0
scripts/install_builtin_packs.py

@@ -0,0 +1,186 @@
+#!/usr/bin/env python3
+"""
+安装仓库内置的 AgentPack 到本地 agentpacks 目录。
+
+用法(server 必须先运行):
+    python scripts/install_builtin_packs.py
+
+也可以不启动 server,直接写入 agentpacks_dir:
+    python scripts/install_builtin_packs.py --offline
+
+内置 pack 位置:agentexample/agentpacks/
+安装目标:    ~/.agentpaas/data/agentpacks/   (desktop 模式)
+              ~/LambdAgentDesktop/agentpacks/  (如果是 desktop 模式且未改数据目录)
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import zipfile
+import tempfile
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).parent.parent.resolve()
+PACKS_SRC  = REPO_ROOT / "agentexample" / "agentpacks"
+
+BUILTIN_PACKS = [
+    "research.top-journal-reviewer",
+    "research.literature-mapper",
+    "research.grant-planner",
+]
+
+
+def _zip_pack(pack_id: str, dest_dir: Path) -> str:
+    pack_dir = PACKS_SRC / pack_id
+    if not pack_dir.is_dir():
+        sys.exit(f"❌  Pack 目录不存在: {pack_dir}")
+    zip_path = str(dest_dir / f"{pack_id}.zip")
+    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
+        for p in pack_dir.rglob("*"):
+            if p.is_file():
+                zf.write(p, p.relative_to(pack_dir))
+    return zip_path
+
+
+def _default_packs_dir() -> str:
+    """读取 ~/.agentpaas/config.json 或按 desktop 模式默认值推断 agentpacks_dir。"""
+    cfg = Path.home() / ".agentpaas" / "config.json"
+    try:
+        with open(cfg) as f:
+            data = json.load(f) or {}
+        data_dir = data.get("data_dir", "")
+    except Exception:
+        data_dir = ""
+    if not data_dir:
+        # desktop 默认
+        data_dir = str(Path.home() / "LambdAgentDesktop")
+    return os.path.join(data_dir, "agentpacks")
+
+
+# ── 离线模式:直接调用 install_from_zip ──────────────────────────────────────
+
+def install_offline(packs_dir: str, packs: list[str]) -> None:
+    from agentpaas.engine.agentpack_store import install_from_zip, get_installed
+
+    os.makedirs(packs_dir, exist_ok=True)
+    print(f"\n安装目标目录: {packs_dir}\n")
+
+    with tempfile.TemporaryDirectory() as td:
+        for pack_id in packs:
+            existing = get_installed(packs_dir, pack_id)
+            if existing:
+                print(f"  ✓  {pack_id} v{existing.version}  — 已安装,跳过")
+                continue
+            zip_path = _zip_pack(pack_id, Path(td))
+            try:
+                pack = install_from_zip(zip_path, packs_dir)
+                print(f"  ✓  {pack_id} v{pack.version}  — 安装完成")
+            except Exception as e:
+                print(f"  ✗  {pack_id}  — 失败: {e}", file=sys.stderr)
+
+
+# ── 在线模式:通过 HTTP API 安装(需要 server 运行)───────────────────────────
+
+def install_online(base_url: str, api_key: str, packs: list[str]) -> None:
+    try:
+        import requests
+    except ImportError:
+        sys.exit("缺少依赖: pip install requests")
+
+    session = requests.Session()
+    if api_key:
+        session.headers["Authorization"] = f"Bearer {api_key}"
+
+    print(f"\n连接 server: {base_url}\n")
+
+    # 先列出已安装
+    try:
+        r = session.get(f"{base_url}/api/v1/agentpacks", timeout=5)
+        r.raise_for_status()
+        already = {p["id"] for p in r.json().get("agentpacks", [])}
+    except Exception as e:
+        sys.exit(f"❌  无法连接 server: {e}\n    请先启动: uvicorn agentpaas.api.app:app --port 8000")
+
+    with tempfile.TemporaryDirectory() as td:
+        for pack_id in packs:
+            if pack_id in already:
+                print(f"  ✓  {pack_id}  — 已安装,跳过")
+                continue
+            zip_path = _zip_pack(pack_id, Path(td))
+            r = session.post(
+                f"{base_url}/api/v1/agentpacks/install",
+                json={"zip_path": zip_path},
+                timeout=30,
+            )
+            if r.status_code == 200:
+                pack = r.json()["installed"]
+                print(f"  ✓  {pack['id']} v{pack['version']}  — 安装完成")
+            else:
+                print(f"  ✗  {pack_id}  — HTTP {r.status_code}: {r.text[:120]}", file=sys.stderr)
+
+
+# ── CLI ───────────────────────────────────────────────────────────────────────
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="安装仓库内置 AgentPack (research.* 三个 pack)",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+示例:
+  # 离线安装(不需要 server 运行,推荐)
+  python scripts/install_builtin_packs.py --offline
+
+  # 在线安装(需要 server 先跑起来)
+  python scripts/install_builtin_packs.py
+
+  # 指定安装目录
+  python scripts/install_builtin_packs.py --offline --packs-dir /data/agentpacks
+""",
+    )
+    parser.add_argument(
+        "--offline", action="store_true",
+        help="直接写入 agentpacks_dir,不需要 server 运行(推荐)",
+    )
+    parser.add_argument(
+        "--packs-dir", default="",
+        help="安装目标目录(默认:从 ~/.agentpaas/config.json 推断)",
+    )
+    parser.add_argument(
+        "--url", default="http://127.0.0.1:8000",
+        help="Server URL(在线模式,默认 http://127.0.0.1:8000)",
+    )
+    parser.add_argument(
+        "--key", default="",
+        help="API key(在线模式,默认从 ~/.agentpaas/config.json 读取)",
+    )
+    parser.add_argument(
+        "--pack", action="append", dest="packs",
+        help="只安装指定 pack(可多次指定,默认全部三个)",
+    )
+    args = parser.parse_args()
+
+    packs = args.packs or BUILTIN_PACKS
+    print(f"内置 Pack 源目录: {PACKS_SRC}")
+
+    if args.offline:
+        packs_dir = args.packs_dir or _default_packs_dir()
+        install_offline(packs_dir, packs)
+    else:
+        # 读取 api_key
+        api_key = args.key
+        if not api_key:
+            cfg = Path.home() / ".agentpaas" / "config.json"
+            try:
+                api_key = (json.load(open(cfg)) or {}).get("api_key", "")
+            except Exception:
+                pass
+        install_online(args.url, api_key, packs)
+
+    print('\n完成。在 webui 的"智能体包"页面可以看到已安装的 pack,点击"创建智能体"即可使用。\n')
+
+
+if __name__ == "__main__":
+    main()