Spaces:
Sleeping
Sleeping
Download src/data_agent_baseline/visualization/narratives.py from CoreModelsResearch/DAO: direct link, hf CLI and curl.
- Browser
- Download file 3.26 kB
-
https://huggingface.co/spaces/CoreModelsResearch/DAO/resolve/main/src/data_agent_baseline/visualization/narratives.py
- Command line
-
hf download hf://spaces/CoreModelsResearch/DAO/src/data_agent_baseline/visualization/narratives.py
-
curl -L -o narratives.py https://huggingface.co/spaces/CoreModelsResearch/DAO/resolve/main/src/data_agent_baseline/visualization/narratives.py
3.26 kB
| """Map technical operations to executive-friendly narratives.""" | |
| from typing import Dict, Tuple | |
| # Action to executive-friendly description mapping | |
| ACTION_NARRATIVES: Dict[str, Tuple[str, str]] = { | |
| # (emoji, narrative description) | |
| "list_context": ("📂", "Cataloging available data sources"), | |
| "read_csv": ("📊", "Examining structured data files"), | |
| "read_knowledge": ("📚", "Reading business context and rules"), | |
| "build_auto_db": ("🔧", "Preparing unified data workspace"), | |
| "semantic_extraction": ("🧠", "Extracting domain knowledge and relationships"), | |
| "planner": ("🎯", "Designing solution strategy"), | |
| "critic_plan": ("🔍", "Validating approach feasibility"), | |
| "execute_python": ("⚙️", "Executing data analysis"), | |
| "critic_execute": ("✅", "Verifying results quality"), | |
| } | |
| def get_action_narrative(action: str) -> Tuple[str, str]: | |
| """Get emoji and narrative for an action.""" | |
| return ACTION_NARRATIVES.get(action, ("🔄", f"Processing: {action}")) | |
| def get_phase_description(phase: str) -> str: | |
| """Get human-readable phase description.""" | |
| phase_map = { | |
| "exploration": "🔍 Data Discovery & Understanding", | |
| "planning": "📋 Solution Strategy Development", | |
| "execution": "⚙️ Analysis Execution & Refinement", | |
| "validation": "✅ Quality Assurance & Verification", | |
| "completion": "🎉 Task Completion", | |
| } | |
| return phase_map.get(phase, phase) | |
| def format_reasoning(reasoning: str, max_length: int = 200) -> str: | |
| """Format plan reasoning for executive display.""" | |
| if not reasoning: | |
| return "No reasoning provided" | |
| # Clean up technical jargon | |
| cleaned = reasoning.replace("filtering by", "finding records where") | |
| cleaned = cleaned.replace("joining", "connecting") | |
| cleaned = cleaned.replace("DataFrame", "data table") | |
| if len(cleaned) > max_length: | |
| return cleaned[:max_length-3] + "..." | |
| return cleaned | |
| def get_success_emoji(succeeded: bool) -> str: | |
| """Get appropriate emoji for success status.""" | |
| return "✅" if succeeded else "❌" | |
| def get_confidence_indicator(attempts: int) -> Tuple[str, str]: | |
| """Get confidence level based on number of attempts.""" | |
| if attempts == 1: | |
| return "🟢", "High - Solved on first try" | |
| elif attempts <= 3: | |
| return "🟡", "Good - Minor refinements needed" | |
| elif attempts <= 5: | |
| return "🟠", "Moderate - Multiple iterations required" | |
| else: | |
| return "🔴", "Low - Significant challenges encountered" | |
| def summarize_issue(issue: str) -> str: | |
| """Convert technical issue to executive summary.""" | |
| # Common patterns | |
| if "empty" in issue.lower(): | |
| return "No matching records found - refining search criteria" | |
| elif "column" in issue.lower(): | |
| return "Data structure mismatch - adjusting field selection" | |
| elif "type" in issue.lower() or "error" in issue.lower(): | |
| return "Data format issue - applying corrections" | |
| elif "filter" in issue.lower(): | |
| return "Search criteria too restrictive - broadening scope" | |
| else: | |
| # Truncate long technical messages | |
| if len(issue) > 80: | |
| return issue[:77] + "..." | |
| return issue | |