{
"tool_response": {
"type": "update",
"filePath": "\/var\/www\/migration\/content-pipeline\/src\/model_registry.py",
"content": "\"\"\"\nModel Registry - Configuration-based AI Model Management.\n\nProvides available models via environment configuration.\nNo database dependency - fully portable.\n\"\"\"\n\nimport os\nfrom typing import Optional\n\n\n# =============================================================================\n# DEFAULT MODELS (Override via environment)\n# =============================================================================\n\nDEFAULT_MODELS = {\n # Embedding models\n \"ollama:mxbai-embed-large:latest\": {\n \"provider\": \"ollama\",\n \"model_id\": \"mxbai-embed-large:latest\",\n \"display_name\": \"MXBai Embed Large (1024d)\",\n \"is_embedding\": True,\n \"is_chat\": False,\n \"is_vision\": False,\n \"context_length\": 512,\n },\n # Chat models - Ollama\n \"ollama:llama3.2:3b\": {\n \"provider\": \"ollama\",\n \"model_id\": \"llama3.2:3b\",\n \"display_name\": \"Llama 3.2 3B\",\n \"is_embedding\": False,\n \"is_chat\": True,\n \"is_vision\": False,\n \"context_length\": 8192,\n },\n \"ollama:mistral:latest\": {\n \"provider\": \"ollama\",\n \"model_id\": \"mistral:latest\",\n \"display_name\": \"Mistral 7B\",\n \"is_embedding\": False,\n \"is_chat\": True,\n \"is_vision\": False,\n \"context_length\": 8192,\n },\n # Vision models\n \"ollama:llava:latest\": {\n \"provider\": \"ollama\",\n \"model_id\": \"llava:latest\",\n \"display_name\": \"LLaVA Vision\",\n \"is_embedding\": False,\n \"is_chat\": True,\n \"is_vision\": True,\n \"context_length\": 4096,\n },\n # Anthropic models\n \"anthropic:claude-sonnet-4-20250514\": {\n \"provider\": \"anthropic\",\n \"model_id\": \"claude-sonnet-4-20250514\",\n \"display_name\": \"Claude Sonnet 4\",\n \"is_embedding\": False,\n \"is_chat\": True,\n \"is_vision\": True,\n \"context_length\": 200000,\n },\n}\n\n\nclass ModelRegistry:\n \"\"\"Central registry for all AI models.\"\"\"\n\n _instance: Optional[\"ModelRegistry\"] = None\n\n def __init__(self):\n self._models = self._load_models()\n\n @classmethod\n def get_instance(cls) -> \"ModelRegistry\":\n \"\"\"Get singleton instance.\"\"\"\n if cls._instance is None:\n cls._instance = cls()\n return cls._instance\n\n @classmethod\n def clear_cache(cls):\n \"\"\"Clear cached instance.\"\"\"\n cls._instance = None\n\n def _load_models(self) -> dict:\n \"\"\"Load models from defaults + environment overrides.\"\"\"\n models = dict(DEFAULT_MODELS)\n\n # Add custom models from environment\n # Format: PIPELINE_MODEL_<N>=provider:model_id:display_name\n for key, value in os.environ.items():\n if key.startswith(\"PIPELINE_MODEL_\"):\n try:\n parts = value.split(\":\", 2)\n if len(parts) >= 2:\n provider, model_id = parts[0], parts[1]\n display_name = parts[2] if len(parts) > 2 else model_id\n full_key = f\"{provider}:{model_id}\"\n models[full_key] = {\n \"provider\": provider,\n \"model_id\": model_id,\n \"display_name\": display_name,\n \"is_embedding\": False,\n \"is_chat\": True,\n \"is_vision\": False,\n \"context_length\": 8192,\n }\n except Exception:\n pass\n\n return models\n\n def get_chat_models(self) -> dict:\n \"\"\"Get all available chat models.\"\"\"\n return {k: v[\"display_name\"] for k, v in self._models.items() if v.get(\"is_chat\")}\n\n def get_vision_models(self) -> dict:\n \"\"\"Get all vision-capable models.\"\"\"\n return {k: v[\"display_name\"] for k, v in self._models.items() if v.get(\"is_vision\")}\n\n def get_embedding_models(self) -> dict:\n \"\"\"Get all embedding models.\"\"\"\n return {k: v[\"display_name\"] for k, v in self._models.items() if v.get(\"is_embedding\")}\n\n def get_model(self, full_key: str) -> dict | None:\n \"\"\"Get a single model by full_key.\"\"\"\n return self._models.get(full_key)\n\n def get_label(self, full_key: str) -> str:\n \"\"\"Get display label for a model.\"\"\"\n model = self.get_model(full_key)\n return model[\"display_name\"] if model else full_key\n\n def is_valid(self, full_key: str) -> bool:\n \"\"\"Check if model exists.\"\"\"\n return full_key in self._models\n\n def is_local(self, full_key: str) -> bool:\n \"\"\"Check if model is local (Ollama).\"\"\"\n return full_key.startswith(\"ollama:\")\n\n def get_default_chat_model(self) -> str:\n \"\"\"Get default chat model from environment or fallback.\"\"\"\n env_model = os.environ.get(\"OLLAMA_CHAT_MODEL\")\n if env_model:\n return f\"ollama:{env_model}\" if not env_model.startswith(\"ollama:\") else env_model\n # Fallback to first available Ollama chat model\n for key, model in self._models.items():\n if model.get(\"is_chat\") and model.get(\"provider\") == \"ollama\":\n return key\n return \"ollama:mistral:latest\"\n\n def get_default_embedding_model(self) -> str:\n \"\"\"Get default embedding model.\"\"\"\n env_model = os.environ.get(\"OLLAMA_EMBEDDING_MODEL\")\n if env_model:\n return f\"ollama:{env_model}:latest\"\n return \"ollama:mxbai-embed-large:latest\"\n\n def get_ollama_model_id(self, full_key: str) -> str:\n \"\"\"Extract Ollama model ID from full_key.\"\"\"\n if full_key.startswith(\"ollama:\"):\n return full_key[7:]\n return full_key\n\n\n# =============================================================================\n# CONVENIENCE FUNCTIONS\n# =============================================================================\n\ndef get_chat_model() -> str:\n \"\"\"Get the default chat model.\"\"\"\n return ModelRegistry.get_instance().get_default_chat_model()\n\n\ndef get_embedding_model() -> str:\n \"\"\"Get the default embedding model.\"\"\"\n return ModelRegistry.get_instance().get_default_embedding_model()\n\n\ndef get_model_id_for_ollama(full_key: str) -> str:\n \"\"\"Extract Ollama model ID from full_key.\"\"\"\n return ModelRegistry.get_instance().get_ollama_model_id(full_key)\n",
"structuredPatch": [
{
"oldStart": 1,
"oldLines": 25,
"newStart": 1,
"newLines": 78,
"lines": [
" \"\"\"",
"-Model Registry - Single Source of Truth for AI Models.",
"+Model Registry - Configuration-based AI Model Management.",
" ",
"-Reads available models from ki_dev.ai_models database table.",
"-Synchronizes with PHP ModelRegistry for consistent model availability.",
"+Provides available models via environment configuration.",
"+No database dependency - fully portable.",
" \"\"\"",
" ",
"+import os",
" from typing import Optional",
" ",
"-import mysql.connector",
" ",
"-from config import get_db_password",
"+# =============================================================================",
"+# DEFAULT MODELS (Override via environment)",
"+# =============================================================================",
" ",
"+DEFAULT_MODELS = {",
"+ # Embedding models",
"+ \"ollama:mxbai-embed-large:latest\": {",
"+ \"provider\": \"ollama\",",
"+ \"model_id\": \"mxbai-embed-large:latest\",",
"+ \"display_name\": \"MXBai Embed Large (1024d)\",",
"+ \"is_embedding\": True,",
"+ \"is_chat\": False,",
"+ \"is_vision\": False,",
"+ \"context_length\": 512,",
"+ },",
"+ # Chat models - Ollama",
"+ \"ollama:llama3.2:3b\": {",
"+ \"provider\": \"ollama\",",
"+ \"model_id\": \"llama3.2:3b\",",
"+ \"display_name\": \"Llama 3.2 3B\",",
"+ \"is_embedding\": False,",
"+ \"is_chat\": True,",
"+ \"is_vision\": False,",
"+ \"context_length\": 8192,",
"+ },",
"+ \"ollama:mistral:latest\": {",
"+ \"provider\": \"ollama\",",
"+ \"model_id\": \"mistral:latest\",",
"+ \"display_name\": \"Mistral 7B\",",
"+ \"is_embedding\": False,",
"+ \"is_chat\": True,",
"+ \"is_vision\": False,",
"+ \"context_length\": 8192,",
"+ },",
"+ # Vision models",
"+ \"ollama:llava:latest\": {",
"+ \"provider\": \"ollama\",",
"+ \"model_id\": \"llava:latest\",",
"+ \"display_name\": \"LLaVA Vision\",",
"+ \"is_embedding\": False,",
"+ \"is_chat\": True,",
"+ \"is_vision\": True,",
"+ \"context_length\": 4096,",
"+ },",
"+ # Anthropic models",
"+ \"anthropic:claude-sonnet-4-20250514\": {",
"+ \"provider\": \"anthropic\",",
"+ \"model_id\": \"claude-sonnet-4-20250514\",",
"+ \"display_name\": \"Claude Sonnet 4\",",
"+ \"is_embedding\": False,",
"+ \"is_chat\": True,",
"+ \"is_vision\": True,",
"+ \"context_length\": 200000,",
"+ },",
"+}",
" ",
"+",
" class ModelRegistry:",
" \"\"\"Central registry for all AI models.\"\"\"",
" ",
"- _cache: list | None = None",
" _instance: Optional[\"ModelRegistry\"] = None",
" ",
" def __init__(self):",
"- self._conn = None",
"+ self._models = self._load_models()",
" ",
" @classmethod",
" def get_instance(cls) -> \"ModelRegistry\":"
]
},
{
"oldStart": 28,
"oldLines": 158,
"newStart": 81,
"newLines": 104,
"lines": [
" cls._instance = cls()",
" return cls._instance",
" ",
"- def _get_connection(self):",
"- \"\"\"Get or create database connection.\"\"\"",
"- if self._conn is None or not self._conn.is_connected():",
"- self._conn = mysql.connector.connect(",
"- host=\"localhost\",",
"- user=\"root\",",
"- password=get_db_password(),",
"- database=\"ki_dev\",",
"- autocommit=True,",
"- )",
"- return self._conn",
"-",
" @classmethod",
" def clear_cache(cls):",
"- \"\"\"Clear cached models.\"\"\"",
"- cls._cache = None",
"+ \"\"\"Clear cached instance.\"\"\"",
"+ cls._instance = None",
" ",
"- def get_chat_models(self) -> dict:",
"- \"\"\"Get all available chat models.",
"+ def _load_models(self) -> dict:",
"+ \"\"\"Load models from defaults + environment overrides.\"\"\"",
"+ models = dict(DEFAULT_MODELS)",
" ",
"- Returns:",
"- dict: {full_key: display_name}",
"- \"\"\"",
"- return self._get_models(is_chat=True)",
"+ # Add custom models from environment",
"+ # Format: PIPELINE_MODEL_<N>=provider:model_id:display_name",
"+ for key, value in os.environ.items():",
"+ if key.startswith(\"PIPELINE_MODEL_\"):",
"+ try:",
"+ parts = value.split(\":\", 2)",
"+ if len(parts) >= 2:",
"+ provider, model_id = parts[0], parts[1]",
"+ display_name = parts[2] if len(parts) > 2 else model_id",
"+ full_key = f\"{provider}:{model_id}\"",
"+ models[full_key] = {",
"+ \"provider\": provider,",
"+ \"model_id\": model_id,",
"+ \"display_name\": display_name,",
"+ \"is_embedding\": False,",
"+ \"is_chat\": True,",
"+ \"is_vision\": False,",
"+ \"context_length\": 8192,",
"+ }",
"+ except Exception:",
"+ pass",
" ",
"- def get_vision_models(self) -> dict:",
"- \"\"\"Get all vision-capable models.",
"+ return models",
" ",
"- Returns:",
"- dict: {full_key: display_name}",
"- \"\"\"",
"- return self._get_models(is_vision=True)",
"+ def get_chat_models(self) -> dict:",
"+ \"\"\"Get all available chat models.\"\"\"",
"+ return {k: v[\"display_name\"] for k, v in self._models.items() if v.get(\"is_chat\")}",
" ",
"+ def get_vision_models(self) -> dict:",
"+ \"\"\"Get all vision-capable models.\"\"\"",
"+ return {k: v[\"display_name\"] for k, v in self._models.items() if v.get(\"is_vision\")}",
"+",
" def get_embedding_models(self) -> dict:",
"- \"\"\"Get all embedding models.",
"+ \"\"\"Get all embedding models.\"\"\"",
"+ return {k: v[\"display_name\"] for k, v in self._models.items() if v.get(\"is_embedding\")}",
" ",
"- Returns:",
"- dict: {full_key: display_name}",
"- \"\"\"",
"- return self._get_models(is_embedding=True)",
"-",
"- def _get_models(",
"- self,",
"- is_chat: bool | None = None,",
"- is_vision: bool | None = None,",
"- is_embedding: bool | None = None,",
"- provider: str | None = None,",
"- ) -> dict:",
"- \"\"\"Get models with optional filters.\"\"\"",
"- all_models = self._load_all_models()",
"- result = {}",
"-",
"- for model in all_models:",
"- if not model[\"is_available\"]:",
"- continue",
"- if is_chat is not None and bool(model[\"is_chat\"]) != is_chat:",
"- continue",
"- if is_vision is not None and bool(model[\"is_vision\"]) != is_vision:",
"- continue",
"- if is_embedding is not None and bool(model[\"is_embedding\"]) != is_embedding:",
"- continue",
"- if provider is not None and model[\"provider\"] != provider:",
"- continue",
"-",
"- result[model[\"full_key\"]] = model[\"display_name\"]",
"-",
"- return result",
"-",
" def get_model(self, full_key: str) -> dict | None:",
" \"\"\"Get a single model by full_key.\"\"\"",
"- all_models = self._load_all_models()",
"+ return self._models.get(full_key)",
" ",
"- for model in all_models:",
"- if model[\"full_key\"] == full_key:",
"- return model",
"-",
"- return None",
"-",
" def get_label(self, full_key: str) -> str:",
" \"\"\"Get display label for a model.\"\"\"",
" model = self.get_model(full_key)",
" return model[\"display_name\"] if model else full_key",
" ",
" def is_valid(self, full_key: str) -> bool:",
"- \"\"\"Check if model exists and is available.\"\"\"",
"- model = self.get_model(full_key)",
"- return model is not None and model[\"is_available\"]",
"+ \"\"\"Check if model exists.\"\"\"",
"+ return full_key in self._models",
" ",
" def is_local(self, full_key: str) -> bool:",
" \"\"\"Check if model is local (Ollama).\"\"\"",
" return full_key.startswith(\"ollama:\")",
" ",
" def get_default_chat_model(self) -> str:",
"- \"\"\"Get default chat model (first available by priority).\"\"\"",
"- chat_models = self.get_chat_models()",
"- if chat_models:",
"- return next(iter(chat_models.keys()))",
"+ \"\"\"Get default chat model from environment or fallback.\"\"\"",
"+ env_model = os.environ.get(\"OLLAMA_CHAT_MODEL\")",
"+ if env_model:",
"+ return f\"ollama:{env_model}\" if not env_model.startswith(\"ollama:\") else env_model",
"+ # Fallback to first available Ollama chat model",
"+ for key, model in self._models.items():",
"+ if model.get(\"is_chat\") and model.get(\"provider\") == \"ollama\":",
"+ return key",
" return \"ollama:mistral:latest\"",
" ",
" def get_default_embedding_model(self) -> str:",
" \"\"\"Get default embedding model.\"\"\"",
"- embed_models = self.get_embedding_models()",
"- if embed_models:",
"- return next(iter(embed_models.keys()))",
"+ env_model = os.environ.get(\"OLLAMA_EMBEDDING_MODEL\")",
"+ if env_model:",
"+ return f\"ollama:{env_model}:latest\"",
" return \"ollama:mxbai-embed-large:latest\"",
" ",
" def get_ollama_model_id(self, full_key: str) -> str:",
"- \"\"\"Extract Ollama model ID from full_key.",
"-",
"- Example: 'ollama:gemma3:27b-it-qat' -> 'gemma3:27b-it-qat'",
"- \"\"\"",
"+ \"\"\"Extract Ollama model ID from full_key.\"\"\"",
" if full_key.startswith(\"ollama:\"):",
"- return full_key[7:] # Remove 'ollama:' prefix",
"+ return full_key[7:]",
" return full_key",
" ",
"- def _load_all_models(self) -> list:",
"- \"\"\"Load all models from database (with caching).\"\"\"",
"- if ModelRegistry._cache is not None:",
"- return ModelRegistry._cache",
" ",
"- conn = self._get_connection()",
"- cursor = conn.cursor(dictionary=True)",
"+# =============================================================================",
"+# CONVENIENCE FUNCTIONS",
"+# =============================================================================",
" ",
"- cursor.execute(",
"- \"\"\"",
"- SELECT id, provider, model_id, display_name, full_key,",
"- is_available, is_chat, is_embedding, is_vision,",
"- context_length, parameters, priority",
"- FROM ai_models",
"- WHERE is_available = 1",
"- ORDER BY priority ASC",
"- \"\"\"",
"- )",
"-",
"- ModelRegistry._cache = cursor.fetchall()",
"- cursor.close()",
"- return ModelRegistry._cache",
"-",
"-",
"-# Convenience functions for backward compatibility",
" def get_chat_model() -> str:",
"- \"\"\"Get the default chat model from registry.\"\"\"",
"- registry = ModelRegistry.get_instance()",
"- # Prefer local Ollama models for pipeline",
"- ollama_models = registry._get_models(is_chat=True, provider=\"ollama\")",
"- if ollama_models:",
"- return next(iter(ollama_models.keys()))",
"- return registry.get_default_chat_model()",
"+ \"\"\"Get the default chat model.\"\"\"",
"+ return ModelRegistry.get_instance().get_default_chat_model()",
" ",
" ",
" def get_embedding_model() -> str:",
"- \"\"\"Get the default embedding model from registry.\"\"\"",
"+ \"\"\"Get the default embedding model.\"\"\"",
" return ModelRegistry.get_instance().get_default_embedding_model()",
" ",
" "
]
}
],
"originalFile": "\"\"\"\nModel Registry - Single Source of Truth for AI Models.\n\nReads available models from ki_dev.ai_models database table.\nSynchronizes with PHP ModelRegistry for consistent model availability.\n\"\"\"\n\nfrom typing import Optional\n\nimport mysql.connector\n\nfrom config import get_db_password\n\n\nclass ModelRegistry:\n \"\"\"Central registry for all AI models.\"\"\"\n\n _cache: list | None = None\n _instance: Optional[\"ModelRegistry\"] = None\n\n def __init__(self):\n self._conn = None\n\n @classmethod\n def get_instance(cls) -> \"ModelRegistry\":\n \"\"\"Get singleton instance.\"\"\"\n if cls._instance is None:\n cls._instance = cls()\n return cls._instance\n\n def _get_connection(self):\n \"\"\"Get or create database connection.\"\"\"\n if self._conn is None or not self._conn.is_connected():\n self._conn = mysql.connector.connect(\n host=\"localhost\",\n user=\"root\",\n password=get_db_password(),\n database=\"ki_dev\",\n autocommit=True,\n )\n return self._conn\n\n @classmethod\n def clear_cache(cls):\n \"\"\"Clear cached models.\"\"\"\n cls._cache = None\n\n def get_chat_models(self) -> dict:\n \"\"\"Get all available chat models.\n\n Returns:\n dict: {full_key: display_name}\n \"\"\"\n return self._get_models(is_chat=True)\n\n def get_vision_models(self) -> dict:\n \"\"\"Get all vision-capable models.\n\n Returns:\n dict: {full_key: display_name}\n \"\"\"\n return self._get_models(is_vision=True)\n\n def get_embedding_models(self) -> dict:\n \"\"\"Get all embedding models.\n\n Returns:\n dict: {full_key: display_name}\n \"\"\"\n return self._get_models(is_embedding=True)\n\n def _get_models(\n self,\n is_chat: bool | None = None,\n is_vision: bool | None = None,\n is_embedding: bool | None = None,\n provider: str | None = None,\n ) -> dict:\n \"\"\"Get models with optional filters.\"\"\"\n all_models = self._load_all_models()\n result = {}\n\n for model in all_models:\n if not model[\"is_available\"]:\n continue\n if is_chat is not None and bool(model[\"is_chat\"]) != is_chat:\n continue\n if is_vision is not None and bool(model[\"is_vision\"]) != is_vision:\n continue\n if is_embedding is not None and bool(model[\"is_embedding\"]) != is_embedding:\n continue\n if provider is not None and model[\"provider\"] != provider:\n continue\n\n result[model[\"full_key\"]] = model[\"display_name\"]\n\n return result\n\n def get_model(self, full_key: str) -> dict | None:\n \"\"\"Get a single model by full_key.\"\"\"\n all_models = self._load_all_models()\n\n for model in all_models:\n if model[\"full_key\"] == full_key:\n return model\n\n return None\n\n def get_label(self, full_key: str) -> str:\n \"\"\"Get display label for a model.\"\"\"\n model = self.get_model(full_key)\n return model[\"display_name\"] if model else full_key\n\n def is_valid(self, full_key: str) -> bool:\n \"\"\"Check if model exists and is available.\"\"\"\n model = self.get_model(full_key)\n return model is not None and model[\"is_available\"]\n\n def is_local(self, full_key: str) -> bool:\n \"\"\"Check if model is local (Ollama).\"\"\"\n return full_key.startswith(\"ollama:\")\n\n def get_default_chat_model(self) -> str:\n \"\"\"Get default chat model (first available by priority).\"\"\"\n chat_models = self.get_chat_models()\n if chat_models:\n return next(iter(chat_models.keys()))\n return \"ollama:mistral:latest\"\n\n def get_default_embedding_model(self) -> str:\n \"\"\"Get default embedding model.\"\"\"\n embed_models = self.get_embedding_models()\n if embed_models:\n return next(iter(embed_models.keys()))\n return \"ollama:mxbai-embed-large:latest\"\n\n def get_ollama_model_id(self, full_key: str) -> str:\n \"\"\"Extract Ollama model ID from full_key.\n\n Example: 'ollama:gemma3:27b-it-qat' -> 'gemma3:27b-it-qat'\n \"\"\"\n if full_key.startswith(\"ollama:\"):\n return full_key[7:] # Remove 'ollama:' prefix\n return full_key\n\n def _load_all_models(self) -> list:\n \"\"\"Load all models from database (with caching).\"\"\"\n if ModelRegistry._cache is not None:\n return ModelRegistry._cache\n\n conn = self._get_connection()\n cursor = conn.cursor(dictionary=True)\n\n cursor.execute(\n \"\"\"\n SELECT id, provider, model_id, display_name, full_key,\n is_available, is_chat, is_embedding, is_vision,\n context_length, parameters, priority\n FROM ai_models\n WHERE is_available = 1\n ORDER BY priority ASC\n \"\"\"\n )\n\n ModelRegistry._cache = cursor.fetchall()\n cursor.close()\n return ModelRegistry._cache\n\n\n# Convenience functions for backward compatibility\ndef get_chat_model() -> str:\n \"\"\"Get the default chat model from registry.\"\"\"\n registry = ModelRegistry.get_instance()\n # Prefer local Ollama models for pipeline\n ollama_models = registry._get_models(is_chat=True, provider=\"ollama\")\n if ollama_models:\n return next(iter(ollama_models.keys()))\n return registry.get_default_chat_model()\n\n\ndef get_embedding_model() -> str:\n \"\"\"Get the default embedding model from registry.\"\"\"\n return ModelRegistry.get_instance().get_default_embedding_model()\n\n\ndef get_model_id_for_ollama(full_key: str) -> str:\n \"\"\"Extract Ollama model ID from full_key.\"\"\"\n return ModelRegistry.get_instance().get_ollama_model_id(full_key)\n"
}
}