#!/usr/bin/env python3
"""
Textilindo AI Assistant - Hugging Face Spaces FastAPI Application
Simplified version for HF Spaces deployment
"""
import os
import json
import logging
from pathlib import Path
from datetime import datetime
from typing import Optional, Dict, Any
from fastapi import FastAPI, HTTPException, Request, BackgroundTasks
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import uvicorn
import requests
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Initialize FastAPI app
app = FastAPI(
title="Textilindo AI Assistant",
description="AI Assistant for Textilindo textile company",
version="1.0.0"
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Request/Response models
class ChatRequest(BaseModel):
message: str
conversation_id: Optional[str] = None
class ChatResponse(BaseModel):
response: str
conversation_id: str
status: str = "success"
class HealthResponse(BaseModel):
status: str
message: str
version: str = "1.0.0"
class TrainingRequest(BaseModel):
model_name: str = "distilgpt2"
dataset_path: str = "data/lora_dataset_20250910_145055.jsonl"
config_path: str = "configs/training_config.yaml"
max_samples: int = 20
epochs: int = 1
batch_size: int = 1
learning_rate: float = 5e-5
class TrainingResponse(BaseModel):
success: bool
message: str
training_id: str
status: str
# Training status storage
training_status = {
"is_training": False,
"progress": 0,
"status": "idle",
"current_step": 0,
"total_steps": 0,
"loss": 0.0,
"start_time": None,
"end_time": None,
"error": None
}
class TextilindoAI:
"""Textilindo AI Assistant using HuggingFace Inference API"""
def __init__(self):
self.api_key = os.getenv('HUGGINGFACE_API_KEY')
# Use Llama model for better performance
self.model = os.getenv('DEFAULT_MODEL', 'meta-llama/Llama-3.1-8B-Instruct')
self.system_prompt = self.load_system_prompt()
if not self.api_key:
logger.warning("HUGGINGFACE_API_KEY not found. Using mock responses.")
self.client = None
else:
try:
from huggingface_hub import InferenceClient
self.client = InferenceClient(
token=self.api_key,
model=self.model
)
logger.info(f"Initialized with model: {self.model}")
except Exception as e:
logger.error(f"Failed to initialize InferenceClient: {e}")
self.client = None
def load_system_prompt(self) -> str:
"""Load system prompt from config file"""
try:
prompt_path = Path("configs/system_prompt.md")
if prompt_path.exists():
with open(prompt_path, 'r', encoding='utf-8') as f:
content = f.read()
# Extract system prompt from markdown
if 'SYSTEM_PROMPT = """' in content:
start = content.find('SYSTEM_PROMPT = """') + len('SYSTEM_PROMPT = """')
end = content.find('"""', start)
return content[start:end].strip()
else:
# Fallback: use entire content
return content.strip()
else:
return self.get_default_system_prompt()
except Exception as e:
logger.error(f"Error loading system prompt: {e}")
return self.get_default_system_prompt()
def get_default_system_prompt(self) -> str:
"""Default system prompt if file not found"""
return """You are a friendly and helpful AI assistant for Textilindo, a textile company.
Always respond in Indonesian (Bahasa Indonesia).
Keep responses short and direct.
Be friendly and helpful.
Use exact information from the knowledge base.
The company uses yards for sales.
Minimum purchase is 1 roll (67-70 yards)."""
def generate_response(self, user_message: str) -> str:
"""Generate response using HuggingFace Inference API"""
if not self.client:
logger.warning("No HuggingFace client available, using mock response")
return self.get_mock_response(user_message)
try:
# Use Llama conversation format
prompt = f"<|begin_of_text|><|start_header_id|>user<|end_header_id|>\n\n{user_message}<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n\n"
logger.info(f"Using model: {self.model}")
logger.info(f"API Key present: {bool(self.api_key)}")
logger.info(f"Generating response for prompt: {prompt[:100]}...")
# Generate response with Llama parameters
response = self.client.text_generation(
prompt,
max_new_tokens=200,
temperature=0.7,
top_p=0.9,
top_k=40,
repetition_penalty=1.1,
do_sample=True,
stop_sequences=["<|eot_id|>", "<|end_of_text|>", "User:", "Assistant:"]
)
logger.info(f"Raw AI response: {response[:200]}...")
# Clean up the response for Llama
assistant_response = response.strip()
# Remove Llama special tokens
assistant_response = assistant_response.replace("<|eot_id|>", "").replace("<|end_of_text|>", "")
assistant_response = assistant_response.replace("<|begin_of_text|>", "").replace("<|start_header_id|>", "")
assistant_response = assistant_response.replace("<|end_header_id|>", "").replace("user", "").replace("assistant", "")
# Clean up any remaining formatting
assistant_response = assistant_response.strip()
# If response is too short, use mock response
if len(assistant_response) < 10:
logger.warning("AI response too short, using mock response")
return self.get_mock_response(user_message)
logger.info(f"Cleaned AI response: {assistant_response[:100]}...")
# If response is too short or generic, use mock response
if len(assistant_response) < 10 or "I don't know" in assistant_response.lower():
logger.warning("AI response too short, using mock response")
return self.get_mock_response(user_message)
# For testing: if it's a non-Textilindo question, return the AI response directly
if not any(keyword in user_message.lower() for keyword in ['textilindo', 'lokasi', 'jam', 'katalog', 'produk', 'sample', 'pembelian', 'pembayaran', 'ongkir']):
logger.info("Non-Textilindo question detected, returning AI response directly")
return assistant_response
return assistant_response
except Exception as e:
logger.error(f"Error generating response: {e}")
return self.get_mock_response(user_message)
def get_mock_response(self, user_message: str) -> str:
"""Enhanced mock responses with better context awareness"""
mock_responses = {
"dimana lokasi textilindo": "Textilindo berkantor pusat di Jl. Raya Prancis No.39, Kosambi Tim., Kec. Kosambi, Kabupaten Tangerang, Banten 15213",
"jam berapa textilindo beroperasional": "Jam operasional Senin-Jumat 08:00-17:00, Sabtu 08:00-12:00.",
"berapa ketentuan pembelian": "Minimal order 1 roll per jenis kain",
"bagaimana dengan pembayarannya": "Pembayaran dapat dilakukan via transfer bank atau cash on delivery",
"apa ada gratis ongkir": "Gratis ongkir untuk order minimal 5 roll.",
"apa bisa dikirimkan sample": "hallo kak untuk sampel kita bisa kirimkan gratis ya kak 😊",
"katalog": "Katalog produk Textilindo tersedia dalam bentuk Buku, PDF, atau Katalog Website.",
"harga": "Harga kain berbeda-beda tergantung jenis kainnya. Untuk informasi lengkap bisa hubungi admin kami.",
"produk": "Kami menjual berbagai jenis kain woven dan knitting. Ada rayon twill, baby doll, voal, dan masih banyak lagi.",
"what is 2+2": "2 + 2 = 4",
"what is the capital of france": "The capital of France is Paris.",
"explain machine learning": "Machine learning is a subset of artificial intelligence that enables computers to learn and make decisions from data without being explicitly programmed.",
"write a poem": "Here's a short poem:\n\nIn lines of code we find our way,\nThrough logic's maze we play,\nEach function calls, each loop runs true,\nCreating something bright and new.",
"hello": "Hello! I'm the Textilindo AI assistant. How can I help you today?",
"hi": "Hi there! I'm here to help with any questions about Textilindo. What would you like to know?",
"how are you": "I'm doing well, thank you for asking! I'm ready to help you with any questions about Textilindo's products and services.",
"thank you": "You're welcome! I'm happy to help. Is there anything else you'd like to know about Textilindo?",
"goodbye": "Goodbye! Thank you for chatting with me. Have a great day!",
"bye": "Bye! Feel free to come back anytime if you have more questions about Textilindo."
}
# More specific keyword matching
user_lower = user_message.lower()
# Check for exact phrase matches first
for key, response in mock_responses.items():
if key in user_lower:
return response
# Check for specific keywords with better matching
if any(word in user_lower for word in ["lokasi", "alamat", "dimana"]):
return "Textilindo berkantor pusat di Jl. Raya Prancis No.39, Kosambi Tim., Kec. Kosambi, Kabupaten Tangerang, Banten 15213"
elif any(word in user_lower for word in ["jam", "buka", "operasional"]):
return "Jam operasional Senin-Jumat 08:00-17:00, Sabtu 08:00-12:00."
elif any(word in user_lower for word in ["pembelian", "beli", "order"]):
return "Minimal order 1 roll per jenis kain"
elif any(word in user_lower for word in ["pembayaran", "bayar", "payment"]):
return "Pembayaran dapat dilakukan via transfer bank atau cash on delivery"
elif any(word in user_lower for word in ["ongkir", "ongkos", "kirim"]):
return "Gratis ongkir untuk order minimal 5 roll."
elif any(word in user_lower for word in ["sample", "sampel", "contoh"]):
return "hallo kak untuk sampel kita bisa kirimkan gratis ya kak 😊"
elif any(word in user_lower for word in ["katalog", "katalog"]):
return "Katalog produk Textilindo tersedia dalam bentuk Buku, PDF, atau Katalog Website."
elif any(word in user_lower for word in ["harga", "price", "cost"]):
return "Harga kain berbeda-beda tergantung jenis kainnya. Untuk informasi lengkap bisa hubungi admin kami."
elif any(word in user_lower for word in ["produk", "kain", "bahan"]):
return "Kami menjual berbagai jenis kain woven dan knitting. Ada rayon twill, baby doll, voal, dan masih banyak lagi."
elif any(word in user_lower for word in ["math", "mathematics", "calculate", "addition", "subtraction", "multiplication", "division"]):
return "I can help with basic math questions! Please ask me a specific math problem and I'll do my best to help."
elif any(word in user_lower for word in ["capital", "country", "geography", "world"]):
return "I can help with geography questions! Please ask me about a specific country or capital city."
elif any(word in user_lower for word in ["technology", "ai", "artificial intelligence", "machine learning", "programming", "coding"]):
return "I'd be happy to discuss technology topics! Please ask me a specific question about AI, programming, or technology."
elif any(word in user_lower for word in ["poem", "poetry", "creative", "write"]):
return "I enjoy creative writing! I can help with poems, stories, or other creative content. What would you like me to write about?"
elif any(word in user_lower for word in ["hello", "hi", "hey", "greetings"]):
return "Hello! I'm the Textilindo AI assistant. I'm here to help with questions about our products and services, or just have a friendly conversation!"
elif any(word in user_lower for word in ["how are you", "how do you do", "how's it going"]):
return "I'm doing great, thank you for asking! I'm ready to help you with any questions about Textilindo or just chat about anything you'd like."
elif any(word in user_lower for word in ["thank you", "thanks", "appreciate"]):
return "You're very welcome! I'm happy to help. Is there anything else you'd like to know about Textilindo or anything else I can assist you with?"
elif any(word in user_lower for word in ["goodbye", "bye", "see you", "farewell"]):
return "Goodbye! It was great chatting with you. Feel free to come back anytime if you have more questions about Textilindo or just want to chat!"
return "Halo! Saya adalah asisten AI Textilindo. Saya bisa membantu Anda dengan pertanyaan tentang produk dan layanan kami, atau sekadar mengobrol! Bagaimana saya bisa membantu Anda hari ini? 😊"
# Initialize AI assistant
ai_assistant = TextilindoAI()
# Routes
@app.get("/", response_class=HTMLResponse)
async def root():
"""Serve the main chat interface"""
try:
with open("templates/chat.html", "r", encoding="utf-8") as f:
return HTMLResponse(content=f.read())
except FileNotFoundError:
return HTMLResponse(content="""
Textilindo AI Assistant
🤖 Textilindo AI Assistant
Asisten AI untuk membantu pertanyaan tentang Textilindo
Halo! Saya adalah asisten AI Textilindo. Saya bisa membantu Anda dengan informasi tentang:
• Lokasi dan jam operasional
• Produk dan katalog
• Ketentuan pembelian
• Pengiriman dan pembayaran
Silakan tanyakan apa yang ingin Anda ketahui! 😊
Siap membantu Anda
""")
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
"""Chat endpoint"""
try:
response = ai_assistant.generate_response(request.message)
return ChatResponse(
response=response,
conversation_id=request.conversation_id or "default",
status="success"
)
except Exception as e:
logger.error(f"Error in chat endpoint: {e}")
raise HTTPException(status_code=500, detail="Internal server error")
@app.get("/health", response_model=HealthResponse)
async def health_check():
"""Health check endpoint"""
return HealthResponse(
status="healthy",
message="Textilindo AI Assistant is running",
version="1.0.0"
)
@app.get("/info")
async def get_info():
"""Get application information"""
return {
"name": "Textilindo AI Assistant",
"version": "1.0.0",
"model": ai_assistant.model,
"has_api_key": bool(ai_assistant.api_key),
"client_initialized": bool(ai_assistant.client),
"endpoints": {
"training": {
"start": "POST /api/train/start",
"status": "GET /api/train/status",
"data": "GET /api/train/data",
"gpu": "GET /api/train/gpu",
"test": "POST /api/train/test"
},
"chat": {
"chat": "POST /chat",
"health": "GET /health"
}
}
}
# Training API endpoints (simplified for HF Spaces)
@app.post("/api/train/start", response_model=TrainingResponse)
async def start_training(request: TrainingRequest, background_tasks: BackgroundTasks):
"""Start training process (simplified for HF Spaces)"""
global training_status
if training_status["is_training"]:
raise HTTPException(status_code=400, detail="Training already in progress")
# For HF Spaces, we'll simulate training
training_id = f"train_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
# Update status to show training started
training_status.update({
"is_training": True,
"status": "started",
"progress": 0,
"start_time": datetime.now().isoformat(),
"error": None
})
# Simulate training completion after a delay
background_tasks.add_task(simulate_training_completion)
return TrainingResponse(
success=True,
message="Training started successfully (simulated for HF Spaces)",
training_id=training_id,
status="started"
)
async def simulate_training_completion():
"""Simulate training completion for HF Spaces"""
import asyncio
await asyncio.sleep(10) # Simulate 10 seconds of training
global training_status
training_status.update({
"is_training": False,
"status": "completed",
"progress": 100,
"end_time": datetime.now().isoformat()
})
@app.get("/api/train/status")
async def get_training_status():
"""Get current training status"""
return training_status
@app.get("/api/train/data")
async def get_training_data_info():
"""Get information about available training data"""
data_dir = Path("data")
if not data_dir.exists():
return {"files": [], "count": 0}
jsonl_files = list(data_dir.glob("*.jsonl"))
files_info = []
for file in jsonl_files:
try:
with open(file, 'r', encoding='utf-8') as f:
lines = f.readlines()
files_info.append({
"name": file.name,
"size": file.stat().st_size,
"lines": len(lines)
})
except Exception as e:
files_info.append({
"name": file.name,
"error": str(e)
})
return {
"files": files_info,
"count": len(jsonl_files)
}
@app.get("/api/train/gpu")
async def get_gpu_info():
"""Get GPU information (simulated for HF Spaces)"""
return {
"available": False,
"message": "GPU not available in HF Spaces free tier",
"recommendation": "Use local training or upgrade to paid tier"
}
@app.post("/api/train/test")
async def test_trained_model():
"""Test the trained model (simulated)"""
return {
"success": True,
"message": "Model testing simulated for HF Spaces",
"test_prompt": "dimana lokasi textilindo?",
"response": "Textilindo berkantor pusat di Jl. Raya Prancis No.39, Kosambi Tim., Kec. Kosambi, Kabupaten Tangerang, Banten 15213",
"note": "This is a simulated response for HF Spaces demo"
}
@app.post("/api/test/ai")
async def test_ai_directly(request: ChatRequest):
"""Test AI directly without fallback to mock responses"""
try:
if not ai_assistant.client:
return {
"success": False,
"message": "No HuggingFace client available",
"response": None
}
# Test with a simple prompt
test_prompt = f"User: {request.message}\nAssistant:"
logger.info(f"Testing AI with prompt: {test_prompt}")
response = ai_assistant.client.text_generation(
test_prompt,
max_new_tokens=100,
temperature=0.7,
top_p=0.9,
top_k=40
)
logger.info(f"Direct AI response: {response}")
return {
"success": True,
"message": "AI response generated successfully",
"raw_response": response,
"model": ai_assistant.model,
"api_key_available": bool(ai_assistant.api_key)
}
except Exception as e:
logger.error(f"Error in direct AI test: {e}")
return {
"success": False,
"message": f"Error: {str(e)}",
"response": None
}
if __name__ == "__main__":
# Get port from environment variable (Hugging Face Spaces uses 7860)
port = int(os.getenv("PORT", 7860))
# Run the application
uvicorn.run(
"app:app",
host="0.0.0.0",
port=port,
log_level="info"
)