from typing import Dict, Set, Union, Optional
import json
from .intent_classifier import LLMIntentClassifier
from .semantic_router_components import AgentRegistryBase
from .intent_registry import IntentRegistry
from ..zerocode.developer_agent import DeveloperAgent, FlowAgentCreator, PromptAgentCreator
from agentopera.ui.vercel_ai_request import UserAgentInfo
# TODO: we should wrap this up with an orchestrator
def create_agent_from_info(agent: UserAgentInfo) -> DeveloperAgent:
if agent.source == 3:
return FlowAgentCreator(
agent_url=agent.agent_url,
is_output_streaming=agent.is_output_streaming,
agent_name=agent.agent_name,
agent_intent=agent.agent_intent,
agent_description=agent.agent_description,
agent_type=agent.agent_type,
is_session_supported=agent.is_session_supported,
required_param={k: "string" for k in agent.required_param}, # optional default schema typing
agent_api_key=agent.agent_api_key,
required_payload_structure=json.dumps(agent.required_payload_structure) if agent.required_payload_structure else None
)
elif agent.source == 4:
return PromptAgentCreator(
agent_url=agent.agent_url,
is_output_streaming=agent.is_output_streaming,
agent_name=agent.agent_name,
agent_intent=agent.agent_intent,
agent_description=agent.agent_description,
agent_type=agent.agent_type,
system_prompt=agent.system_prompt,
model_name=agent.model_name,
agent_api_key=agent.agent_api_key,
)
[docs]
class AgentFactory:
[docs]
@classmethod
def create_agent_registry(cls, intent_registry: IntentRegistry) -> AgentRegistryBase:
"""
Creates an AgentRegistry from an IntentRegistry.
Args:
intent_registry (IntentRegistry): The shared config source.
Returns:
AgentRegistryBase: A runtime-compatible registry.
"""
return AgentRegistry(intent_registry)
[docs]
@classmethod
def create_intent_classifier(cls, intent_registry: IntentRegistry) -> LLMIntentClassifier:
"""
Creates an LLM-powered intent classifier using the provided intent registry.
Args:
intent_registry (IntentRegistry): The shared config source.
Returns:
LLMIntentClassifier: Configured classifier using OpenAI-compatible tools.
"""
return LLMIntentClassifier(intent_registry)
[docs]
@classmethod
def create_intent_registry(
cls,
intent_descriptions: dict[str, str],
agent_intent_mapping: dict[str, str],
agent_descriptions: dict[str, str],
schema_overrides: Optional[Dict[str, dict]] = None
) -> IntentRegistry:
return IntentRegistry(intent_descriptions, agent_intent_mapping, agent_descriptions, schema_overrides)
[docs]
class AgentRegistry(AgentRegistryBase):
"""
Runtime-compatible agent registry adapter.
Wraps around `IntentRegistry` to expose methods required by
SemanticRouterAgent and other runtime components.
Responsibilities:
- Asynchronously fetch agents based on classified intent
- Serve agent metadata for monitoring and validation
"""
def __init__(self, intent_registry: IntentRegistry):
"""
Args:
intent_registry: Instance of `IntentRegistry` holding intent-agent mappings and descriptions.
"""
self.intent_registry = intent_registry
[docs]
async def get_agent(self, intent: str) -> str:
"""
Retrieves the agent type corresponding to a given intent.
Falls back to 'chat' if intent is not mapped.
Args:
intent: The intent for which to retrieve an agent.
Returns:
The agent type (e.g., 'chat', 'vercel', etc.)
"""
return self.intent_registry.get_agent_for_intent(intent)
[docs]
async def get_agents(self) -> Set[str]:
"""
Retrieves the set of all registered agent IDs.
Returns:
A set of agent IDs.
"""
return self.intent_registry.get_all_agents()
[docs]
async def get_agent_description(self, agent_id: str) -> str:
"""
Retrieves the description of a specific agent by its ID.
Args:
agent_id: The agent ID for which to retrieve the description.
Returns:
The agent's description.
"""
return self.intent_registry.get_agent_description(agent_id)
[docs]
async def get_agent_descriptions(self) -> Dict[str, Union[str, list[str]]]:
"""
Retrieves descriptions of all registered agents.
Returns:
A dictionary mapping agent IDs to their descriptions.
"""
return self.intent_registry.agent_descriptions