Spaces:
Running
Running
File size: 6,606 Bytes
d64fd55 | 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 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 | from typing import Dict, Any, Optional
import logging
import json
import time
from observability import logger as obs_logger
from observability import components as obs_components
from domain.training.agent_models import PlanOutput, WeeklySummary
from domain.runner.profile import RunnerProfile
from domain.runner.goal import Goal
from llm.base import LLMClient
from pathlib import Path
logger = logging.getLogger(__name__)
class PlanAgent:
"""
Generates a weekly plan using an LLM based on recent mileage and consistency 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/plan/../../prompts/
base_path = Path(__file__).parent.parent.parent / "prompts"
filename = f"plan_{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 / "plan_en.txt"
if not file_path.exists():
logger.error("English prompt file missing!")
return "You are generating a weekly running plan. Output 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 generating a weekly running plan. Output JSON."
async def run(
self,
features: WeeklySummary,
language: str = "en",
profile: Optional[RunnerProfile] = None,
goal: Optional[Goal] = None,
) -> Dict[str, Any]:
with obs_logger.start_span("plan_agent.run", obs_components.AGENT):
start_time = time.time()
# Load language-specific instruction
self.instruction = self._load_instruction(language)
# Take most recent week from WeeklySummary
if features.weekly_km:
# take most recent week by sorting keys (YYYY-WW)
weeks = sorted(features.weekly_km.items(), reverse=True)
recent_km = weeks[0][1]
else:
recent_km = 0.0
cons = features.consistency_score
# Construct Prompt
prompt = self._construct_prompt(
recent_km, cons, language=language, profile=profile, goal=goal
)
try:
# Call LLM via Client
with obs_logger.start_span("plan_agent.llm", obs_components.AGENT):
plan_output = await self.llm_client.generate(
prompt, instruction=self.instruction, schema=None, name="plan_agent"
)
result = None
if isinstance(plan_output, PlanOutput):
result = {"plan": plan_output.plan}
elif isinstance(plan_output, dict):
if "plan" in plan_output:
if isinstance(plan_output["plan"], str):
result = {"plan": plan_output["plan"]}
else:
result = {"plan": json.dumps(plan_output)}
else:
result = {"plan": json.dumps({"plan": plan_output})}
elif isinstance(plan_output, str):
raw = plan_output.strip()
result = {"plan": raw}
else:
logger.error(f"Unexpected response type from LLM: {type(plan_output)}")
fail_msg = (
"Não foi possível gerar o plano neste momento."
if language == "pt-BR"
else "Could not generate plan at this time."
)
result = {"plan": fail_msg}
return result
except Exception as e:
duration_ms = (time.time() - start_time) * 1000
obs_logger.log_event(
"error",
f"Failed to generate plan: {e}",
event="error",
component=obs_components.AGENT,
duration_ms=duration_ms,
)
logger.error(f"Failed to generate plan with LLM: {e}", exc_info=True)
fail_msg = (
"Não foi possível gerar o plano neste momento devido a um erro."
if language == "pt-BR"
else "Could not generate plan at this time due to an error."
)
return {"plan": fail_msg}
def _construct_prompt(
self,
recent_km: float,
consistency: int,
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"""
**Perfil do Corredor:**
{profile_context}
- Quilometragem Semanal Recente: {recent_km:.1f} km
- Score de Consistência: {consistency}/100
"""
return f"""
**Runner Profile:**
{profile_context}
- Recent Weekly Mileage: {recent_km:.1f} km
- Consistency Score: {consistency}/100
"""
|