""" LLM service for converting natural language to structured task data. """ import json import logging from abc import ABC, abstractmethod from typing import Dict, List from app.core.config import settings # Configure logger logger = logging.getLogger(__name__) class LLMService(ABC): """Abstract base class for LLM service implementations.""" @abstractmethod async def chat_to_tasks(self, prompt: str) -> List[Dict]: """ Convert natural language input to structured task objects. Args: prompt: User's natural language input describing tasks Returns: List of dictionary objects representing tasks """ pass class OpenAIService(LLMService): """OpenAI implementation of LLM service.""" def __init__(self): """Initialize the OpenAI service.""" try: import openai self.client = openai.AsyncOpenAI(api_key=settings.openai_api_key) self.model = settings.openai_model except (ImportError, AttributeError) as e: logger.error(f"Failed to initialize OpenAI service: {e}") raise RuntimeError(f"OpenAI service initialization failed: {e}") async def chat_to_tasks(self, prompt: str) -> List[Dict]: """ Convert natural language to tasks using OpenAI. Args: prompt: User's natural language input Returns: List of task dictionaries """ system_prompt = """ You are a task extraction assistant. Your job is to convert the user's natural language input into one or more structured task objects. Each task should have these properties: - title: A short, clear title for the task - description: A more detailed description of what needs to be done - due_date: When the task is due (ISO format date string, or null if not specified) - priority: The priority level (high, medium, low) - status: The status (defaults to "pending") Return ONLY a JSON array of task objects without any additional text or explanation. """ try: response = await self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt} ], response_format={"type": "json_object"}, temperature=0.2, ) # Extract the JSON content from the response content = response.choices[0].message.content result = json.loads(content) # Expect a "tasks" key in the response JSON if "tasks" in result: return result["tasks"] return [result] # Return as a single-item list if no "tasks" key except Exception as e: logger.error(f"OpenAI API error: {e}") raise RuntimeError(f"Failed to process request with OpenAI: {e}") class GeminiService(LLMService): """Google Gemini implementation of LLM service.""" def __init__(self): """Initialize the Gemini service.""" try: import google.generativeai as genai genai.configure(api_key=settings.gemini_api_key) self.model = genai.GenerativeModel(settings.gemini_model) except (ImportError, AttributeError) as e: logger.error(f"Failed to initialize Gemini service: {e}") raise RuntimeError(f"Gemini service initialization failed: {e}") async def chat_to_tasks(self, prompt: str) -> List[Dict]: """ Convert natural language to tasks using Google Gemini. Args: prompt: User's natural language input Returns: List of task dictionaries """ system_prompt = """ You are a task extraction assistant. Your job is to convert the user's natural language input into one or more structured task objects. Each task should have these properties: - title: A short, clear title for the task - description: A more detailed description of what needs to be done - due_date: When the task is due (ISO format date string, or null if not specified) - priority: The priority level (high, medium, low) - status: The status (defaults to "pending") Return ONLY a JSON array of task objects without any additional text or explanation. Format your response as valid JSON with a "tasks" key that contains an array of task objects. """ try: chat = self.model.start_chat(history=[ {"role": "user", "parts": [system_prompt]}, {"role": "model", "parts": ["I understand. I'll convert user inputs into JSON task objects with the specified properties."]} ]) response = await chat.send_message_async(prompt) content = response.text # Extract JSON from the response # This handles cases where the model might add markdown code blocks if "```json" in content: json_str = content.split("```json")[1].split("```")[0].strip() elif "```" in content: json_str = content.split("```")[1].strip() else: json_str = content.strip() result = json.loads(json_str) # Expect a "tasks" key in the response JSON if "tasks" in result: return result["tasks"] return [result] # Return as a single-item list if no "tasks" key except Exception as e: logger.error(f"Gemini API error: {e}") raise RuntimeError(f"Failed to process request with Gemini: {e}") class MockLLMService(LLMService): """Mock LLM service for testing.""" async def chat_to_tasks(self, prompt: str) -> List[Dict]: """ Return mock tasks based on the prompt. Args: prompt: User's natural language input Returns: List of task dictionaries """ # Simple parsing logic for testing words = prompt.lower().split() priority = "medium" if "urgent" in words or "important" in words: priority = "high" elif "low" in words or "minor" in words: priority = "low" # Create a basic task from the prompt return [{ "title": prompt[:50] + ("..." if len(prompt) > 50 else ""), "description": prompt, "due_date": None, "priority": priority, "status": "pending" }] # Factory function to create the appropriate LLM service def get_llm_service() -> LLMService: """ Factory function for LLM service dependency injection. Returns: An instance of a concrete LLMService implementation """ llm_provider = settings.LLM_PROVIDER.lower() if llm_provider == "openai" and settings.openai_api_key: return OpenAIService() elif llm_provider == "gemini" and settings.gemini_api_key: return GeminiService() elif llm_provider == "mock" or settings.environment == "test": # Use mock service for testing or when configured return MockLLMService() else: # Default to mock service if configuration is incomplete logger.warning(f"LLM provider '{llm_provider}' not properly configured - using mock service") return MockLLMService()