Spaces:
Running
Running
File size: 2,398 Bytes
016b413 |
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 |
"""Factory for creating orchestrators."""
from typing import Any, Literal
import structlog
from src.legacy_orchestrator import (
JudgeHandlerProtocol,
Orchestrator,
SearchHandlerProtocol,
)
from src.utils.config import settings
from src.utils.models import OrchestratorConfig
logger = structlog.get_logger()
def _get_magentic_orchestrator_class() -> Any:
"""Import MagenticOrchestrator lazily to avoid hard dependency."""
try:
from src.orchestrator_magentic import MagenticOrchestrator
return MagenticOrchestrator
except ImportError as e:
logger.error("Failed to import MagenticOrchestrator", error=str(e))
raise ValueError(
"Advanced mode requires agent-framework-core. Please install it or use mode='simple'."
) from e
def create_orchestrator(
search_handler: SearchHandlerProtocol | None = None,
judge_handler: JudgeHandlerProtocol | None = None,
config: OrchestratorConfig | None = None,
mode: Literal["simple", "magentic", "advanced"] | None = None,
) -> Any:
"""
Create an orchestrator instance.
Args:
search_handler: The search handler (required for simple mode)
judge_handler: The judge handler (required for simple mode)
config: Optional configuration
mode: "simple", "magentic", "advanced" or None (auto-detect)
Returns:
Orchestrator instance
"""
effective_mode = _determine_mode(mode)
logger.info("Creating orchestrator", mode=effective_mode)
if effective_mode == "advanced":
orchestrator_cls = _get_magentic_orchestrator_class()
return orchestrator_cls(
max_rounds=config.max_iterations if config else 10,
)
# Simple mode requires handlers
if search_handler is None or judge_handler is None:
raise ValueError("Simple mode requires search_handler and judge_handler")
return Orchestrator(
search_handler=search_handler,
judge_handler=judge_handler,
config=config,
)
def _determine_mode(explicit_mode: str | None) -> str:
"""Determine which mode to use."""
if explicit_mode:
if explicit_mode in ("magentic", "advanced"):
return "advanced"
return "simple"
# Auto-detect: advanced if paid API key available
if settings.has_openai_key:
return "advanced"
return "simple"
|