"""Recipe Enrichment — uses LLM to generate complete meals and detailed instructions. Transforms basic protein-centric recipes into complete meals with: - Complementary side dishes (grain/starch + vegetables) - Detailed step-by-step instructions with quantities, temperatures, timing - Sauce breakdowns with specific measurements Runs automatically during recipe ingestion so new recipes are complete meals. """ from __future__ import annotations import json import logging import re from typing import Dict, List, Optional, Any import requests from app.config import settings logger = logging.getLogger(__name__) _RATE_LIMIT_SECS = 0.5 class RecipeEnrichmentService: """Enrich recipes with sides and detailed instructions via LLM.""" def __init__(self) -> None: self.enabled = bool(settings.OLLAMA_API_KEY) def enrich(self, recipe_name: str, ingredients: List[dict], instructions: List[str]) -> dict: """Enrich a recipe. Returns dict with 'side_dishes' and 'instructions'. On any failure, returns original instructions with no side dishes. """ if not self.enabled: logger.info("RecipeEnrichment: OLLAMA_API_KEY not set — skipping enrichment") return {"side_dishes": [], "instructions": instructions} prompt = self._build_prompt(recipe_name, ingredients, instructions) raw = self._call_llm(prompt) if not raw: return {"side_dishes": [], "instructions": instructions} parsed = self._parse_response(raw) if parsed is None: return {"side_dishes": [], "instructions": instructions} return parsed def _build_prompt(self, recipe_name: str, ingredients: List[dict], instructions: List[str]) -> str: """Construct the enrichment prompt.""" ing_lines = "\n".join( f"- {ing.get('name', '')}: {ing.get('qty', '')} {ing.get('unit', '')}".strip() for ing in ingredients ) inst_lines = "\n".join(f"{i+1}. {step}" for i, step in enumerate(instructions)) return ( f"You are a professional chef and recipe writer. Rewrite the following recipe " f"into a complete, detailed meal plan with specific instructions.\n\n" f"RECIPE: {recipe_name}\n\n" f"CURRENT INGREDIENTS:\n{ing_lines}\n\n" f"CURRENT INSTRUCTIONS:\n{inst_lines}\n\n" f"TASK:\n" f"1. Rewrite ALL instructions to be extremely specific for a home cook. Include:\n" f" - Exact temperatures (in °F)\n" f" - Precise quantities in each step\n" f" - Exact timing (e.g., 'sauté for 3 minutes')\n" f" - Sauce preparation broken into separate steps with exact measurements\n" f" - Visual cues (e.g., 'until golden brown', 'until internal temp reaches 145°F')\n" f" - Resting times\n\n" f"2. Suggest 1-2 complementary side dishes that create a balanced, complete meal.\n" f" For each side dish include:\n" f" - Name (e.g., 'Garlic Butter Rice')\n" f" - Ingredients list with quantities\n" f" - Brief 2-3 sentence preparation description\n" f" Choose grains/starches (rice, pasta, quinoa, couscous) and/or vegetables.\n\n" f"RETURN FORMAT — Valid JSON only (no markdown, no commentary):\n" f"{{\n" f' "instructions": ["Step 1...", "Step 2..."],\n' f' "side_dishes": [\n' f' {{\n' f' "name": "Side Dish Name",\n' f' "ingredients": [\n' f' {{"name": "ingredient name", "qty": 1.0, "unit": "cup"}}\n' f' ],\n' f' "prep_notes": "Brief preparation steps..."\n' f' }}\n' f' ]\n' f"}}" ) def _call_llm(self, prompt: str) -> Optional[str]: """Call Ollama LLM and return raw content string.""" try: resp = requests.post( f"{settings.OLLAMA_BASE_URL}/chat/completions", headers={ "Content-Type": "application/json", "Authorization": f"Bearer {settings.OLLAMA_API_KEY}", }, json={ "model": settings.OLLAMA_MODEL, "messages": [{"role": "user", "content": prompt}], "max_tokens": 2000, "temperature": 0.3, }, timeout=60, ) resp.raise_for_status() except requests.RequestException as exc: logger.warning("RecipeEnrichment LLM call failed: %s", exc) return None content = resp.json().get("choices", [{}])[0].get("message", {}).get("content", "") # Strip thinking blocks that some models emit content = re.sub(r".*?", "", content, flags=re.DOTALL).strip() return content def _parse_response(self, raw: str) -> Optional[dict]: """Parse LLM response into structured dict. Handles markdown code blocks.""" # Try to extract JSON from markdown code block code_block = re.search(r"```(?:json)?\s*(.*?)\s*```", raw, re.DOTALL) if code_block: raw = code_block.group(1).strip() # Try to find JSON object boundaries start = raw.find("{") end = raw.rfind("}") if start == -1 or end == -1 or end <= start: logger.warning("RecipeEnrichment: no JSON object found in response") return None json_str = raw[start:end + 1] try: data = json.loads(json_str) except json.JSONDecodeError as exc: logger.warning("RecipeEnrichment: JSON parse failed: %s", exc) return None # Validate structure if not isinstance(data.get("instructions"), list): logger.warning("RecipeEnrichment: 'instructions' missing or not a list") return None side_dishes = data.get("side_dishes", []) if not isinstance(side_dishes, list): logger.warning("RecipeEnrichment: 'side_dishes' not a list") side_dishes = [] # Validate side dish structure validated_sides = [] for side in side_dishes: if not isinstance(side, dict): continue if not side.get("name"): continue validated_sides.append({ "name": side["name"], "ingredients": [ { "name": ing.get("name", ""), "qty": ing.get("qty", 0), "unit": ing.get("unit", ""), } for ing in (side.get("ingredients") or []) if ing.get("name") ], "prep_notes": side.get("prep_notes", ""), }) return { "instructions": [str(step) for step in data["instructions"] if step], "side_dishes": validated_sides, }