File size: 3,609 Bytes
8ede856 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | from __future__ import annotations
from typing import Any
from astrbot import logger
from astrbot.core.agent.agent import Agent
from astrbot.core.agent.handoff import HandoffTool
from astrbot.core.persona_mgr import PersonaManager
from astrbot.core.provider.func_tool_manager import FunctionToolManager
class SubAgentOrchestrator:
"""Loads subagent definitions from config and registers handoff tools.
This is intentionally lightweight: it does not execute agents itself.
Execution happens via HandoffTool in FunctionToolExecutor.
"""
def __init__(
self, tool_mgr: FunctionToolManager, persona_mgr: PersonaManager
) -> None:
self._tool_mgr = tool_mgr
self._persona_mgr = persona_mgr
self.handoffs: list[HandoffTool] = []
async def reload_from_config(self, cfg: dict[str, Any]) -> None:
from astrbot.core.astr_agent_context import AstrAgentContext
agents = cfg.get("agents", [])
if not isinstance(agents, list):
logger.warning("subagent_orchestrator.agents must be a list")
return
handoffs: list[HandoffTool] = []
for item in agents:
if not isinstance(item, dict):
continue
if not item.get("enabled", True):
continue
name = str(item.get("name", "")).strip()
if not name:
continue
persona_id = item.get("persona_id")
persona_data = None
if persona_id:
try:
persona_data = await self._persona_mgr.get_persona(persona_id)
except StopIteration:
logger.warning(
"SubAgent persona %s not found, fallback to inline prompt.",
persona_id,
)
instructions = str(item.get("system_prompt", "")).strip()
public_description = str(item.get("public_description", "")).strip()
provider_id = item.get("provider_id")
if provider_id is not None:
provider_id = str(provider_id).strip() or None
tools = item.get("tools", [])
begin_dialogs = None
if persona_data:
instructions = persona_data.system_prompt or instructions
begin_dialogs = persona_data.begin_dialogs
tools = persona_data.tools
if public_description == "" and persona_data.system_prompt:
public_description = persona_data.system_prompt[:120]
if tools is None:
tools = None
elif not isinstance(tools, list):
tools = []
else:
tools = [str(t).strip() for t in tools if str(t).strip()]
agent = Agent[AstrAgentContext](
name=name,
instructions=instructions,
tools=tools, # type: ignore
)
agent.begin_dialogs = begin_dialogs
# The tool description should be a short description for the main LLM,
# while the subagent system prompt can be longer/more specific.
handoff = HandoffTool(
agent=agent,
tool_description=public_description or None,
)
# Optional per-subagent chat provider override.
handoff.provider_id = provider_id
handoffs.append(handoff)
for handoff in handoffs:
logger.info(f"Registered subagent handoff tool: {handoff.name}")
self.handoffs = handoffs
|