Automated Action 97c002ac88 Add AI-powered chat-to-tasks feature
Implement a new endpoint that converts natural language input into structured tasks using an LLM. Features include:
- LLM service abstraction with support for OpenAI and Google Gemini
- Dependency injection pattern for easy provider switching
- Robust error handling and response formatting
- Integration with existing user authentication and task creation
- Fallback to mock LLM service for testing or development
2025-05-17 07:44:19 +00:00

211 lines
7.6 KiB
Python

"""
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()