Spaces:
Paused
Paused
| from __future__ import annotations | |
| import json | |
| import re | |
| import time | |
| from typing import Any | |
| import requests | |
| from app.config import settings | |
| REQUIRED_KEYS = {"summary", "strengths", "improvements", "drills", "practice_plan"} | |
| TRANSITION_PHRASES = ( | |
| "first", | |
| "second", | |
| "third", | |
| "because", | |
| "therefore", | |
| "however", | |
| "for example", | |
| "for instance", | |
| "in practice", | |
| "the takeaway", | |
| "here is why", | |
| "why this matters", | |
| "under the hood", | |
| "as a result", | |
| ) | |
| def build_coaching_prompt( | |
| transcript: str, | |
| metrics: dict[str, Any], | |
| history: list[dict[str, Any]], | |
| previous_feedback: dict[str, Any] | None = None, | |
| active_goals: list[dict[str, Any]] | None = None, | |
| ) -> str: | |
| max_chars = max(1000, settings.coaching_max_transcript_chars) | |
| truncated = transcript[:max_chars] | |
| if len(transcript) > max_chars: | |
| truncated = f"{truncated}\n[transcript truncated]" | |
| history_lines = [] | |
| for item in history[:10]: | |
| history_lines.append( | |
| f"- {item.get('recorded_at', 'unknown')}: score={item.get('overall_score', 0)}, " | |
| f"wpm={item.get('wpm', 0)}, fillers/min={item.get('filler_rate', 0)}" | |
| ) | |
| previous = "" | |
| if previous_feedback: | |
| previous = f"Previous coaching focus: {json.dumps(previous_feedback, ensure_ascii=False)[:1200]}" | |
| goal_lines = [] | |
| for goal in active_goals or []: | |
| target = str(goal.get("target_value", 0)) | |
| if goal.get("comparison") == "between": | |
| target = f"{target} to {goal.get('target_upper')}" | |
| goal_lines.append( | |
| f"- {goal.get('metric_name')}: target {goal.get('comparison')} {target}; " | |
| f"current={goal.get('current_value', 'not available')}; achieved={goal.get('achieved', False)}" | |
| ) | |
| return f"""You are a camera-presenting & technology storytelling AI coach. Analyze the practice session and return only valid JSON. | |
| Current metrics: | |
| - duration_seconds: {metrics.get("duration_seconds", 0)} | |
| - word_count: {metrics.get("word_count", 0)} | |
| - wpm: {metrics.get("wpm", 0)} | |
| - filler_count: {metrics.get("filler_count", 0)} | |
| - filler_rate: {metrics.get("filler_rate", 0)} | |
| - pause_count: {metrics.get("pause_count", 0)} | |
| - vocabulary_density: {metrics.get("vocabulary_density", 0)} | |
| - speech_score: {metrics.get("speech_score", 0)} | |
| - delivery_score: {metrics.get("delivery_score", 0)} | |
| - overall_score: {metrics.get("overall_score", 0)} | |
| - eye_contact_score: {metrics.get("eye_contact_score", "n/a")} | |
| - filler_breakdown: {json.dumps(metrics.get("filler_breakdown") or {}, ensure_ascii=False)} | |
| - session_mode: {metrics.get("session_mode", "Practice")} | |
| - curriculum_day: {metrics.get("curriculum_day", "none")} | |
| - target_keywords: {json.dumps(metrics.get("target_keywords") or [], ensure_ascii=False)} | |
| Recent history: | |
| {chr(10).join(history_lines) if history_lines else "- no prior sessions"} | |
| Active user goals: | |
| {chr(10).join(goal_lines) if goal_lines else "- no active goals"} | |
| {previous} | |
| Transcript: | |
| {truncated} | |
| Return this exact JSON shape, with no markdown: | |
| {{ | |
| "summary": "2-3 sentence assessment", | |
| "strengths": [ | |
| {{"area": "pace", "detail": "specific observation", "score_contribution": "why it matters"}} | |
| ], | |
| "improvements": [ | |
| {{"area": "fillers", "detail": "specific next action", "priority": "high", "example": "optional example"}} | |
| ], | |
| "drills": [ | |
| {{"name": "drill name", "description": "steps", "duration_minutes": 5, "targets": "metric"}} | |
| ], | |
| "practice_plan": {{ | |
| "focus_areas": ["area"], | |
| "suggested_duration_minutes": 15, | |
| "difficulty": "beginner", | |
| "next_session_goal": "specific measurable goal" | |
| }}, | |
| "progress_note": "optional progress comment" | |
| }} | |
| Rules: | |
| - Use 2-4 strengths and 2-4 improvements. | |
| - Use 1-3 drills. | |
| - Cite exact metrics where possible. | |
| - Align the next_session_goal and drills with active user goals when present. | |
| - Difficulty must be beginner, intermediate, or advanced. | |
| - If history is empty, progress_note may be empty. | |
| """ | |
| def _strip_timestamps(transcript: str) -> str: | |
| return re.sub(r"^\[\d+(?:\.\d+)?s\]\s*", "", transcript, flags=re.MULTILINE).strip() | |
| def generate_builtin_coaching_feedback( | |
| transcript: str, | |
| metrics: dict[str, Any], | |
| history: list[dict[str, Any]], | |
| previous_feedback: dict[str, Any] | None = None, | |
| active_goals: list[dict[str, Any]] | None = None, | |
| ) -> dict[str, Any]: | |
| started = time.perf_counter() | |
| clean_text = _strip_timestamps(transcript) | |
| lower_text = clean_text.lower() | |
| sentences = [s.strip() for s in re.split(r"[.!?]+", clean_text) if s.strip()] | |
| avg_sentence_words = ( | |
| round(sum(len(s.split()) for s in sentences) / len(sentences), 1) if sentences else 0.0 | |
| ) | |
| opening_line = sentences[0] if sentences else "" | |
| wpm = float(metrics.get("wpm") or 0) | |
| duration_sec = float(metrics.get("duration_seconds") or 0) | |
| word_count = int(metrics.get("word_count") or 0) | |
| filler_count = int(metrics.get("filler_count") or 0) | |
| filler_rate = float(metrics.get("filler_rate") or 0) | |
| filler_breakdown: dict[str, int] = metrics.get("filler_breakdown") or {} | |
| eye_contact = metrics.get("eye_contact_score") | |
| eye_pct = round(float(eye_contact) * 100) if eye_contact is not None else None | |
| overall_score = float(metrics.get("overall_score") or 0) | |
| speech_score = float(metrics.get("speech_score") or 0) | |
| delivery_score = float(metrics.get("delivery_score") or 0) | |
| vocab_density = float(metrics.get("vocabulary_density") or 0) | |
| pause_count = int(metrics.get("pause_count") or 0) | |
| target_keywords: list[str] = [str(k) for k in (metrics.get("target_keywords") or []) if str(k).strip()] | |
| transitions_used = [phrase for phrase in TRANSITION_PHRASES if phrase in lower_text] | |
| covered_keywords = [kw for kw in target_keywords if kw.lower() in lower_text] | |
| missing_keywords = [kw for kw in target_keywords if kw.lower() not in lower_text] | |
| strengths: list[dict[str, str]] = [] | |
| improvements: list[dict[str, str]] = [] | |
| drills: list[dict[str, Any]] = [] | |
| focus_areas: list[str] = [] | |
| # 1. Pace analysis | |
| if 125 <= wpm <= 160: | |
| strengths.append({ | |
| "area": "Presenter Pace", | |
| "detail": f"Your speaking speed averaged {wpm:.0f} WPM — right inside the ideal 125–160 WPM tech keynote window.", | |
| "score_contribution": "Gives viewers enough cognitive room to absorb technical concepts without losing energy.", | |
| }) | |
| elif wpm > 160: | |
| focus_areas.append("pace") | |
| improvements.append({ | |
| "area": "Rushing Pace", | |
| "detail": f"You averaged {wpm:.0f} WPM, which risks compressing technical terms before the audience processes them. Insert a 1-second breath after every key claim.", | |
| "priority": "high", | |
| "example": opening_line[:110] if opening_line else "Pause for one beat right after your headline statement.", | |
| }) | |
| drills.append({ | |
| "name": "140 WPM Beat-Pause Cadence Drill", | |
| "description": "Deliver a 60-second architecture or product explanation using the teleprompter at 1.0x speed, deliberately pausing for one full breath at every period.", | |
| "duration_minutes": 5, | |
| "targets": "wpm", | |
| }) | |
| elif 0 < wpm < 120: | |
| focus_areas.append("pace") | |
| improvements.append({ | |
| "area": "Vocal Momentum", | |
| "detail": f"Your pace averaged {wpm:.0f} WPM. Tightening transitions between sentences toward 130–150 WPM will make your camera presence feel sharper and more authoritative.", | |
| "priority": "medium", | |
| "example": "Start each point with a crisp verb or headline rather than thinking aloud.", | |
| }) | |
| drills.append({ | |
| "name": "60-Second Rapid Tech Briefing", | |
| "description": "Summarize one tech news story in exactly 60 seconds using 3 bullet cues only, aiming for 135 WPM with crisp sentence endings.", | |
| "duration_minutes": 5, | |
| "targets": "wpm", | |
| }) | |
| # 2. Filler analysis | |
| top_fillers = sorted(filler_breakdown.items(), key=lambda item: item[1], reverse=True)[:3] | |
| top_filler_str = ", ".join(f"'{w}' ({c}x)" for w, c in top_fillers) | |
| if filler_rate <= 2.0 and word_count >= 15: | |
| strengths.append({ | |
| "area": "Clean Articulation", | |
| "detail": f"You held filler words to {filler_rate:.1f}/min ({filler_count} total across {word_count} words), projecting calm authority.", | |
| "score_contribution": "Clean transitions keep technical explanations crisp and credible.", | |
| }) | |
| elif filler_rate > 2.0 or filler_count >= 3: | |
| focus_areas.append("fillers") | |
| detail_suffix = f" Top habits: {top_filler_str}." if top_filler_str else "" | |
| improvements.append({ | |
| "area": "Filler Discipline", | |
| "detail": f"You used {filler_count} fillers ({filler_rate:.1f}/min).{detail_suffix} Replace vocalized thinking with a silent 0.8-second pause while holding lens contact.", | |
| "priority": "high" if filler_rate > 4.5 else "medium", | |
| "example": f"Swap '{top_fillers[0][0]}' for a silent breath before your next clause." if top_fillers else "Close your lips briefly between thoughts instead of bridging with 'um' or 'like'.", | |
| }) | |
| drills.append({ | |
| "name": "Silent Pause Replacement Sprint", | |
| "description": "Speak for 90 seconds on how a favorite developer tool works. Every time you feel an 'um', 'like', or 'basically' coming, stop speaking for one beat, then start the next sentence cleanly.", | |
| "duration_minutes": 6, | |
| "targets": "filler_rate", | |
| }) | |
| # 3. Camera Presence & Eye Contact | |
| if eye_pct is not None and eye_pct >= 70: | |
| strengths.append({ | |
| "area": "Camera Lens Presence", | |
| "detail": f"Strong lens connection at {eye_pct}% eye contact (delivery score {delivery_score:.0f}/100), making viewers feel spoken to directly.", | |
| "score_contribution": "Direct lens contact builds immediate trust in on-camera tech demos and videos.", | |
| }) | |
| elif eye_pct is not None and eye_pct < 65: | |
| focus_areas.append("eye_contact") | |
| improvements.append({ | |
| "area": "Lens Eye Contact", | |
| "detail": f"Camera eye contact measured {eye_pct}%. Position the built-in teleprompter overlay at the top of the frame right beneath your webcam lens and speak to the lens as if to one engineer.", | |
| "priority": "high" if eye_pct < 45 else "medium", | |
| "example": "Lock onto the lens during your opening hook and your closing takeaway sentence.", | |
| }) | |
| drills.append({ | |
| "name": "Lens-Lock Hook & Takeaway Drill", | |
| "description": "Use Cue-Card teleprompter mode and deliver 3 back-to-back 30-second intros while keeping your eyes anchored on the camera ring for >75% of the take.", | |
| "duration_minutes": 5, | |
| "targets": "eye_contact_score", | |
| }) | |
| # 4. Storytelling Structure, Clarity & Keywords | |
| if len(transitions_used) >= 2: | |
| strengths.append({ | |
| "area": "Structured Signposting", | |
| "detail": f"Clear narrative signposts ({', '.join(transitions_used[:4])}) guided the listener cleanly from premise to takeaway.", | |
| "score_contribution": "Signposting helps audiences follow complex architecture and product arguments effortlessly.", | |
| }) | |
| else: | |
| focus_areas.append("structure") | |
| improvements.append({ | |
| "area": "Tech Storytelling Structure", | |
| "detail": "Add explicit structural signposts ('Here is the problem...', 'Under the hood...', 'For example...', 'The takeaway is...') so your audience always knows where they are in your story.", | |
| "priority": "medium", | |
| "example": "Open with a 1-sentence hook, state 2–3 numbered pillars, and land a concrete takeaway.", | |
| }) | |
| if avg_sentence_words > 24: | |
| improvements.append({ | |
| "area": "Sentence Conciseness", | |
| "detail": f"Your sentences averaged {avg_sentence_words:.0f} words. Break compound technical clauses into punchy 12–18 word sentences for stronger on-camera clarity.", | |
| "priority": "medium", | |
| "example": opening_line[:120] if opening_line else "Turn 'and so which means that' chains into crisp standalone statements.", | |
| }) | |
| elif 8 <= avg_sentence_words <= 20 and len(strengths) < 3: | |
| strengths.append({ | |
| "area": "Crisp Sentence Clarity", | |
| "detail": f"Average sentence length of {avg_sentence_words:.0f} words kept your explanation conversational and easy to follow.", | |
| "score_contribution": "Short, direct sentences sound natural on camera and quote well.", | |
| }) | |
| if target_keywords: | |
| if not missing_keywords: | |
| strengths.append({ | |
| "area": "Topic & Keyword Mastery", | |
| "detail": f"You wove in 100% of your target concepts ({', '.join(covered_keywords[:5])}).", | |
| "score_contribution": "Ensures every key technical takeaway lands as planned.", | |
| }) | |
| elif missing_keywords: | |
| improvements.append({ | |
| "area": "Key Concept Coverage", | |
| "detail": f"Covered {len(covered_keywords)}/{len(target_keywords)} target terms; consider weaving in: {', '.join(missing_keywords[:4])}.", | |
| "priority": "low", | |
| "example": f"Anchor '{missing_keywords[0]}' with a concrete real-world example.", | |
| }) | |
| # Ensure 2-4 strengths, 2-4 improvements, and 1-3 drills per contract | |
| if len(strengths) < 2: | |
| strengths.append({ | |
| "area": "Consistent Practice Rep", | |
| "detail": f"Completed a {max(1, round(duration_sec))}s on-camera rep ({word_count} words, vocabulary density {vocab_density:.2f}) with {pause_count} natural pause markers.", | |
| "score_contribution": "Daily reps compound rapidly into effortless camera confidence.", | |
| }) | |
| if len(strengths) < 2: | |
| strengths.append({ | |
| "area": "Vocal Engagement", | |
| "detail": f"Delivered an overall session score of {overall_score:.0f}/100 (speech {speech_score:.0f}/100) — a solid baseline to sharpen on your next take.", | |
| "score_contribution": "Provides measurable benchmarks for your 30-day presenter progression.", | |
| }) | |
| if len(improvements) < 2: | |
| improvements.append({ | |
| "area": "Cold-Open Hook Energy", | |
| "detail": "Challenge yourself to make the first 8 seconds 15% more energetic: lead with a surprising stat, sharp problem, or relatable developer moment before diving into details.", | |
| "priority": "low", | |
| "example": opening_line[:110] if opening_line else "Instead of 'Today I want to talk about...', open directly with the stakes.", | |
| }) | |
| if len(improvements) < 2: | |
| improvements.append({ | |
| "area": "Deliberate Emphasis Pauses", | |
| "detail": "Hold a deliberate 1-second pause right before your final takeaway sentence to underline its importance on camera.", | |
| "priority": "low", | |
| "example": "Pause, smile slightly at the lens, then deliver your closing one-liner.", | |
| }) | |
| if not drills: | |
| drills.append({ | |
| "name": "Hook → Mechanism → Impact Take-Two", | |
| "description": "Re-record the same topic in 90 seconds using 3 bullet cues: (1) 8-second sharp hook, (2) 1 clear analogy or mechanism, (3) 1 actionable takeaway while holding 75%+ lens contact.", | |
| "duration_minutes": 5, | |
| "targets": "overall_score", | |
| }) | |
| # Align next_session_goal with active user goals when present | |
| unmet_goals = [g for g in (active_goals or []) if not g.get("achieved")] | |
| if unmet_goals: | |
| g0 = unmet_goals[0] | |
| target_desc = ( | |
| f"{g0['target_value']}–{g0['target_upper']}" | |
| if g0.get("comparison") == "between" | |
| else f"{g0.get('comparison')} {g0.get('target_value')}" | |
| ) | |
| next_goal = f"Hit your active '{g0.get('metric_name')}' goal ({target_desc}) on a 90-second camera take." | |
| elif "fillers" in focus_areas: | |
| next_goal = f"Cut filler rate from {filler_rate:.1f}/min to under {max(1.5, round(filler_rate * 0.6, 1))}/min using silent beat-pauses." | |
| elif "pace" in focus_areas: | |
| next_goal = "Hold steady conversational cadence between 130 and 155 WPM across a 2-minute tech explanation." | |
| elif "eye_contact" in focus_areas: | |
| next_goal = "Maintain at least 75% camera lens eye contact while speaking from bullet cues." | |
| else: | |
| next_goal = f"Push overall score from {overall_score:.0f} to {min(98, int(overall_score + 6))}+ with a punchy 8-second hook and crisp signposts." | |
| # Progress note vs recent history | |
| prior_rows = [row for row in history if row.get("overall_score") is not None] | |
| progress_note = "" | |
| if len(prior_rows) >= 2: | |
| prior_scores = [float(r.get("overall_score") or 0) for r in prior_rows[1:6]] | |
| if prior_scores: | |
| avg_prior = sum(prior_scores) / len(prior_scores) | |
| delta = overall_score - avg_prior | |
| direction = f"up +{delta:.1f} pts" if delta >= 0 else f"{delta:.1f} pts" | |
| progress_note = ( | |
| f"Compared with your recent {len(prior_scores)}-session average ({avg_prior:.1f}/100), " | |
| f"this session scored {overall_score:.1f}/100 ({direction})." | |
| ) | |
| difficulty = "advanced" if overall_score >= 80 else "intermediate" if overall_score >= 58 else "beginner" | |
| mode_label = metrics.get("session_mode") or "presenting" | |
| day_badge = f" (30-Day Curriculum Day {metrics['curriculum_day']})" if metrics.get("curriculum_day") else "" | |
| summary = ( | |
| f"Solid {mode_label} take{day_badge} scoring {overall_score:.0f}/100 at {wpm:.0f} WPM " | |
| f"with {filler_count} filler{'s' if filler_count != 1 else ''} ({filler_rate:.1f}/min)" | |
| + (f" and {eye_pct}% lens eye contact." if eye_pct is not None else ".") | |
| + f" Your strongest asset was {strengths[0]['area'].lower()}, while your highest-leverage next upgrade is {improvements[0]['area'].lower()}." | |
| ) | |
| payload = { | |
| "summary": summary, | |
| "strengths": strengths[:4], | |
| "improvements": improvements[:4], | |
| "drills": drills[:3], | |
| "practice_plan": { | |
| "focus_areas": (focus_areas or ["storytelling", "camera_presence"])[:3], | |
| "suggested_duration_minutes": 15, | |
| "difficulty": difficulty, | |
| "next_session_goal": next_goal, | |
| }, | |
| "progress_note": progress_note, | |
| } | |
| raw_json = json.dumps(payload, ensure_ascii=False) | |
| payload["raw_response"] = raw_json | |
| payload["model_used"] = "builtin-presenter-coach" | |
| payload["generation_time_ms"] = max(1, int((time.perf_counter() - started) * 1000)) | |
| return payload | |
| def generate_coaching_feedback( | |
| transcript: str, | |
| metrics: dict[str, Any], | |
| history: list[dict[str, Any]], | |
| previous_feedback: dict[str, Any] | None = None, | |
| active_goals: list[dict[str, Any]] | None = None, | |
| ) -> dict[str, Any]: | |
| provider = settings.coaching_provider.lower() | |
| if provider == "builtin": | |
| return generate_builtin_coaching_feedback(transcript, metrics, history, previous_feedback, active_goals) | |
| prompt = build_coaching_prompt(transcript, metrics, history, previous_feedback, active_goals) | |
| started = time.perf_counter() | |
| if provider == "auto": | |
| try: | |
| raw_text = _call_ollama(prompt, connect_timeout=1.5) | |
| parsed = parse_coaching_response(raw_text) | |
| parsed["raw_response"] = raw_text | |
| parsed["model_used"] = settings.coaching_model | |
| parsed["generation_time_ms"] = int((time.perf_counter() - started) * 1000) | |
| return parsed | |
| except Exception: | |
| return generate_builtin_coaching_feedback(transcript, metrics, history, previous_feedback, active_goals) | |
| if provider not in {"ollama", "lmstudio", "llamacpp"}: | |
| raise RuntimeError(f"Unsupported coaching provider '{settings.coaching_provider}'") | |
| try: | |
| if provider == "ollama": | |
| raw_text = _call_ollama(prompt) | |
| else: | |
| raw_text = _call_openai_compatible(prompt, provider) | |
| parsed = parse_coaching_response(raw_text) | |
| parsed["raw_response"] = raw_text | |
| parsed["model_used"] = settings.coaching_model | |
| parsed["generation_time_ms"] = int((time.perf_counter() - started) * 1000) | |
| return parsed | |
| except Exception: | |
| if settings.coaching_fallback_builtin: | |
| return generate_builtin_coaching_feedback(transcript, metrics, history, previous_feedback, active_goals) | |
| raise | |
| def parse_coaching_response(raw_text: str) -> dict[str, Any]: | |
| cleaned = raw_text.strip() | |
| if cleaned.startswith("```"): | |
| cleaned = cleaned.split("\n", 1)[1] if "\n" in cleaned else cleaned[3:] | |
| if cleaned.endswith("```"): | |
| cleaned = cleaned[:-3] | |
| cleaned = cleaned.strip() | |
| start = cleaned.find("{") | |
| end = cleaned.rfind("}") | |
| if start >= 0 and end > start: | |
| cleaned = cleaned[start:end + 1] | |
| try: | |
| parsed = json.loads(cleaned) | |
| except json.JSONDecodeError as exc: | |
| raise RuntimeError(f"Local LLM returned invalid JSON: {exc}") from exc | |
| missing = REQUIRED_KEYS - set(parsed) | |
| if missing: | |
| raise RuntimeError(f"Local LLM response missing required keys: {', '.join(sorted(missing))}") | |
| parsed.setdefault("progress_note", "") | |
| return parsed | |
| def _call_ollama(prompt: str, connect_timeout: float = 3.0) -> str: | |
| url = f"{settings.ollama_base_url.rstrip('/')}/api/generate" | |
| response = requests.post( | |
| url, | |
| json={ | |
| "model": settings.coaching_model, | |
| "prompt": prompt, | |
| "stream": False, | |
| "format": "json", | |
| }, | |
| timeout=(connect_timeout, settings.coaching_timeout_seconds), | |
| ) | |
| response.raise_for_status() | |
| payload = response.json() | |
| text = payload.get("response") | |
| if not isinstance(text, str) or not text.strip(): | |
| raise RuntimeError("Ollama returned an empty coaching response") | |
| return text | |
| def _call_openai_compatible(prompt: str, provider: str) -> str: | |
| base_url = settings.lmstudio_base_url if provider == "lmstudio" else settings.llama_cpp_base_url | |
| url = f"{base_url.rstrip('/')}/v1/chat/completions" | |
| response = requests.post( | |
| url, | |
| json={ | |
| "model": settings.coaching_model, | |
| "temperature": 0.3, | |
| "messages": [{"role": "user", "content": prompt}], | |
| }, | |
| timeout=(3.0, settings.coaching_timeout_seconds), | |
| ) | |
| response.raise_for_status() | |
| payload = response.json() | |
| try: | |
| text = payload["choices"][0]["message"]["content"] | |
| except (KeyError, IndexError, TypeError) as exc: | |
| raise RuntimeError("OpenAI-compatible provider returned an invalid response") from exc | |
| if not isinstance(text, str) or not text.strip(): | |
| raise RuntimeError("OpenAI-compatible provider returned an empty coaching response") | |
| return text | |