"""
Planning Display for BeatDebate Chat Interface
This module visualizes the PlannerAgent's strategic thinking process
in real-time for the UI, showcasing the planning behavior for AgentX.
"""
import logging
from typing import Dict, Any
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class PlanningDisplay:
"""
Visualizes planning strategies from the PlannerAgent.
Features:
- Strategic planning breakdown
- Agent coordination visualization
- Evaluation criteria display
- Execution monitoring indicators
"""
def __init__(self):
"""Initialize the planning display."""
self.logger = logger
def format_planning_strategy(self, strategy: Dict[str, Any]) -> str:
"""
Format a complete planning strategy for display.
Args:
strategy: Planning strategy from PlannerAgent
Returns:
Formatted HTML string for display
"""
try:
if not strategy:
return self._format_no_strategy()
html_parts = []
# Header
html_parts.append(self._format_strategy_header())
# Task Analysis
task_analysis = strategy.get("task_analysis", {})
if task_analysis:
html_parts.append(self._format_task_analysis(task_analysis))
# Coordination Strategy
coordination = strategy.get("coordination_strategy", {})
if coordination:
coordination_html = self._format_coordination_strategy(
coordination
)
html_parts.append(coordination_html)
# Evaluation Framework
evaluation = strategy.get("evaluation_framework", {})
if evaluation:
evaluation_html = self._format_evaluation_framework(evaluation)
html_parts.append(evaluation_html)
# Execution Monitoring
execution = strategy.get("execution_monitoring", {})
if execution:
html_parts.append(self._format_execution_monitoring(execution))
return "\n".join(html_parts)
except Exception as e:
self.logger.error(f"Error formatting planning strategy: {e}")
return self._format_error_display(str(e))
def _format_strategy_header(self) -> str:
"""Format the strategy header."""
return """
🧠 Strategic Planning Analysis
"""
def _format_task_analysis(self, task_analysis: Dict[str, Any]) -> str:
"""Format task analysis section."""
primary_goal = task_analysis.get("primary_goal", "Music discovery")
complexity = task_analysis.get("complexity_level", "medium")
context_factors = task_analysis.get("context_factors", [])
# Complexity indicator
complexity_color = {
"simple": "#28a745",
"medium": "#ffc107",
"complex": "#dc3545"
}.get(complexity, "#6c757d")
# Context factors list
context_html = ""
if context_factors:
context_items = "\n".join([
f"{factor}"
for factor in context_factors[:3] # Limit to 3 items
])
context_html = f"""
"""
flex_style = (
"display: flex; justify-content: space-between; "
"align-items: center;"
)
goal_style = "color: #333; font-size: 0.95em;"
return f"""
🎯 Primary Goal
{complexity.upper()}
{primary_goal}
{context_html}
"""
def _format_coordination_strategy(
self,
coordination: Dict[str, Any]
) -> str:
"""Format agent coordination strategy."""
agents_html = []
# GenreMoodAgent strategy
genre_mood = coordination.get("genre_mood_agent", {})
if genre_mood:
focus = genre_mood.get("focus", "Genre and mood analysis")
genre_style = "color: #155724; font-size: 0.85em;"
genre_p_style = (
"margin: 3px 0 0 0; color: #155724; font-size: 0.8em;"
)
agents_html.append(f"""
""")
# DiscoveryAgent strategy
discovery = coordination.get("discovery_agent", {})
if discovery:
focus = discovery.get("focus", "Discovery and novelty search")
discovery_style = "color: #856404; font-size: 0.85em;"
discovery_p_style = (
"margin: 3px 0 0 0; color: #856404; font-size: 0.8em;"
)
agents_html.append(f"""
""")
agents_content = "\n".join(agents_html)
coord_style = "color: #333; font-size: 0.95em;"
return f"""
🤝 Agent Coordination
{agents_content}
"""
def _format_evaluation_framework(
self,
evaluation: Dict[str, Any]
) -> str:
"""Format evaluation framework."""
weights = evaluation.get("primary_weights", {})
diversity_targets = evaluation.get("diversity_targets", {})
# Primary weights
weights_html = ""
if weights:
weight_items = []
# Limit to 3 items
for criterion, weight in list(weights.items())[:3]:
if isinstance(weight, float):
weight_percent = int(weight * 100)
else:
weight_percent = weight
criterion_name = criterion.replace('_', ' ').title()
flex_style = (
"display: flex; justify-content: space-between; "
"margin: 3px 0;"
)
criterion_style = "font-size: 0.8em;"
weight_style = "font-weight: bold; font-size: 0.8em;"
weight_items.append(f"""
{criterion_name}
{weight_percent}%
""")
weights_html = "\n".join(weight_items)
# Diversity targets
diversity_html = ""
if diversity_targets:
diversity_count = len(diversity_targets)
diversity_html = f"""
Diversity Targets: {diversity_count} criteria
"""
eval_style = "color: #333; font-size: 0.95em;"
return f"""
⚖️ Evaluation Criteria
{weights_html}
{diversity_html}
"""
def _format_execution_monitoring(
self,
execution: Dict[str, Any]
) -> str:
"""Format execution monitoring setup."""
quality_thresholds = execution.get("quality_thresholds", {})
fallback_strategies = execution.get("fallback_strategies", [])
# Quality thresholds
threshold_html = ""
if quality_thresholds:
min_confidence = quality_thresholds.get("min_confidence", 0.6)
threshold_html = f"""
Quality Gate: {min_confidence:.0%} minimum
"""
# Fallback strategies
fallback_html = ""
if fallback_strategies:
fallback_count = len(fallback_strategies)
fallback_html = f"""
Fallback Plans: {fallback_count} strategies
"""
exec_style = "color: #333; font-size: 0.95em;"
return f"""
📊 Execution Monitoring
{threshold_html}
{fallback_html}
"""
def _format_no_strategy(self) -> str:
"""Format display when no strategy is available."""
return """
🧠 Planning strategy will appear here...
"""
def _format_error_display(self, error_message: str) -> str:
"""Format error display."""
return f"""
❌ Error displaying strategy: {error_message}
"""
def format_planning_progress(self, stage: str, details: str = "") -> str:
"""
Format planning progress indicator.
Args:
stage: Current planning stage
details: Additional details about the stage
Returns:
Formatted HTML for progress display
"""
stage_icons = {
"analyzing": "🔍",
"coordinating": "🤝",
"evaluating": "⚖️",
"monitoring": "📊",
"complete": "✅"
}
icon = stage_icons.get(stage, "🧠")
details_text = f": {details}" if details else ""
return f"""
{icon} {stage.title()}{details_text}
"""