From 97c002ac884b3f182fe0027b345443e03548da0f Mon Sep 17 00:00:00 2001 From: Automated Action Date: Sat, 17 May 2025 07:44:19 +0000 Subject: [PATCH] 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 --- README.md | 47 +++++++- app/api/routers/__init__.py | 3 +- app/api/routers/chat.py | 126 +++++++++++++++++++++ app/core/config.py | 18 ++- app/schemas/chat.py | 44 ++++++++ app/schemas/task.py | 6 + app/services/__init__.py | 3 + app/services/llm_service.py | 211 ++++++++++++++++++++++++++++++++++++ main.py | 22 ++-- requirements.txt | 6 +- 10 files changed, 467 insertions(+), 19 deletions(-) create mode 100644 app/api/routers/chat.py create mode 100644 app/schemas/chat.py create mode 100644 app/services/__init__.py create mode 100644 app/services/llm_service.py diff --git a/README.md b/README.md index a928e21..88f8945 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,7 @@ A RESTful API for managing tasks, built with FastAPI and SQLite. - Task CRUD operations with user-based access control - Task status and priority management - Task completion tracking +- AI-powered chat-to-tasks conversion - API documentation with Swagger UI and ReDoc - Health endpoint for monitoring @@ -23,6 +24,8 @@ A RESTful API for managing tasks, built with FastAPI and SQLite. - JWT: JSON Web Tokens for authentication - Passlib: Password hashing and verification - Python-Jose: Python implementation of JWT +- OpenAI/Google Gemini API: AI models for natural language processing +- HTTPX: Async HTTP client for API requests ## API Endpoints @@ -46,6 +49,10 @@ A RESTful API for managing tasks, built with FastAPI and SQLite. - `DELETE /tasks/{task_id}`: Delete a task for the current user - `POST /tasks/{task_id}/complete`: Mark a task as completed for the current user +### AI-Powered Task Creation (requires authentication) + +- `POST /chat/chat-to-tasks`: Convert natural language input into structured tasks + ### Health and Diagnostic Endpoints - `GET /health`: Application health check @@ -146,6 +153,19 @@ curl -X 'POST' \ -H 'Authorization: Bearer YOUR_ACCESS_TOKEN' ``` +### Create tasks from natural language (with authentication) + +```bash +curl -X 'POST' \ + 'https://taskmanagerapi-ttkjqk.backend.im/chat/chat-to-tasks' \ + -H 'accept: application/json' \ + -H 'Authorization: Bearer YOUR_ACCESS_TOKEN' \ + -H 'Content-Type: application/json' \ + -d '{ + "message": "I need to prepare a presentation for Monday, pick up groceries tomorrow, and call John about the project by Friday." +}' +``` + ### Get current user information (with authentication) ```bash @@ -166,6 +186,7 @@ taskmanagerapi/ │ │ ├── deps.py # Dependency injection for authentication │ │ └── routers/ # API route definitions │ │ ├── auth.py # Authentication endpoints +│ │ ├── chat.py # Chat-to-tasks endpoints │ │ └── tasks.py # Task management endpoints │ ├── core/ # Core application code │ │ ├── config.py # Application configuration @@ -182,9 +203,12 @@ taskmanagerapi/ │ ├── models/ # SQLAlchemy models │ │ ├── task.py # Task model │ │ └── user.py # User model -│ └── schemas/ # Pydantic schemas/models -│ ├── task.py # Task schemas -│ └── user.py # User and authentication schemas +│ ├── schemas/ # Pydantic schemas/models +│ │ ├── chat.py # Chat-to-tasks schemas +│ │ ├── task.py # Task schemas +│ │ └── user.py # User and authentication schemas +│ └── services/ # Service integrations +│ └── llm_service.py # LLM service for chat-to-tasks conversion ├── main.py # Application entry point └── requirements.txt # Project dependencies ``` @@ -267,6 +291,23 @@ You can use these credentials to authenticate and get started right away. 3. **Use token**: Include the token in your requests as a Bearer token in the Authorization header 4. **Access protected endpoints**: Use the token to access protected task management endpoints +### Chat-to-Tasks Feature + +The application includes an AI-powered feature that converts natural language inputs into structured task objects: + +1. **Environment Configuration**: Set the `LLM_PROVIDER` environment variable to select the AI model provider: + - `openai`: Uses OpenAI's GPT models (requires `OPENAI_API_KEY` and optionally `OPENAI_MODEL`) + - `gemini`: Uses Google's Gemini models (requires `GEMINI_API_KEY` and optionally `GEMINI_MODEL`) + - `mock`: Uses a simple rule-based mock implementation (default, no API keys needed) + +2. **Using the Feature**: Send a POST request to `/chat/chat-to-tasks` with a natural language message describing your tasks. The API will: + - Process the message using the configured AI model + - Extract task details (title, description, due date, priority) + - Create tasks in the database for the authenticated user + - Return the created tasks in a structured format + +3. **Example**: "I need to finish the report by Friday, call Susan about the meeting tomorrow morning, and buy groceries tonight" will be converted into three separate tasks with appropriate details. + ### API Documentation - Swagger UI: http://localhost:8000/docs diff --git a/app/api/routers/__init__.py b/app/api/routers/__init__.py index ec7a5e3..a0f581e 100644 --- a/app/api/routers/__init__.py +++ b/app/api/routers/__init__.py @@ -1,7 +1,8 @@ from fastapi import APIRouter -from app.api.routers import tasks, auth +from app.api.routers import tasks, auth, chat api_router = APIRouter() api_router.include_router(tasks.router, prefix="/tasks", tags=["tasks"]) api_router.include_router(auth.router, prefix="/auth", tags=["auth"]) +api_router.include_router(chat.router, prefix="/chat", tags=["chat"]) diff --git a/app/api/routers/chat.py b/app/api/routers/chat.py new file mode 100644 index 0000000..c458cb8 --- /dev/null +++ b/app/api/routers/chat.py @@ -0,0 +1,126 @@ +""" +Router for the chat-to-tasks functionality. +""" +import logging + +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy.orm import Session + +from app.api import deps +from app.crud import task as task_crud +from app.models.user import User +from app.schemas.chat import ChatInput, ChatResponse, ChatProcessingError +from app.schemas.task import TaskCreate, TaskRead +from app.services.llm_service import LLMService, get_llm_service +from app.db.session import get_db + +# Set up logger +logger = logging.getLogger(__name__) + +router = APIRouter() + + +@router.post("/chat-to-tasks", response_model=ChatResponse) +async def create_tasks_from_chat( + chat_input: ChatInput, + db: Session = Depends(get_db), + current_user: User = Depends(deps.get_current_active_user), + llm_service: LLMService = Depends(get_llm_service), +): + """ + Convert natural language chat input into one or more task objects. + + This endpoint: + 1. Takes the user's natural language input + 2. Sends it to an LLM for processing + 3. Parses the LLM's response into TaskCreate objects + 4. Creates the tasks in the database + 5. Returns the created tasks + + All tasks are associated with the authenticated user. + """ + if not chat_input.message or len(chat_input.message.strip()) < 3: + raise HTTPException( + status_code=400, + detail="Message must be at least 3 characters long", + ) + + # Initialize response + response = ChatResponse(original_message=chat_input.message) + + try: + # Process the chat message with the LLM service + logger.info(f"Processing chat input: {chat_input.message[:50]}...") + llm_tasks_data = await llm_service.chat_to_tasks(chat_input.message) + + if not llm_tasks_data: + logger.warning("LLM returned no tasks") + response.processing_successful = False + response.error = ChatProcessingError( + error_type="parsing_error", + error_detail="No tasks could be extracted from your message", + ) + return response + + # Convert LLM response to TaskCreate objects and create in DB + created_tasks = [] + + for task_data in llm_tasks_data: + try: + # Map LLM response fields to TaskCreate schema + # Handle different field names or formats that might come from the LLM + task_create_data = { + "title": task_data.get("title", "Untitled Task"), + "description": task_data.get("description", ""), + "priority": task_data.get("priority", "medium").lower(), + } + + # Handle due_date if present + if due_date := task_data.get("due_date"): + if due_date != "null" and due_date is not None: + task_create_data["due_date"] = due_date + + # Map status if present (convert "pending" to "todo" if needed) + if status := task_data.get("status"): + if status.lower() == "pending": + task_create_data["status"] = "todo" + else: + task_create_data["status"] = status.lower() + + # Create TaskCreate object and validate + task_in = TaskCreate(**task_create_data) + + # Create task in database with current user as owner + db_task = task_crud.task.create_with_owner( + db=db, obj_in=task_in, user_id=current_user.id + ) + + # Add created task to response + created_tasks.append(TaskRead.model_validate(db_task)) + + except Exception as e: + logger.error(f"Error creating task: {e}") + # Continue with other tasks if one fails + continue + + if not created_tasks: + # If no tasks were successfully created + response.processing_successful = False + response.error = ChatProcessingError( + error_type="creation_error", + error_detail="Could not create any tasks from your message", + ) + else: + # Add created tasks to response + response.tasks = created_tasks + + return response + + except Exception as e: + logger.exception(f"Error in chat-to-tasks endpoint: {e}") + response.processing_successful = False + response.error = ChatProcessingError( + error_type="processing_error", + error_detail=f"An error occurred while processing your request: {str(e)}", + ) + return response \ No newline at end of file diff --git a/app/core/config.py b/app/core/config.py index 6149543..78ca340 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -77,12 +77,12 @@ else: class Settings(BaseSettings): + # Application settings PROJECT_NAME: str = "Task Manager API" - # No API version prefix - use direct paths API_PREFIX: str = "" SECRET_KEY: str = secrets.token_urlsafe(32) - # 60 minutes * 24 hours * 8 days = 8 days - ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8 + ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8 # 8 days + ENVIRONMENT: str = "development" # CORS Configuration BACKEND_CORS_ORIGINS: List[AnyHttpUrl] = [] @@ -98,6 +98,18 @@ class Settings(BaseSettings): # Database configuration SQLALCHEMY_DATABASE_URL: str = f"sqlite:///{DB_DIR}/db.sqlite" + # LLM provider settings - defaults to Mock service if not configured + # Options: "openai", "gemini", "mock" + LLM_PROVIDER: str = os.environ.get("LLM_PROVIDER", "mock") + + # OpenAI settings + OPENAI_API_KEY: str = os.environ.get("OPENAI_API_KEY", "") + OPENAI_MODEL: str = os.environ.get("OPENAI_MODEL", "gpt-3.5-turbo") + + # Google Gemini settings + GEMINI_API_KEY: str = os.environ.get("GEMINI_API_KEY", "") + GEMINI_MODEL: str = os.environ.get("GEMINI_MODEL", "gemini-pro") + model_config = {"case_sensitive": True} diff --git a/app/schemas/chat.py b/app/schemas/chat.py new file mode 100644 index 0000000..bb6af36 --- /dev/null +++ b/app/schemas/chat.py @@ -0,0 +1,44 @@ +""" +Pydantic schemas for the Chat-to-Tasks feature. +""" +from typing import List, Optional + +from pydantic import BaseModel, Field + +from app.schemas.task import TaskRead + + +class ChatInput(BaseModel): + """Schema for chat input from user.""" + + message: str = Field( + ..., + description="Natural language input describing tasks to be created", + min_length=3, + max_length=2000, + ) + + +class ChatProcessingError(BaseModel): + """Schema for error details when processing chat.""" + + error_type: str = Field(..., description="Type of error encountered") + error_detail: str = Field(..., description="Detailed error information") + + +class ChatResponse(BaseModel): + """Schema for chat response with parsed tasks.""" + + original_message: str = Field(..., description="Original user message") + tasks: List[TaskRead] = Field( + default_factory=list, + description="Tasks extracted from the message", + ) + processing_successful: bool = Field( + default=True, + description="Indicates if processing was successful", + ) + error: Optional[ChatProcessingError] = Field( + default=None, + description="Error details if processing was not successful", + ) \ No newline at end of file diff --git a/app/schemas/task.py b/app/schemas/task.py index ba425d6..e147ead 100644 --- a/app/schemas/task.py +++ b/app/schemas/task.py @@ -61,4 +61,10 @@ class TaskInDBBase(TaskBase): class Task(TaskInDBBase): + """Schema for tasks returned from database operations.""" pass + + +class TaskRead(TaskInDBBase): + """Schema for task data returned to clients.""" + user_id: Optional[int] = None diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..23f4570 --- /dev/null +++ b/app/services/__init__.py @@ -0,0 +1,3 @@ +""" +Service module initialization. +""" \ No newline at end of file diff --git a/app/services/llm_service.py b/app/services/llm_service.py new file mode 100644 index 0000000..1ec9fb3 --- /dev/null +++ b/app/services/llm_service.py @@ -0,0 +1,211 @@ +""" +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() \ No newline at end of file diff --git a/main.py b/main.py index d5a60f0..3b0aed4 100644 --- a/main.py +++ b/main.py @@ -1,17 +1,22 @@ import sys import os +import traceback +import datetime +import sqlite3 from pathlib import Path from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse +from sqlalchemy import text # Add project root to Python path for imports in alembic migrations project_root = Path(__file__).parent.absolute() sys.path.insert(0, str(project_root)) -from app.api.routers import api_router -from app.core.config import settings -from app.db import init_db +# Import app modules after setting up project path +from app.api.routers import api_router # noqa: E402 +from app.core.config import settings # noqa: E402 +from app.db import init_db # noqa: E402 # Initialize the database on startup print("Starting database initialization...") @@ -97,7 +102,6 @@ async def global_exception_handler(request: Request, exc: Exception): # Add SQLite diagnostic check try: - import sqlite3 from app.db.session import db_file # Try basic SQLite operations @@ -113,8 +117,6 @@ async def global_exception_handler(request: Request, exc: Exception): task_table_exists = cursor.fetchone() is not None # Get file info - import os - file_exists = os.path.exists(db_file) file_size = os.path.getsize(db_file) if file_exists else 0 @@ -159,6 +161,9 @@ def api_info(): "test-token": "/auth/test-token", }, "tasks": "/tasks", + "chat": { + "chat_to_tasks": "/chat/chat-to-tasks", + }, "docs": "/docs", "redoc": "/redoc", "health": "/health", @@ -181,11 +186,6 @@ def test_db_connection(): Test database connection and table creation """ try: - import os - import sqlite3 - import traceback - import datetime - from sqlalchemy import text from app.db.session import engine, db_file from app.core.config import DB_DIR diff --git a/requirements.txt b/requirements.txt index 9b3edd1..0e83776 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,4 +9,8 @@ ruff>=0.1.3 passlib>=1.7.4 bcrypt>=4.0.1 python-jose>=3.3.0 -email-validator>=2.0.0 \ No newline at end of file +email-validator>=2.0.0 +# LLM libraries for chat-to-tasks feature +openai>=1.6.0 +google-generativeai>=0.3.0 +httpx>=0.25.0 \ No newline at end of file