File size: 4,562 Bytes
e82d9c9 11888fc e82d9c9 11888fc e82d9c9 11888fc e82d9c9 11888fc e82d9c9 11888fc e82d9c9 11888fc e82d9c9 e0c585c e82d9c9 11888fc e82d9c9 11888fc e0c585c e82d9c9 11888fc e82d9c9 11888fc e82d9c9 11888fc e82d9c9 11888fc e82d9c9 11888fc e82d9c9 e0c585c e82d9c9 |
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 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 |
"""Factory for creating orchestrators.
Implements the Factory Pattern (GoF) for creating the appropriate
orchestrator based on configuration and available credentials.
Design Principles:
- Open/Closed: Easy to add new orchestrator types without modifying existing code
- Dependency Inversion: Returns protocol-compatible objects, not concrete types
- Single Responsibility: Only handles orchestrator creation logic
"""
from typing import TYPE_CHECKING, Literal
import structlog
from src.orchestrators.base import (
JudgeHandlerProtocol,
OrchestratorProtocol,
SearchHandlerProtocol,
)
from src.orchestrators.simple import Orchestrator
from src.utils.config import settings
from src.utils.models import OrchestratorConfig
if TYPE_CHECKING:
from src.orchestrators.advanced import AdvancedOrchestrator
logger = structlog.get_logger()
def _get_advanced_orchestrator_class() -> type["AdvancedOrchestrator"]:
"""Import AdvancedOrchestrator lazily to avoid hard dependency.
This allows the simple mode to work without agent-framework-core installed.
Returns:
The AdvancedOrchestrator class
Raises:
ValueError: If agent-framework-core is not installed
"""
try:
from src.orchestrators.advanced import AdvancedOrchestrator
return AdvancedOrchestrator
except ImportError as e:
logger.error("Failed to import AdvancedOrchestrator", error=str(e))
raise ValueError(
"Advanced mode requires agent-framework-core. "
"Install with: pip install agent-framework-core. "
"Or use mode='simple' instead."
) from e
def create_orchestrator(
search_handler: SearchHandlerProtocol | None = None,
judge_handler: JudgeHandlerProtocol | None = None,
config: OrchestratorConfig | None = None,
mode: Literal["simple", "magentic", "advanced", "hierarchical"] | None = None,
api_key: str | None = None,
) -> OrchestratorProtocol:
"""
Create an orchestrator instance.
This factory automatically selects the appropriate orchestrator based on:
1. Explicit mode parameter (if provided)
2. Available API keys (auto-detection)
Args:
search_handler: The search handler (required for simple mode)
judge_handler: The judge handler (required for simple mode)
config: Optional configuration (max_iterations, timeouts, etc.)
mode: "simple", "magentic", "advanced", or "hierarchical"
Note: "magentic" is an alias for "advanced" (kept for backwards compatibility)
api_key: Optional API key for advanced mode (OpenAI)
Returns:
Orchestrator instance implementing OrchestratorProtocol
Raises:
ValueError: If required handlers are missing for simple mode
ValueError: If advanced mode is requested but dependencies are missing
"""
effective_config = config or OrchestratorConfig()
effective_mode = _determine_mode(mode, api_key)
logger.info("Creating orchestrator", mode=effective_mode)
if effective_mode == "advanced":
orchestrator_cls = _get_advanced_orchestrator_class()
return orchestrator_cls(
max_rounds=effective_config.max_iterations,
api_key=api_key,
)
if effective_mode == "hierarchical":
from src.orchestrators.hierarchical import HierarchicalOrchestrator
return HierarchicalOrchestrator(config=effective_config)
# 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=effective_config,
)
def _determine_mode(explicit_mode: str | None, api_key: str | None) -> str:
"""Determine which mode to use.
Priority:
1. Explicit mode parameter
2. Auto-detect based on available API keys
Args:
explicit_mode: Mode explicitly requested by caller
api_key: API key provided by caller
Returns:
Effective mode string: "simple", "advanced", or "hierarchical"
"""
if explicit_mode:
if explicit_mode in ("magentic", "advanced"):
return "advanced"
if explicit_mode == "hierarchical":
return "hierarchical"
return "simple"
# Auto-detect: advanced if paid API key available
if settings.has_openai_key or (api_key and api_key.startswith("sk-")):
return "advanced"
return "simple"
|