Spaces:
Running
Running
| from typing import Dict, Any, Union, Optional | |
| from datetime import datetime | |
| import logging | |
| import json | |
| import time | |
| from pathlib import Path | |
| from observability import logger as obs_logger | |
| from observability import components as obs_components | |
| from domain.training.agent_models import WeeklyTrends, Insight, InsightsOutput, RiskAssessment | |
| from domain.training.run import Run | |
| from domain.runner.profile import RunnerProfile | |
| from domain.runner.goal import Goal | |
| from llm.base import LLMClient | |
| logger = logging.getLogger(__name__) | |
| class InsightsAgent: | |
| """ | |
| Generate actionable insights for a runner using an LLM via the provided LLMClient. | |
| """ | |
| def __init__(self, llm_client: LLMClient): | |
| self.llm_client = llm_client | |
| self.instruction = self._load_instruction("en") | |
| def _load_instruction(self, language: str = "en") -> str: | |
| try: | |
| # Resolve path relative to this file: src/agents/insights/../../prompts/ | |
| base_path = Path(__file__).parent.parent.parent / "prompts" | |
| filename = f"insights_{language}.txt" | |
| file_path = base_path / filename | |
| if not file_path.exists(): | |
| logger.warning(f"Prompt file not found: {file_path}. Falling back to English.") | |
| file_path = base_path / "insights_en.txt" | |
| if not file_path.exists(): | |
| # Ultimate fallback if even English file is missing | |
| logger.error("English prompt file missing!") | |
| return "You are an expert running coach. Generate insights in JSON." | |
| return file_path.read_text(encoding="utf-8") | |
| except Exception as e: | |
| logger.error(f"Error loading prompt for language {language}: {e}") | |
| return "You are an expert running coach. Generate insights in JSON." | |
| async def run( | |
| self, | |
| run_features: Run, | |
| weekly_trends: WeeklyTrends, | |
| risk_level: str = "LOW", | |
| language: str = "en", | |
| profile: Optional[RunnerProfile] = None, | |
| goal: Optional[Goal] = None, | |
| ) -> Dict[str, Any]: | |
| with obs_logger.start_span("insights_agent.run", obs_components.AGENT): | |
| start_time = time.time() | |
| # Load language-specific instruction | |
| self.instruction = self._load_instruction(language) | |
| # Construct Prompt | |
| prompt = self._construct_prompt( | |
| run_features, | |
| weekly_trends, | |
| risk_level, | |
| language=language, | |
| profile=profile, | |
| goal=goal, | |
| ) | |
| try: | |
| # Call LLM via Client | |
| with obs_logger.start_span("insights_agent.llm", obs_components.AGENT): | |
| insights_output = await self.llm_client.generate( | |
| prompt, instruction=self.instruction, schema=InsightsOutput, name="insights_agent" | |
| ) | |
| if isinstance(insights_output, InsightsOutput): | |
| result = insights_output.model_dump() | |
| else: | |
| # Handle unexpected response type | |
| logger.error(f"Unexpected response type from LLM: {type(insights_output)}") | |
| result = self._fallback_error(language) | |
| return result | |
| except Exception as e: | |
| duration_ms = (time.time() - start_time) * 1000 | |
| obs_logger.log_event( | |
| "error", | |
| f"Failed to generate insights: {e}", | |
| event="error", | |
| component=obs_components.AGENT, | |
| duration_ms=duration_ms, | |
| ) | |
| logger.error(f"Failed to generate insights with LLM: {e}", exc_info=True) | |
| return self._fallback_error(language) | |
| def _fallback_error(self, language: str = "en") -> Dict[str, Any]: | |
| is_pt = language == "pt-BR" | |
| if is_pt: | |
| return { | |
| "error": "Não foi possível gerar os insights neste momento.", | |
| "primary_lever": {"message": "Não foi possível gerar insights."}, | |
| "risk_signal": {"message": "Avaliação de risco indisponível."}, | |
| "key_observations": [], | |
| "summary": {"message": "Falha na geração de insights."}, | |
| } | |
| return { | |
| "error": "Could not generate insights at this time.", | |
| "primary_lever": {"message": "Could not generate insights."}, | |
| "risk_signal": {"message": "Risk assessment unavailable."}, | |
| "key_observations": [], | |
| "summary": {"message": "Insights generation failed."}, | |
| } | |
| def _construct_prompt( | |
| self, | |
| run: Run, | |
| trends: WeeklyTrends, | |
| risk_level: str, | |
| language: str = "en", | |
| profile: Optional[RunnerProfile] = None, | |
| goal: Optional[Goal] = None, | |
| ) -> str: | |
| is_pt = language == "pt-BR" | |
| profile_context = "" | |
| if profile: | |
| profile_context = "\n**Runner Profile Context:**\n" | |
| if profile.runner_display_name: | |
| profile_context += f"- Display Name: {profile.runner_display_name}\n" | |
| if profile.age: | |
| profile_context += f"- Age: {profile.age}\n" | |
| if profile.experience_level: | |
| profile_context += f"- Experience Level: {profile.experience_level}\n" | |
| if profile.baseline_weekly_km: | |
| profile_context += f"- Baseline Weekly KM: {profile.baseline_weekly_km}\n" | |
| if profile.injury_history_notes: | |
| profile_context += f"- Injury Notes: {profile.injury_history_notes}\n" | |
| if goal: | |
| goal_type_label = goal.type.replace("_", " ").title() | |
| date_str = goal.target_date.strftime("%Y-%m-%d") if goal.target_date else "N/A" | |
| profile_context += f"\n**Current Active Goal:**\n" | |
| profile_context += f"- Type: {goal_type_label}\n" | |
| profile_context += f"- Target: {goal.target_value} {goal.unit}\n" | |
| profile_context += f"- Target Date: {date_str}\n" | |
| if is_pt: | |
| return f""" | |
| Analise os seguintes dados de corrida e tendências semanais. | |
| Observe que nossa avaliação de risco heurística é atualmente: **{risk_level}**. | |
| {profile_context} | |
| **Dados da Corrida:** | |
| - Distância: {run.total_distance_m} metros | |
| - Duração: {run.total_duration_s} segundos | |
| - Ritmo Médio: {run.avg_pace_min_per_km} min/km | |
| - FC Média: {run.avg_hr_bpm} bpm | |
| - FC Máxima: {run.max_hr_bpm} bpm | |
| - Ganho de Elevação: {run.elevation_gain_m} metros | |
| **Tendências Semanais:** | |
| - Tendência de Ritmo: {trends.pace_trend_s_per_km} s/km (negativo significa mais rápido) | |
| - Tendência de Distância: {trends.distance_trend_m} metros | |
| - Média de Corridas/Semana: {trends.avg_runs_per_week} | |
| - Monotonia: {trends.run_monotony} | |
| Identifique: | |
| 1. A 'Alavanca Principal' (mudança de treino com maior prioridade). | |
| 2. 'Sinais de Risco' significativos (alinhados com nossa heurística de {risk_level}). | |
| 3. Duas 'Observações Principais' que importam para a tomada de decisão. | |
| 4. Um 'Resumo do Treinador' (uma frase final de encorajamento ou instrução). | |
| IMPORTANTE: Todas as mensagens, evidências e justificativas DEVEM estar em Português do Brasil (pt-BR). | |
| """ | |
| return f""" | |
| Analyze the following run data and weekly trends. | |
| Note that our heuristic risk assessment is currently: **{risk_level}**. | |
| **Run Data:** | |
| - Distance: {run.total_distance_m} meters | |
| - Duration: {run.total_duration_s} seconds | |
| - Avg Pace: {run.avg_pace_min_per_km} min/km | |
| - Avg HR: {run.avg_hr_bpm} bpm | |
| - Max HR: {run.max_hr_bpm} bpm | |
| - Elevation Gain: {run.elevation_gain_m} meters | |
| **Weekly Trends:** | |
| - Pace Trend: {trends.pace_trend_s_per_km} s/km (negative means faster) | |
| - Distance Trend: {trends.distance_trend_m} meters | |
| - Avg Runs/Week: {trends.avg_runs_per_week} | |
| - Monotony: {trends.run_monotony} | |
| Identify: | |
| 1. The 'Primary Lever' (top priority training change). | |
| 2. Significant 'Risk Signals' (aligned with our {risk_level} heuristic). | |
| 3. Two 'Key Observations' that matter for decision making. | |
| 4. A 'Coaching Summary' (one-sentence final encouraging or instructional takeaway). | |
| """ | |