338 lines
11 KiB
Python
338 lines
11 KiB
Python
# #region LlmRoutes [C:3] [TYPE Module] [SEMANTICS fastapi, llm, api, provider]
|
|
# @BRIEF API routes for LLM provider configuration and management.
|
|
# @LAYER: UI (API)
|
|
# @RELATION DEPENDS_ON -> [LLMProviderService]
|
|
# @RELATION DEPENDS_ON -> [LLMProviderConfig]
|
|
# @RELATION DEPENDS_ON -> [get_current_user]
|
|
# @RELATION DEPENDS_ON -> [get_db]
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from typing import List, Optional
|
|
from ...core.logger import logger
|
|
from ...schemas.auth import User
|
|
from ...dependencies import get_current_user as get_current_active_user
|
|
from ...plugins.llm_analysis.models import LLMProviderConfig, LLMProviderType
|
|
from ...services.llm_provider import LLMProviderService
|
|
from ...core.database import get_db
|
|
from sqlalchemy.orm import Session
|
|
|
|
# #region router [TYPE Global]
|
|
# @BRIEF APIRouter instance for LLM routes.
|
|
router = APIRouter(tags=["LLM"])
|
|
# #endregion router
|
|
|
|
|
|
# #region _is_valid_runtime_api_key [TYPE Function]
|
|
# @BRIEF Validate decrypted runtime API key presence/shape.
|
|
# @PRE: value can be None.
|
|
# @POST: Returns True only for non-placeholder key.
|
|
# @RELATION BINDS_TO -> [LlmRoutes]
|
|
def _is_valid_runtime_api_key(value: Optional[str]) -> bool:
|
|
key = (value or "").strip()
|
|
if not key:
|
|
return False
|
|
if key in {"********", "EMPTY_OR_NONE"}:
|
|
return False
|
|
return len(key) >= 16
|
|
|
|
|
|
# #endregion _is_valid_runtime_api_key
|
|
|
|
|
|
# #region get_providers [TYPE Function]
|
|
# @BRIEF Retrieve all LLM provider configurations.
|
|
# @PRE: User is authenticated.
|
|
# @POST: Returns list of LLMProviderConfig.
|
|
# @RELATION CALLS -> [LLMProviderService]
|
|
# @RELATION DEPENDS_ON -> [LLMProviderConfig]
|
|
@router.get("/providers", response_model=List[LLMProviderConfig])
|
|
async def get_providers(
|
|
current_user: User = Depends(get_current_active_user), db: Session = Depends(get_db)
|
|
):
|
|
"""
|
|
Get all LLM provider configurations.
|
|
"""
|
|
logger.reason(
|
|
f"Fetching providers for user: {current_user.username}",
|
|
extra={"src": "llm_routes.get_providers"},
|
|
)
|
|
service = LLMProviderService(db)
|
|
providers = service.get_all_providers()
|
|
return [
|
|
LLMProviderConfig(
|
|
id=p.id,
|
|
provider_type=LLMProviderType(p.provider_type),
|
|
name=p.name,
|
|
base_url=p.base_url,
|
|
api_key="********",
|
|
default_model=p.default_model,
|
|
is_active=p.is_active,
|
|
)
|
|
for p in providers
|
|
]
|
|
|
|
|
|
# #endregion get_providers
|
|
|
|
|
|
# #region get_llm_status [TYPE Function]
|
|
# @BRIEF Returns whether LLM runtime is configured for dashboard validation.
|
|
# @PRE: User is authenticated.
|
|
# @POST: configured=true only when an active provider with valid decrypted key exists.
|
|
# @RELATION CALLS -> [LLMProviderService]
|
|
# @RELATION CALLS -> [_is_valid_runtime_api_key]
|
|
@router.get("/status")
|
|
async def get_llm_status(
|
|
current_user: User = Depends(get_current_active_user), db: Session = Depends(get_db)
|
|
):
|
|
service = LLMProviderService(db)
|
|
providers = service.get_all_providers()
|
|
active_provider = next((p for p in providers if p.is_active), None)
|
|
|
|
if not providers:
|
|
return {
|
|
"configured": False,
|
|
"reason": "no_providers_configured",
|
|
"provider_count": 0,
|
|
"active_provider_count": 0,
|
|
}
|
|
|
|
if not active_provider:
|
|
return {
|
|
"configured": False,
|
|
"reason": "no_active_provider",
|
|
"provider_count": len(providers),
|
|
"active_provider_count": 0,
|
|
"providers": [
|
|
{
|
|
"id": provider.id,
|
|
"name": provider.name,
|
|
"provider_type": provider.provider_type,
|
|
"is_active": bool(provider.is_active),
|
|
}
|
|
for provider in providers
|
|
],
|
|
}
|
|
|
|
api_key = service.get_decrypted_api_key(active_provider.id)
|
|
if not _is_valid_runtime_api_key(api_key):
|
|
return {
|
|
"configured": False,
|
|
"reason": "invalid_api_key",
|
|
"provider_count": len(providers),
|
|
"active_provider_count": len(
|
|
[provider for provider in providers if provider.is_active]
|
|
),
|
|
"provider_id": active_provider.id,
|
|
"provider_name": active_provider.name,
|
|
"provider_type": active_provider.provider_type,
|
|
"default_model": active_provider.default_model,
|
|
}
|
|
|
|
return {
|
|
"configured": True,
|
|
"reason": "ok",
|
|
"provider_count": len(providers),
|
|
"active_provider_count": len(
|
|
[provider for provider in providers if provider.is_active]
|
|
),
|
|
"provider_id": active_provider.id,
|
|
"provider_name": active_provider.name,
|
|
"provider_type": active_provider.provider_type,
|
|
"default_model": active_provider.default_model,
|
|
}
|
|
|
|
|
|
# #endregion get_llm_status
|
|
|
|
|
|
# #region create_provider [TYPE Function]
|
|
# @BRIEF Create a new LLM provider configuration.
|
|
# @PRE: User is authenticated and has admin permissions.
|
|
# @POST: Returns the created LLMProviderConfig.
|
|
# @RELATION CALLS -> [LLMProviderService]
|
|
# @RELATION DEPENDS_ON -> [LLMProviderConfig]
|
|
@router.post(
|
|
"/providers", response_model=LLMProviderConfig, status_code=status.HTTP_201_CREATED
|
|
)
|
|
async def create_provider(
|
|
config: LLMProviderConfig,
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Create a new LLM provider configuration.
|
|
"""
|
|
service = LLMProviderService(db)
|
|
provider = service.create_provider(config)
|
|
return LLMProviderConfig(
|
|
id=provider.id,
|
|
provider_type=LLMProviderType(provider.provider_type),
|
|
name=provider.name,
|
|
base_url=provider.base_url,
|
|
api_key="********",
|
|
default_model=provider.default_model,
|
|
is_active=provider.is_active,
|
|
)
|
|
|
|
|
|
# #endregion create_provider
|
|
|
|
|
|
# #region update_provider [TYPE Function]
|
|
# @BRIEF Update an existing LLM provider configuration.
|
|
# @PRE: User is authenticated and has admin permissions.
|
|
# @POST: Returns the updated LLMProviderConfig.
|
|
# @RELATION CALLS -> [LLMProviderService]
|
|
# @RELATION DEPENDS_ON -> [LLMProviderConfig]
|
|
@router.put("/providers/{provider_id}", response_model=LLMProviderConfig)
|
|
async def update_provider(
|
|
provider_id: str,
|
|
config: LLMProviderConfig,
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Update an existing LLM provider configuration.
|
|
"""
|
|
service = LLMProviderService(db)
|
|
provider = service.update_provider(provider_id, config)
|
|
if not provider:
|
|
raise HTTPException(status_code=404, detail="Provider not found")
|
|
|
|
return LLMProviderConfig(
|
|
id=provider.id,
|
|
provider_type=LLMProviderType(provider.provider_type),
|
|
name=provider.name,
|
|
base_url=provider.base_url,
|
|
api_key="********",
|
|
default_model=provider.default_model,
|
|
is_active=provider.is_active,
|
|
)
|
|
|
|
|
|
# #endregion update_provider
|
|
|
|
|
|
# #region delete_provider [TYPE Function]
|
|
# @BRIEF Delete an LLM provider configuration.
|
|
# @PRE: User is authenticated and has admin permissions.
|
|
# @POST: Returns success status.
|
|
# @RELATION CALLS -> [LLMProviderService]
|
|
@router.delete("/providers/{provider_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def delete_provider(
|
|
provider_id: str,
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""
|
|
Delete an LLM provider configuration.
|
|
"""
|
|
service = LLMProviderService(db)
|
|
if not service.delete_provider(provider_id):
|
|
raise HTTPException(status_code=404, detail="Provider not found")
|
|
return
|
|
|
|
|
|
# #endregion delete_provider
|
|
|
|
|
|
# #region test_connection [TYPE Function]
|
|
# @BRIEF Test connection to an LLM provider.
|
|
# @PRE: User is authenticated.
|
|
# @POST: Returns success status and message.
|
|
# @RELATION CALLS -> [LLMProviderService]
|
|
# @RELATION DEPENDS_ON -> [LLMClient]
|
|
@router.post("/providers/{provider_id}/test")
|
|
async def test_connection(
|
|
provider_id: str,
|
|
current_user: User = Depends(get_current_active_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
logger.reason(
|
|
f"Testing connection for provider_id: {provider_id}",
|
|
extra={"src": "llm_routes.test_connection"},
|
|
)
|
|
"""
|
|
Test connection to an LLM provider.
|
|
"""
|
|
from ...plugins.llm_analysis.service import LLMClient
|
|
|
|
service = LLMProviderService(db)
|
|
db_provider = service.get_provider(provider_id)
|
|
if not db_provider:
|
|
raise HTTPException(status_code=404, detail="Provider not found")
|
|
|
|
api_key = service.get_decrypted_api_key(provider_id)
|
|
|
|
# Check if API key was successfully decrypted
|
|
if not api_key:
|
|
logger.error(
|
|
f"[llm_routes][test_connection] Failed to decrypt API key for provider {provider_id}"
|
|
)
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail="Failed to decrypt API key. The provider may have been encrypted with a different encryption key. Please update the provider with a new API key.",
|
|
)
|
|
|
|
client = LLMClient(
|
|
provider_type=LLMProviderType(db_provider.provider_type),
|
|
api_key=api_key,
|
|
base_url=db_provider.base_url,
|
|
default_model=db_provider.default_model,
|
|
)
|
|
|
|
try:
|
|
await client.test_runtime_connection()
|
|
return {"success": True, "message": "Connection successful"}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
# #endregion test_connection
|
|
|
|
|
|
# #region test_provider_config [TYPE Function]
|
|
# @BRIEF Test connection with a provided configuration (not yet saved).
|
|
# @PRE: User is authenticated.
|
|
# @POST: Returns success status and message.
|
|
# @RELATION DEPENDS_ON -> [LLMClient]
|
|
# @RELATION DEPENDS_ON -> [LLMProviderConfig]
|
|
@router.post("/providers/test")
|
|
async def test_provider_config(
|
|
config: LLMProviderConfig, current_user: User = Depends(get_current_active_user)
|
|
):
|
|
"""
|
|
Test connection with a provided configuration.
|
|
"""
|
|
from ...plugins.llm_analysis.service import LLMClient
|
|
|
|
logger.reason(
|
|
f"Testing config for {config.name}",
|
|
extra={"src": "llm_routes.test_provider_config"},
|
|
)
|
|
|
|
# Check if API key is provided
|
|
if not config.api_key or config.api_key == "********":
|
|
raise HTTPException(
|
|
status_code=400, detail="API key is required for testing connection"
|
|
)
|
|
|
|
client = LLMClient(
|
|
provider_type=config.provider_type,
|
|
api_key=config.api_key,
|
|
base_url=config.base_url,
|
|
default_model=config.default_model,
|
|
)
|
|
|
|
try:
|
|
await client.test_runtime_connection()
|
|
return {"success": True, "message": "Connection successful"}
|
|
except Exception as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
# #endregion test_provider_config
|
|
|
|
# #endregion LlmRoutes
|