- supervisor, memory, skills 模块 - LLM 工厂 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
47 lines
1.2 KiB
Python
47 lines
1.2 KiB
Python
"""
|
|
Skill Router - 技能路由器
|
|
"""
|
|
from typing import List, Dict
|
|
|
|
|
|
class SkillRouter:
|
|
"""根据 LLM 决策选择要调用的技能"""
|
|
|
|
def __init__(self, available_skills: List[int]):
|
|
"""
|
|
初始化技能路由器
|
|
|
|
Args:
|
|
available_skills: 可用技能 ID 列表
|
|
"""
|
|
self.available_skills = available_skills
|
|
|
|
async def route(self, llm_decision: dict) -> List[dict]:
|
|
"""
|
|
解析 LLM 的技能调用决策
|
|
|
|
Args:
|
|
llm_decision: LLM 决策结果
|
|
|
|
Returns:
|
|
List[dict]: 要执行的技能列表
|
|
"""
|
|
if not llm_decision.get('tool_calls'):
|
|
return []
|
|
|
|
routes = []
|
|
for tool_call in llm_decision['tool_calls']:
|
|
skill_id = tool_call.get('skill_id') or tool_call.get('tool_name')
|
|
|
|
# 检查技能是否可用
|
|
if self.available_skills and skill_id not in self.available_skills:
|
|
continue
|
|
|
|
routes.append({
|
|
'skill_id': skill_id,
|
|
'params': tool_call.get('parameters', {}),
|
|
'reason': tool_call.get('reason', '')
|
|
})
|
|
|
|
return routes
|