Create app.py

app.py

@@ -0,0 +1,361 @@
+# app.py - General API endpoint
+from fastapi import FastAPI, HTTPException, Depends, Header
+from pydantic import BaseModel
+from typing import Optional, List, Dict, Any
+import logging
+from transformers import pipeline
+import uvicorn
+import os
+from datetime import datetime
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Initialize FastAPI app
+app = FastAPI(
+    title="AI Chat API for n8n",
+    description="General AI processing API that accepts prompts from n8n workflows",
+    version="1.0.0"
+)
+
+# Request/Response models
+class PromptRequest(BaseModel):
+    """Request model for prompt processing"""
+    prompt: str  # User's instruction/query
+    content: Optional[str] = None  # Optional content to process
+    parameters: Optional[Dict[str, Any]] = None  # Optional parameters
+    task_type: Optional[str] = None  # Optional: summarize, generate, classify, etc.
+    max_length: Optional[int] = 200
+    temperature: Optional[float] = 0.7
+    return_type: Optional[str] = "text"  # text, json, list
+    
+class PromptResponse(BaseModel):
+    """Response model"""
+    success: bool
+    result: Optional[Any] = None
+    error: Optional[str] = None
+    processing_time: Optional[float] = None
+    model_used: Optional[str] = None
+
+class BatchRequest(BaseModel):
+    """Batch processing request"""
+    prompts: List[PromptRequest]
+    parallel: Optional[bool] = False
+
+# Initialize models
+class AIModelManager:
+    """Manages AI models dynamically"""
+    def __init__(self):
+        self.models = {}
+        self.load_models()
+    
+    def load_models(self):
+        """Load essential models"""
+        try:
+            # Load a general text generation model
+            self.models["text-generation"] = pipeline(
+                "text-generation",
+                model="gpt2",
+                max_length=200,
+                device=-1  # CPU by default
+            )
+            
+            # Load summarization model
+            self.models["summarization"] = pipeline(
+                "summarization",
+                model="facebook/bart-large-cnn",
+                device=-1
+            )
+            
+            # Load text classification for intent detection
+            self.models["text-classification"] = pipeline(
+                "text-classification",
+                model="distilbert-base-uncased-finetuned-sst-2-english",
+                device=-1
+            )
+            
+            logger.info("Models loaded successfully")
+            
+        except Exception as e:
+            logger.error(f"Error loading models: {e}")
+            # Create dummy models for testing
+            self.models = {}
+    
+    def process_prompt(self, prompt: str, content: str = None, **kwargs) -> str:
+        """
+        General prompt processing method
+        Args:
+            prompt: Instruction/query from user
+            content: Optional content to process
+            **kwargs: Additional parameters
+        """
+        try:
+            # Combine prompt and content
+            full_input = prompt
+            if content:
+                full_input = f"{prompt}\n\nContent: {content}"
+            
+            # Determine task type from prompt
+            task_type = self._detect_task_type(prompt, content)
+            
+            # Process based on task type
+            if task_type == "summarize" and content:
+                return self._process_summarization(content, **kwargs)
+            
+            elif task_type == "generate":
+                return self._process_generation(full_input, **kwargs)
+            
+            elif task_type == "classify" and content:
+                return self._process_classification(content, **kwargs)
+            
+            else:
+                # Default: general text generation
+                return self._process_generation(full_input, **kwargs)
+                
+        except Exception as e:
+            logger.error(f"Error processing prompt: {e}")
+            return f"Error processing your request: {str(e)}"
+    
+    def _detect_task_type(self, prompt: str, content: str = None) -> str:
+        """Detect task type from prompt"""
+        prompt_lower = prompt.lower()
+        
+        task_keywords = {
+            "summarize": ["summarize", "summary", "brief", "overview"],
+            "generate": ["generate", "create", "write", "make", "draft"],
+            "classify": ["classify", "categorize", "label", "tag"],
+            "translate": ["translate", "convert language"],
+            "analyze": ["analyze", "analyze", "evaluate", "assess"]
+        }
+        
+        for task, keywords in task_keywords.items():
+            if any(keyword in prompt_lower for keyword in keywords):
+                return task
+        
+        return "general"
+    
+    def _process_summarization(self, content: str, **kwargs) -> str:
+        """Process summarization task"""
+        if "summarization" in self.models:
+            max_length = kwargs.get("max_length", 150)
+            min_length = kwargs.get("min_length", 30)
+            
+            result = self.models["summarization"](
+                content,
+                max_length=max_length,
+                min_length=min_length,
+                do_sample=False
+            )
+            return result[0]['summary_text']
+        else:
+            # Fallback
+            sentences = content.split('. ')
+            if len(sentences) > 3:
+                return '. '.join(sentences[:2]) + '.'
+            return content[:100] + "..."
+    
+    def _process_generation(self, prompt: str, **kwargs) -> str:
+        """Process text generation task"""
+        if "text-generation" in self.models:
+            max_length = kwargs.get("max_length", 100)
+            temperature = kwargs.get("temperature", 0.7)
+            
+            result = self.models["text-generation"](
+                prompt,
+                max_length=max_length,
+                temperature=temperature,
+                num_return_sequences=1
+            )
+            return result[0]['generated_text']
+        else:
+            # Fallback response
+            return f"Processed: {prompt[:50]}... [Model not loaded]"
+    
+    def _process_classification(self, content: str, **kwargs) -> str:
+        """Process classification task"""
+        if "text-classification" in self.models:
+            result = self.models["text-classification"](content)
+            return str(result)
+        else:
+            return "Classification model not available"
+
+# Initialize model manager
+model_manager = AIModelManager()
+
+# API Endpoints
+@app.get("/")
+async def root():
+    """Root endpoint"""
+    return {
+        "status": "online",
+        "service": "AI Chat API for n8n",
+        "endpoints": {
+            "/process": "Process single prompt (POST)",
+            "/batch": "Process multiple prompts (POST)",
+            "/health": "Health check (GET)",
+            "/models": "List loaded models (GET)"
+        }
+    }
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint"""
+    return {
+        "status": "healthy",
+        "timestamp": datetime.now().isoformat(),
+        "models_loaded": len(model_manager.models) > 0
+    }
+
+@app.get("/models")
+async def list_models():
+    """List loaded models"""
+    return {
+        "models": list(model_manager.models.keys()),
+        "count": len(model_manager.models)
+    }
+
+@app.post("/process", response_model=PromptResponse)
+async def process_prompt(request: PromptRequest):
+    """
+    Main endpoint for processing prompts from n8n
+    """
+    start_time = datetime.now()
+    
+    try:
+        logger.info(f"Processing prompt: {request.prompt[:50]}...")
+        
+        # Process the prompt
+        result = model_manager.process_prompt(
+            prompt=request.prompt,
+            content=request.content,
+            max_length=request.max_length,
+            temperature=request.temperature
+        )
+        
+        processing_time = (datetime.now() - start_time).total_seconds()
+        
+        return PromptResponse(
+            success=True,
+            result=result,
+            processing_time=processing_time,
+            model_used="text-generation"  # You can make this dynamic
+        )
+        
+    except Exception as e:
+        logger.error(f"Error in process_prompt: {e}")
+        return PromptResponse(
+            success=False,
+            error=str(e),
+            processing_time=(datetime.now() - start_time).total_seconds()
+        )
+
+@app.post("/batch", response_model=List[PromptResponse])
+async def process_batch(request: BatchRequest):
+    """
+    Process multiple prompts in batch
+    """
+    responses = []
+    
+    for prompt_req in request.prompts:
+        start_time = datetime.now()
+        
+        try:
+            result = model_manager.process_prompt(
+                prompt=prompt_req.prompt,
+                content=prompt_req.content,
+                max_length=prompt_req.max_length,
+                temperature=prompt_req.temperature
+            )
+            
+            responses.append(PromptResponse(
+                success=True,
+                result=result,
+                processing_time=(datetime.now() - start_time).total_seconds()
+            ))
+            
+        except Exception as e:
+            responses.append(PromptResponse(
+                success=False,
+                error=str(e),
+                processing_time=(datetime.now() - start_time).total_seconds()
+            ))
+    
+    return responses
+
+# Webhook endpoint (for n8n webhook node)
+@app.post("/webhook")
+async def webhook_endpoint(
+    payload: Dict[str, Any],
+    x_n8n_signature: Optional[str] = Header(None)
+):
+    """
+    Webhook endpoint specifically for n8n
+    """
+    logger.info(f"Webhook received from n8n: {payload.keys()}")
+    
+    # Extract prompt from n8n payload
+    prompt = payload.get("prompt") or payload.get("text") or payload.get("message")
+    content = payload.get("content") or payload.get("data")
+    
+    if not prompt:
+        raise HTTPException(status_code=400, detail="No prompt provided in payload")
+    
+    # Process the prompt
+    result = model_manager.process_prompt(prompt, content)
+    
+    # Return in n8n-friendly format
+    return {
+        "success": True,
+        "response": result,
+        "timestamp": datetime.now().isoformat(),
+        "webhook_id": payload.get("webhookId"),
+        "workflow_id": payload.get("workflowId")
+    }
+
+# Async task endpoint
+@app.post("/async")
+async def create_async_task(request: PromptRequest):
+    """
+    Create an async task (returns task ID immediately)
+    """
+    task_id = f"task_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
+    
+    # In production, you'd queue this task
+    return {
+        "task_id": task_id,
+        "status": "queued",
+        "message": "Task created successfully"
+    }
+
+@app.get("/task/{task_id}")
+async def get_task_status(task_id: str):
+    """
+    Check status of async task
+    """
+    return {
+        "task_id": task_id,
+        "status": "completed",  # Mock response
+        "result": "This is a mock result for async task"
+    }
+
+# For Hugging Face Spaces
+@app.get("/hf_space")
+async def hf_space_endpoint(prompt: str = None, content: str = None):
+    """
+    Simple endpoint for Hugging Face Spaces demo
+    """
+    if not prompt:
+        return {"error": "Please provide a prompt parameter"}
+    
+    result = model_manager.process_prompt(prompt, content)
+    
+    return {
+        "prompt": prompt,
+        "response": result,
+        "content_length": len(content) if content else 0
+    }
+
+if __name__ == "__main__":
+    port = int(os.getenv("PORT", 8000))
+    uvicorn.run(app, host="0.0.0.0", port=port)