Spaces:
Build error
Build error
Remove obsolete phase completion summaries and demo test scripts - Deleted `PHASE1_COMPLETION_SUMMARY.md`, `PHASE2_COMPLETION_SUMMARY.md`, `PHASE3_COMPLETION_SUMMARY.md`, and associated demo test scripts to streamline the codebase and eliminate unused documentation. This cleanup supports ongoing refactoring efforts and enhances overall project maintainability.
d5eabda | """ | |
| Workflow Orchestrator for Enhanced Recommendation Service | |
| Manages LangGraph workflow creation, routing, and execution. | |
| Extracted from EnhancedRecommendationService to improve modularity and maintainability. | |
| """ | |
| from typing import Dict, Any, Optional | |
| import structlog | |
| from langgraph.graph import StateGraph, END | |
| # Handle imports gracefully | |
| try: | |
| from ...models.agent_models import MusicRecommenderState | |
| from .agent_coordinator import AgentCoordinator | |
| except ImportError: | |
| # Fallback imports for testing | |
| import sys | |
| sys.path.append('src') | |
| from models.agent_models import MusicRecommenderState | |
| from services.components.agent_coordinator import AgentCoordinator | |
| logger = structlog.get_logger(__name__) | |
| class WorkflowOrchestrator: | |
| """ | |
| Orchestrates the LangGraph workflow for music recommendation. | |
| Responsibilities: | |
| - Building the workflow graph | |
| - Routing between agents | |
| - Executing workflow nodes | |
| - Managing workflow state transitions | |
| """ | |
| def __init__(self, agent_coordinator: AgentCoordinator): | |
| self.agent_coordinator = agent_coordinator | |
| self.logger = structlog.get_logger(__name__) | |
| self.graph: Optional[StateGraph] = None | |
| def build_workflow_graph(self) -> StateGraph: | |
| """Build the LangGraph workflow with conditional routing.""" | |
| workflow = StateGraph(MusicRecommenderState) | |
| # Add nodes | |
| workflow.add_node("planner", self._planner_node) | |
| workflow.add_node("genre_mood_advocate", self._genre_mood_node) | |
| workflow.add_node("discovery_advocate", self._discovery_node) | |
| workflow.add_node("judge", self._judge_node) | |
| # Set entry point | |
| workflow.set_entry_point("planner") | |
| # Intent-aware routing: Add conditional edges based on planner's agent sequence | |
| workflow.add_conditional_edges( | |
| "planner", | |
| self._route_agents, # Router function that respects intent-aware sequences | |
| { | |
| "discovery_only": "discovery_advocate", | |
| "genre_mood_only": "genre_mood_advocate", | |
| "both_agents": "discovery_advocate", # Start with discovery, then genre_mood | |
| "judge_only": "judge" # For edge cases | |
| } | |
| ) | |
| # Add conditional edges from discovery to either genre_mood or judge | |
| workflow.add_conditional_edges( | |
| "discovery_advocate", | |
| self._route_after_discovery, | |
| { | |
| "to_genre_mood": "genre_mood_advocate", | |
| "to_judge": "judge" | |
| } | |
| ) | |
| # Add edges from agents to judge | |
| workflow.add_edge("genre_mood_advocate", "judge") | |
| workflow.add_edge("judge", END) | |
| self.graph = workflow.compile() | |
| self.logger.info("Workflow graph built successfully") | |
| return self.graph | |
| def _route_agents(self, state: MusicRecommenderState) -> str: | |
| """ | |
| Route to the appropriate agents based on planner's strategy. | |
| Args: | |
| state: Current workflow state | |
| Returns: | |
| Next node to execute | |
| """ | |
| planning_strategy = getattr(state, 'planning_strategy', {}) | |
| agent_sequence = planning_strategy.get('agent_sequence', ['discovery', 'genre_mood']) | |
| self.logger.debug(f"Routing agents based on sequence: {agent_sequence}") | |
| if not agent_sequence: | |
| self.logger.warning("No agent sequence found, defaulting to judge_only") | |
| return "judge_only" | |
| # Determine routing based on agent sequence | |
| if len(agent_sequence) == 1: | |
| if agent_sequence[0] == 'discovery_agent': | |
| return "discovery_only" | |
| elif agent_sequence[0] == 'genre_mood_agent': | |
| return "genre_mood_only" | |
| else: | |
| return "judge_only" | |
| elif len(agent_sequence) >= 2: | |
| # Check if it's a genre_mood + judge sequence (no discovery) | |
| if agent_sequence == ['genre_mood_agent', 'judge_agent']: | |
| return "genre_mood_only" | |
| # Otherwise, use both agents | |
| return "both_agents" | |
| else: | |
| return "judge_only" | |
| def _route_after_discovery(self, state: MusicRecommenderState) -> str: | |
| """ | |
| Route after discovery agent execution. | |
| Args: | |
| state: Current workflow state | |
| Returns: | |
| Next node to execute | |
| """ | |
| planning_strategy = getattr(state, 'planning_strategy', {}) | |
| agent_sequence = planning_strategy.get('agent_sequence', ['discovery', 'genre_mood']) | |
| # If genre_mood_agent is in the sequence and we're coming from discovery, go to genre_mood | |
| if 'genre_mood_agent' in agent_sequence and len(agent_sequence) > 1: | |
| self.logger.debug("Routing from discovery to genre_mood") | |
| return "to_genre_mood" | |
| else: | |
| self.logger.debug("Routing from discovery directly to judge") | |
| return "to_judge" | |
| async def _planner_node(self, state: MusicRecommenderState) -> Dict[str, Any]: | |
| """ | |
| Execute the planner agent node. | |
| Args: | |
| state: Current workflow state | |
| Returns: | |
| Updated state dictionary | |
| """ | |
| try: | |
| planner_agent = self.agent_coordinator.get_planner_agent() | |
| if not planner_agent: | |
| raise ValueError("Planner agent not initialized") | |
| self.logger.info("Executing planner node") | |
| # Process with planner agent | |
| updated_state = await planner_agent.process(state) | |
| # Log planner results | |
| planning_strategy = getattr(updated_state, 'planning_strategy', {}) | |
| self.logger.info( | |
| "Planner node completed", | |
| agent_sequence=planning_strategy.get('agent_sequence', []), | |
| intent=planning_strategy.get('intent', 'unknown') | |
| ) | |
| return updated_state.__dict__ if hasattr(updated_state, '__dict__') else updated_state | |
| except Exception as e: | |
| self.logger.error(f"Error in planner node: {e}") | |
| # Return state with error information | |
| state.reasoning_log = getattr(state, 'reasoning_log', []) | |
| state.reasoning_log.append(f"Planner error: {str(e)}") | |
| return state.__dict__ if hasattr(state, '__dict__') else state | |
| async def _genre_mood_node(self, state: MusicRecommenderState) -> Dict[str, Any]: | |
| """ | |
| Execute the genre mood agent node. | |
| Args: | |
| state: Current workflow state | |
| Returns: | |
| Updated state dictionary | |
| """ | |
| try: | |
| genre_mood_agent = self.agent_coordinator.get_genre_mood_agent() | |
| if not genre_mood_agent: | |
| raise ValueError("Genre mood agent not initialized") | |
| self.logger.info("Executing genre mood node") | |
| # Process with genre mood agent | |
| updated_state = await genre_mood_agent.process(state) | |
| # Log results | |
| genre_mood_recs = getattr(updated_state, 'genre_mood_recommendations', []) | |
| self.logger.info( | |
| "Genre mood node completed", | |
| recommendations_count=len(genre_mood_recs) | |
| ) | |
| return updated_state.__dict__ if hasattr(updated_state, '__dict__') else updated_state | |
| except Exception as e: | |
| self.logger.error(f"Error in genre mood node: {e}") | |
| # Return state with error information | |
| state.reasoning_log = getattr(state, 'reasoning_log', []) | |
| state.reasoning_log.append(f"Genre mood error: {str(e)}") | |
| return state.__dict__ if hasattr(state, '__dict__') else state | |
| async def _discovery_node(self, state: MusicRecommenderState) -> Dict[str, Any]: | |
| """ | |
| Execute the discovery agent node. | |
| Args: | |
| state: Current workflow state | |
| Returns: | |
| Updated state dictionary | |
| """ | |
| try: | |
| discovery_agent = self.agent_coordinator.get_discovery_agent() | |
| if not discovery_agent: | |
| raise ValueError("Discovery agent not initialized") | |
| self.logger.info("Executing discovery node") | |
| # Process with discovery agent | |
| updated_state = await discovery_agent.process(state) | |
| # Log results | |
| discovery_recs = getattr(updated_state, 'discovery_recommendations', []) | |
| self.logger.info( | |
| "Discovery node completed", | |
| recommendations_count=len(discovery_recs) | |
| ) | |
| return updated_state.__dict__ if hasattr(updated_state, '__dict__') else updated_state | |
| except Exception as e: | |
| self.logger.error(f"Error in discovery node: {e}") | |
| # Return state with error information | |
| state.reasoning_log = getattr(state, 'reasoning_log', []) | |
| state.reasoning_log.append(f"Discovery error: {str(e)}") | |
| return state.__dict__ if hasattr(state, '__dict__') else state | |
| async def _judge_node(self, state: MusicRecommenderState) -> Dict[str, Any]: | |
| """ | |
| Execute the judge agent node. | |
| Args: | |
| state: Current workflow state | |
| Returns: | |
| Updated state dictionary | |
| """ | |
| try: | |
| judge_agent = self.agent_coordinator.get_judge_agent() | |
| if not judge_agent: | |
| raise ValueError("Judge agent not initialized") | |
| self.logger.info("Executing judge node") | |
| # Process with judge agent | |
| updated_state = await judge_agent.process(state) | |
| # Log results | |
| final_recs = getattr(updated_state, 'final_recommendations', []) | |
| self.logger.info( | |
| "Judge node completed", | |
| final_recommendations_count=len(final_recs) | |
| ) | |
| return updated_state.__dict__ if hasattr(updated_state, '__dict__') else updated_state | |
| except Exception as e: | |
| self.logger.error(f"Error in judge node: {e}") | |
| # Return state with error information | |
| state.reasoning_log = getattr(state, 'reasoning_log', []) | |
| state.reasoning_log.append(f"Judge error: {str(e)}") | |
| return state.__dict__ if hasattr(state, '__dict__') else state | |
| async def execute_workflow(self, initial_state: MusicRecommenderState) -> Dict[str, Any]: | |
| """ | |
| Execute the complete workflow. | |
| Args: | |
| initial_state: Initial workflow state | |
| Returns: | |
| Final workflow state | |
| """ | |
| if not self.graph: | |
| raise ValueError("Workflow graph not built. Call build_workflow_graph() first.") | |
| self.logger.info("Starting workflow execution") | |
| try: | |
| # Execute workflow | |
| final_state = await self.graph.ainvoke(initial_state) | |
| self.logger.info("Workflow execution completed successfully") | |
| return final_state | |
| except Exception as e: | |
| self.logger.error(f"Workflow execution failed: {e}") | |
| raise | |
| def get_workflow_graph(self) -> Optional[StateGraph]: | |
| """Get the compiled workflow graph.""" | |
| return self.graph |