46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
"""
|
||
Skill Registry - Agent 运行时加载 Skills
|
||
"""
|
||
from typing import Optional
|
||
from app.database import async_session
|
||
from app.services.skill_service import SkillService
|
||
|
||
# 缓存：agent_type -> list[Skill]
|
||
_skill_cache: dict[str, list] = {}
|
||
|
||
async def load_skills_for_agent(agent_type: str, force_reload: bool = False) -> list:
|
||
"""加载指定 Agent 类型的可用 Skills"""
|
||
if not force_reload and agent_type in _skill_cache:
|
||
return _skill_cache[agent_type]
|
||
async with async_session() as db:
|
||
svc = SkillService(db)
|
||
skills = await svc.get_by_agent_type(agent_type)
|
||
_skill_cache[agent_type] = skills
|
||
return skills
|
||
|
||
def get_skills_for_agent(agent_type: str) -> list:
|
||
"""同步接口：返回缓存的 Skills（供 Agent 节点调用）"""
|
||
return _skill_cache.get(agent_type, [])
|
||
|
||
def build_skill_context(agent_type: str) -> str:
|
||
"""
|
||
构建 Skill 上下文，供注入到 Agent 系统提示
|
||
格式：Skill 名称 + 描述 + 工具列表
|
||
"""
|
||
skills = get_skills_for_agent(agent_type)
|
||
if not skills:
|
||
return ""
|
||
lines = ["\n\n【可用的 Skills】"]
|
||
for s in skills:
|
||
tools_str = ", ".join(s.tools) if s.tools else "无"
|
||
lines.append(f"""
|
||
## {s.name}
|
||
- 描述: {s.description or '无'}
|
||
- 工具: {tools_str}
|
||
- 指令: {s.instructions[:200]}...""" if len(s.instructions) > 200 else f"- 指令: {s.instructions}")
|
||
return "\n".join(lines)
|
||
|
||
def clear_cache():
|
||
"""清除缓存（配置变更时调用）"""
|
||
global _skill_cache
|
||
_skill_cache = {} |