350 lines
12 KiB
Python
350 lines
12 KiB
Python
"""
|
|
Admin configuration for LLM Manager — ported from Spelunker.
|
|
|
|
Adds system model actions for embedding, chat, and reranker models.
|
|
"""
|
|
|
|
from django.contrib import admin, messages
|
|
from django.db import transaction
|
|
from django.utils.html import format_html
|
|
|
|
from .models import LLMApi, LLMModel, LLMUsage
|
|
from .services import test_llm_api
|
|
|
|
|
|
@admin.register(LLMApi)
|
|
class LLMApiAdmin(admin.ModelAdmin):
|
|
list_display = (
|
|
"name",
|
|
"api_type",
|
|
"base_url",
|
|
"is_active",
|
|
"last_test_status",
|
|
"last_tested_at",
|
|
"supports_streaming",
|
|
"timeout_seconds",
|
|
"created_at",
|
|
)
|
|
list_filter = ("api_type", "is_active", "last_test_status", "supports_streaming")
|
|
search_fields = ("name", "base_url")
|
|
readonly_fields = (
|
|
"created_at",
|
|
"updated_at",
|
|
"last_tested_at",
|
|
"last_test_status",
|
|
"last_test_message",
|
|
)
|
|
actions = ["test_api_connection"]
|
|
fieldsets = (
|
|
("API Info", {"fields": ("name", "api_type", "base_url", "is_active")}),
|
|
("Security", {"fields": ("api_key",)}),
|
|
(
|
|
"Advanced",
|
|
{"fields": ("supports_streaming", "timeout_seconds", "max_retries", "created_by")},
|
|
),
|
|
(
|
|
"Test Status",
|
|
{"fields": ("last_tested_at", "last_test_status", "last_test_message")},
|
|
),
|
|
("Timestamps", {"fields": ("created_at", "updated_at")}),
|
|
)
|
|
|
|
def test_api_connection(self, request, queryset):
|
|
"""Test selected LLM API(s) and discover models."""
|
|
success_count = 0
|
|
failed_count = 0
|
|
total_added = 0
|
|
total_updated = 0
|
|
total_deactivated = 0
|
|
|
|
for api in queryset:
|
|
result = test_llm_api(api)
|
|
|
|
if result["success"]:
|
|
success_count += 1
|
|
total_added += result["models_added"]
|
|
total_updated += result["models_updated"]
|
|
total_deactivated += result["models_deactivated"]
|
|
self.message_user(request, f"✓ {api.name}: {result['message']}", messages.SUCCESS)
|
|
else:
|
|
failed_count += 1
|
|
self.message_user(request, f"✗ {api.name}: {result['error']}", messages.ERROR)
|
|
|
|
if success_count > 0:
|
|
summary = (
|
|
f"Tested {success_count} API(s). "
|
|
f"Total: {total_added} added, {total_updated} updated, "
|
|
f"{total_deactivated} deactivated."
|
|
)
|
|
self.message_user(request, summary, messages.SUCCESS)
|
|
|
|
if failed_count > 0:
|
|
self.message_user(
|
|
request,
|
|
f"Failed to test {failed_count} API(s). Check logs.",
|
|
messages.WARNING,
|
|
)
|
|
|
|
test_api_connection.short_description = "Test API Connection and Discover Models"
|
|
|
|
|
|
@admin.register(LLMModel)
|
|
class LLMModelAdmin(admin.ModelAdmin):
|
|
list_display = (
|
|
"name",
|
|
"api",
|
|
"model_type",
|
|
"vector_dimensions_display",
|
|
"context_window",
|
|
"input_cost_per_1k",
|
|
"system_embedding_badge",
|
|
"system_chat_badge",
|
|
"system_reranker_badge",
|
|
"system_vision_badge",
|
|
"is_active",
|
|
"created_at",
|
|
)
|
|
list_filter = (
|
|
"api",
|
|
"model_type",
|
|
"supports_cache",
|
|
"supports_vision",
|
|
"supports_multimodal",
|
|
"is_active",
|
|
"is_system_embedding_model",
|
|
"is_system_chat_model",
|
|
"is_system_reranker_model",
|
|
"is_system_vision_model",
|
|
)
|
|
search_fields = ("name", "display_name", "api__name")
|
|
readonly_fields = (
|
|
"created_at",
|
|
"updated_at",
|
|
"is_system_embedding_model",
|
|
"is_system_chat_model",
|
|
"is_system_reranker_model",
|
|
"is_system_vision_model",
|
|
)
|
|
actions = [
|
|
"set_as_system_embedding_model",
|
|
"set_as_system_chat_model",
|
|
"set_as_system_reranker_model",
|
|
"set_as_system_vision_model",
|
|
]
|
|
fieldsets = (
|
|
("Model Info", {"fields": ("api", "name", "display_name", "model_type", "is_active")}),
|
|
(
|
|
"System Defaults",
|
|
{
|
|
"fields": (
|
|
"is_system_embedding_model",
|
|
"is_system_chat_model",
|
|
"is_system_reranker_model",
|
|
"is_system_vision_model",
|
|
),
|
|
"classes": ("collapse",),
|
|
"description": (
|
|
"System default models are set via admin actions. "
|
|
"Only one model per type can be system default."
|
|
),
|
|
},
|
|
),
|
|
(
|
|
"Capabilities",
|
|
{
|
|
"fields": (
|
|
"context_window",
|
|
"max_output_tokens",
|
|
"vector_dimensions",
|
|
"supports_cache",
|
|
"supports_vision",
|
|
"supports_multimodal",
|
|
"supports_function_calling",
|
|
"supports_json_mode",
|
|
),
|
|
},
|
|
),
|
|
(
|
|
"Pricing",
|
|
{"fields": ("input_cost_per_1k", "output_cost_per_1k", "cached_cost_per_1k")},
|
|
),
|
|
("Timestamps", {"fields": ("created_at", "updated_at")}),
|
|
)
|
|
|
|
def vector_dimensions_display(self, obj):
|
|
if obj.model_type in ("embedding", "multimodal_embed") and obj.vector_dimensions:
|
|
return format_html(
|
|
'<span style="color: #0066cc; font-weight: bold;">{}</span>',
|
|
obj.vector_dimensions,
|
|
)
|
|
elif obj.model_type in ("embedding", "multimodal_embed"):
|
|
return format_html('<span style="color: #999;">Not set</span>')
|
|
return "-"
|
|
|
|
vector_dimensions_display.short_description = "Dimensions"
|
|
|
|
def system_embedding_badge(self, obj):
|
|
if obj.is_system_embedding_model and obj.model_type in ("embedding", "multimodal_embed"):
|
|
return format_html(
|
|
'<span style="background:#28a745;color:white;padding:3px 8px;'
|
|
'border-radius:3px;font-weight:bold;">SYSTEM DEFAULT</span>'
|
|
)
|
|
return ""
|
|
|
|
system_embedding_badge.short_description = "Embed Default"
|
|
|
|
def system_chat_badge(self, obj):
|
|
if obj.is_system_chat_model and obj.model_type == "chat":
|
|
return format_html(
|
|
'<span style="background:#007bff;color:white;padding:3px 8px;'
|
|
'border-radius:3px;font-weight:bold;">SYSTEM DEFAULT</span>'
|
|
)
|
|
return ""
|
|
|
|
system_chat_badge.short_description = "Chat Default"
|
|
|
|
def system_reranker_badge(self, obj):
|
|
if obj.is_system_reranker_model and obj.model_type == "reranker":
|
|
return format_html(
|
|
'<span style="background:#fd7e14;color:white;padding:3px 8px;'
|
|
'border-radius:3px;font-weight:bold;">SYSTEM DEFAULT</span>'
|
|
)
|
|
return ""
|
|
|
|
system_reranker_badge.short_description = "Reranker Default"
|
|
|
|
def system_vision_badge(self, obj):
|
|
if obj.is_system_vision_model and obj.model_type in ("vision", "chat"):
|
|
return format_html(
|
|
'<span style="background:#6f42c1;color:white;padding:3px 8px;'
|
|
'border-radius:3px;font-weight:bold;">SYSTEM DEFAULT</span>'
|
|
)
|
|
return ""
|
|
|
|
system_vision_badge.short_description = "Vision Default"
|
|
|
|
# --- System model actions -----------------------------------------------
|
|
|
|
def _set_system_model(self, request, queryset, model_type, field_name, label):
|
|
"""Generic helper for set-as-system-model admin actions."""
|
|
if queryset.count() != 1:
|
|
self.message_user(
|
|
request,
|
|
f"Please select exactly ONE model to set as system {label}.",
|
|
messages.ERROR,
|
|
)
|
|
return
|
|
|
|
new_model = queryset.first()
|
|
|
|
valid_types = [model_type]
|
|
if model_type == "embedding":
|
|
valid_types = ["embedding", "multimodal_embed"]
|
|
elif model_type == "vision":
|
|
valid_types = ["vision", "chat"]
|
|
|
|
if new_model.model_type not in valid_types:
|
|
self.message_user(
|
|
request,
|
|
f'Only {label} models can be set as system {label}. '
|
|
f'"{new_model.name}" is type: {new_model.model_type}',
|
|
messages.ERROR,
|
|
)
|
|
return
|
|
|
|
if not new_model.is_active:
|
|
self.message_user(
|
|
request,
|
|
f'Cannot set inactive model "{new_model.name}" as system {label}.',
|
|
messages.ERROR,
|
|
)
|
|
return
|
|
|
|
with transaction.atomic():
|
|
LLMModel.objects.filter(**{field_name: True}).update(**{field_name: False})
|
|
setattr(new_model, field_name, True)
|
|
new_model.save(update_fields=[field_name])
|
|
|
|
self.message_user(
|
|
request,
|
|
f"✓ {new_model.api.name}: {new_model.name} is now the system {label}.",
|
|
messages.SUCCESS,
|
|
)
|
|
|
|
def set_as_system_embedding_model(self, request, queryset):
|
|
self._set_system_model(request, queryset, "embedding", "is_system_embedding_model", "embedding model")
|
|
|
|
set_as_system_embedding_model.short_description = "Set as System Embedding Model"
|
|
|
|
def set_as_system_chat_model(self, request, queryset):
|
|
self._set_system_model(request, queryset, "chat", "is_system_chat_model", "chat model")
|
|
|
|
set_as_system_chat_model.short_description = "Set as System Chat Model"
|
|
|
|
def set_as_system_reranker_model(self, request, queryset):
|
|
self._set_system_model(request, queryset, "reranker", "is_system_reranker_model", "reranker model")
|
|
|
|
set_as_system_reranker_model.short_description = "Set as System Reranker Model"
|
|
|
|
def set_as_system_vision_model(self, request, queryset):
|
|
self._set_system_model(request, queryset, "vision", "is_system_vision_model", "vision model")
|
|
|
|
set_as_system_vision_model.short_description = "Set as System Vision Model"
|
|
|
|
def save_model(self, request, obj, form, change):
|
|
"""Ensure only ONE model per type is marked as system default."""
|
|
type_field_map = {
|
|
"embedding": "is_system_embedding_model",
|
|
"multimodal_embed": "is_system_embedding_model",
|
|
"chat": "is_system_chat_model",
|
|
"reranker": "is_system_reranker_model",
|
|
"vision": "is_system_vision_model",
|
|
}
|
|
for mtype, field in type_field_map.items():
|
|
if getattr(obj, field, False) and obj.model_type == mtype:
|
|
LLMModel.objects.filter(**{field: True}).exclude(pk=obj.pk).update(**{field: False})
|
|
self.message_user(
|
|
request,
|
|
f"{obj.name} is now the system-wide {mtype} model.",
|
|
messages.SUCCESS,
|
|
)
|
|
elif getattr(obj, field, False) and obj.model_type != mtype:
|
|
setattr(obj, field, False)
|
|
super().save_model(request, obj, form, change)
|
|
|
|
|
|
@admin.register(LLMUsage)
|
|
class LLMUsageAdmin(admin.ModelAdmin):
|
|
list_display = (
|
|
"timestamp",
|
|
"user",
|
|
"model",
|
|
"input_tokens",
|
|
"output_tokens",
|
|
"cached_tokens",
|
|
"total_cost",
|
|
"session_id",
|
|
"purpose",
|
|
)
|
|
list_filter = ("model", "purpose", "timestamp")
|
|
search_fields = ("user__username", "session_id", "model__name")
|
|
readonly_fields = (
|
|
"user",
|
|
"model",
|
|
"timestamp",
|
|
"input_tokens",
|
|
"output_tokens",
|
|
"cached_tokens",
|
|
"total_cost",
|
|
"session_id",
|
|
"purpose",
|
|
"request_metadata",
|
|
)
|
|
date_hierarchy = "timestamp"
|
|
|
|
def has_add_permission(self, request):
|
|
return False
|
|
|
|
def has_change_permission(self, request, obj=None):
|
|
return False
|