Pine
82 lines
3.0 KiB
Python
82 lines
3.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, get_db
|
|
from ...infrastructure.repositories import Database
|
|
|
|
router = APIRouter(prefix="/agent-templates", tags=["agent-templates"])
|
|
|
|
# 模板类型 → 目录名映射（"common" 即 TEMPLATES_DIR 根下的语言目录）。
|
|
_TEMPLATE_TYPES = ("common", "qa", "local")
|
|
|
|
# 全局官方注入的最高优先级提示词（fixed_soul）→ system_configs 存储键。
|
|
FIXED_SOUL_CONFIG_PREFIX = "agent.fixed_soul"
|
|
|
|
|
|
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),
|
|
db: Database = Depends(get_db),
|
|
):
|
|
"""返回全部初始化模板：
|
|
|
|
``{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"] = await _fixed_soul(db, base)
|
|
return tree
|
|
|
|
|
|
async def _fixed_soul(db: Database, base: Path) -> dict[str, str]:
|
|
"""读取全局 fixed_soul：优先管理平台配置（system_configs），回退模板文件。
|
|
|
|
管理后台可配置 ``agent.fixed_soul.zh`` / ``agent.fixed_soul.en`` 键。
|
|
"""
|
|
files: dict[str, str] = {}
|
|
root = base / "fixed_soul"
|
|
if root.is_dir():
|
|
for md in sorted(root.glob("*.md")):
|
|
files[md.stem] = md.read_text(encoding="utf-8")
|
|
# 平台配置覆盖（仅对已配置的语言生效）。
|
|
for lang, default in files.items():
|
|
cfg = await db.config.get(f"{FIXED_SOUL_CONFIG_PREFIX}.{lang}")
|
|
if cfg:
|
|
files[lang] = cfg
|
|
# 平台新增配置的语言（模板文件缺失时也下发）。
|
|
for lang in ("zh", "en"):
|
|
cfg = await db.config.get(f"{FIXED_SOUL_CONFIG_PREFIX}.{lang}")
|
|
if cfg and lang not in files:
|
|
files[lang] = cfg
|
|
return files
|