🤖 OpenWebUI
Chat with AI models powered by Ollama on Hugging Face Spaces
# app.py from flask import Flask, request, jsonify, render_template_string, Response import os import requests import json import logging from logging.handlers import RotatingFileHandler from typing import Dict, Any, List import time import re app = Flask(__name__) # Configure logging with file output in a writable directory log_dir = '/app/logs' log_path = os.path.join(log_dir, 'openwebui.log') try: os.makedirs(log_dir, exist_ok=True) log_handler = RotatingFileHandler(log_path, maxBytes=1000000, backupCount=5) log_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')) logging.getLogger().addHandler(log_handler) except Exception as e: logging.getLogger().addHandler(logging.StreamHandler()) # Fallback to stderr logging.error(f"Failed to initialize file logging: {e}") logging.getLogger().setLevel(logging.INFO) # Configuration OLLAMA_BASE_URL = os.getenv('OLLAMA_BASE_URL', 'https://huggingface.co/spaces/tommytracx/ollama-api') ALLOWED_MODELS = os.getenv('ALLOWED_MODELS', 'llama2,llama2:13b,llama2:70b,codellama,neural-chat,gemma-3-270m').split(',') MAX_TOKENS = int(os.getenv('MAX_TOKENS', '2048')) TEMPERATURE = float(os.getenv('TEMPERATURE', '0.7')) class OllamaManager: def __init__(self, base_url: str): self.base_url = base_url.rstrip('/') self.available_models = ALLOWED_MODELS self.refresh_models() def refresh_models(self) -> None: """Refresh the list of available models from Ollama API, falling back to allowed models.""" try: response = requests.get(f"{self.base_url}/api/tags", timeout=10) response.raise_for_status() data = response.json() models = [model['name'] for model in data.get('models', [])] self.available_models = [model for model in models if model in ALLOWED_MODELS] if not self.available_models: self.available_models = ALLOWED_MODELS logging.warning("No allowed models found in API response, using ALLOWED_MODELS") logging.info(f"Available models: {self.available_models}") except requests.exceptions.ConnectionError as e: logging.error(f"Connection error refreshing models: {e}") self.available_models = ALLOWED_MODELS except requests.exceptions.HTTPError as e: logging.error(f"HTTP error refreshing models: {e}") self.available_models = ALLOWED_MODELS except Exception as e: logging.error(f"Unexpected error refreshing models: {e}") self.available_models = ALLOWED_MODELS def list_models(self) -> List[str]: """Return the list of available models.""" return self.available_models def generate(self, model_name: str, prompt: str, stream: bool = False, **kwargs) -> Any: """Generate text using a model, with optional streaming.""" if model_name not in self.available_models: logging.warning(f"Attempted to generate with unavailable model: {model_name}") return {"status": "error", "message": f"Model {model_name} not available"} try: payload = { "model": model_name, "prompt": prompt, "stream": stream, **kwargs } if stream: response = requests.post(f"{self.base_url}/api/generate", json=payload, stream=True, timeout=120) response.raise_for_status() return response else: response = requests.post(f"{self.base_url}/api/generate", json=payload, timeout=120) response.raise_for_status() data = response.json() logging.info(f"Generated response with model {model_name}") return { "status": "success", "response": data.get('response', ''), "model": model_name, "usage": data.get('usage', {}) } except requests.exceptions.ConnectionError as e: logging.error(f"Connection error generating response: {e}") return {"status": "error", "message": f"Connection error: {str(e)}"} except requests.exceptions.HTTPError as e: logging.error(f"HTTP error generating response: {e}") return {"status": "error", "message": f"HTTP error: {str(e)}"} except Exception as e: logging.error(f"Unexpected error generating response: {e}") return {"status": "error", "message": str(e)} def health_check(self) -> Dict[str, Any]: """Check the health of the Ollama API.""" try: response = requests.get(f"{self.base_url}/api/tags", timeout=10) response.raise_for_status() logging.info("Health check successful") return {"status": "healthy", "available_models": len(self.available_models)} except requests.exceptions.ConnectionError as e: logging.error(f"Health check connection error: {e}") return {"status": "unhealthy", "error": f"Connection error: {str(e)}"} except requests.exceptions.HTTPError as e: logging.error(f"Health check HTTP error: {e}") return {"status": "unhealthy", "error": f"HTTP error: {str(e)}"} except Exception as e: logging.error(f"Health check unexpected error: {e}") return {"status": "unhealthy", "error": str(e)} # Initialize Ollama manager ollama_manager = OllamaManager(OLLAMA_BASE_URL) # HTML template for the chat interface with comprehensive, mobile-optimized UI HTML_TEMPLATE = '''
Chat with AI models powered by Ollama on Hugging Face Spaces