Pine
980a2db6d9
- 基础设施层 async：SQLAlchemy 异步引擎/会话、33 Repository async 化、models/security/seed 迁入 infrastructure、新增 cache.py(redis.asyncio) 与 oss.py(aioboto3) - 接口层：routers 迁 api/routers 并全 async，dependencies 迁 api/dependencies（get_db/get_current_user async） - 依赖：sqlalchemy[asyncio]/aiosqlite/asyncmy/redis/aioboto3；config 异步 URL + Redis/OSS 配置 - 删除废弃：旧同步 db/dependencies/repositories/storage - 验证：平台 19 路由 + 培训 48 路由全注册；/health /auth/login /auth/me /admin/tasks /notifications 等接口 async 可用
63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""智能体初始化模板路由。
|
|
|
|
服务端作为智能体初始化 md 模板的唯一来源（qa 模板已统一为 PineAgents 品牌），
|
|
本地在初始化前把整棵树拉到本地缓存。模板以目录 md 存储、经接口暴露，后续可换 JSON/DB。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Depends
|
|
|
|
from ... import config
|
|
from ..dependencies import get_current_user
|
|
|
|
router = APIRouter(prefix="/agent-templates", tags=["agent-templates"])
|
|
|
|
# 模板类型 → 目录名映射（"common" 即 TEMPLATES_DIR 根下的语言目录）。
|
|
_TEMPLATE_TYPES = ("common", "qa", "local")
|
|
|
|
|
|
def _read_type_dir(base: Path, type_name: str) -> dict[str, dict[str, str]]:
|
|
root = base if type_name == "common" else base / type_name
|
|
langs: dict[str, dict[str, str]] = {}
|
|
if not root.is_dir():
|
|
return langs
|
|
for lang_dir in sorted(p for p in root.iterdir() if p.is_dir()):
|
|
files = {
|
|
md.name: md.read_text(encoding="utf-8")
|
|
for md in sorted(lang_dir.glob("*.md"))
|
|
}
|
|
if files:
|
|
langs[lang_dir.name] = files
|
|
return langs
|
|
|
|
|
|
@router.get("", summary="智能体初始化模板树", response_model=dict)
|
|
async def get_agent_templates(
|
|
_user: dict = Depends(get_current_user),
|
|
):
|
|
"""返回全部初始化模板：
|
|
|
|
``{template_type: {language: {filename: content}}}`` + 顶层 ``fixed_soul``
|
|
（服务端托管、不可变、最高优先级的 SOUL 片段）。
|
|
"""
|
|
base = config.TEMPLATES_DIR
|
|
tree = {
|
|
type_name: _read_type_dir(base, type_name)
|
|
for type_name in _TEMPLATE_TYPES
|
|
}
|
|
tree["fixed_soul"] = _read_lang_files(base / "fixed_soul")
|
|
return tree
|
|
|
|
|
|
def _read_lang_files(root: Path) -> dict[str, str]:
|
|
"""读取 ``<root>/<lang>.md`` → ``{lang: content}``。"""
|
|
files: dict[str, str] = {}
|
|
if not root.is_dir():
|
|
return files
|
|
for md in sorted(root.glob("*.md")):
|
|
files[md.stem] = md.read_text(encoding="utf-8")
|
|
return files
|