""" NullAI システム設定API モデル、ドメイン、DBなどの設定をUIから動的に変更するためのAPI。 """ from fastapi import APIRouter, HTTPException, Depends, File, UploadFile from pydantic import BaseModel from typing import List, Dict, Any, Optional import os # For file operations import logging from backend.app.config import app_config_manager, app_model_router, ModelConfig as PydanticModelConfig router = APIRouter() logger = logging.getLogger(__name__) # モデル保存ディレクトリ (設定可能にするべきだが、一旦ハードコード) GGUF_MODELS_DIR = os.path.join(app_config_manager.base_dir, "models", "gguf") os.makedirs(GGUF_MODELS_DIR, exist_ok=True) # ディレクトリが存在しない場合は作成 # --- APIエンドポイント --- @router.get("/models", response_model=List[PydanticModelConfig]) async def get_models_config(): """ 現在利用可能なモデルの設定一覧を取得する。 """ models = list(app_config_manager.models.values()) return models class DomainConfigResponse(BaseModel): # Renamed to avoid conflict domain_id: str name: str # Pydantic models for the new engine management endpoints class EngineStatusResponse(PydanticModelConfig): status: str # 'master' | 'apprentice' | 'available' | 'retired' unique_id: Optional[str] = None # For apprentice engines class EngineSwapRequest(BaseModel): apprentice_model_id: str class EnginePromoteRequest(BaseModel): apprentice_model_id: str class NewApprenticeResponse(BaseModel): new_apprentice_id: str message: str @router.get("/domains", response_model=List[DomainConfigResponse]) async def get_domains_config(): """ 現在利用可能なドメイン（知識DB）の設定一覧を取得する。 """ domains = list(app_config_manager.domains.values()) return [DomainConfigResponse(domain_id=d.get("domain_id"), name=d.get("name")) for d in domains] @router.post("/models", response_model=PydanticModelConfig) async def register_new_model(model_data: Dict[str, Any]): """ 新しいモデルを登録する。 """ new_model = app_config_manager.add_model(model_data) if not new_model: raise HTTPException(status_code=400, detail="Failed to add new model. Check data format.") return new_model @router.get("/engines", response_model=List[EngineStatusResponse]) async def get_all_engines(): """ システムに登録されている全てのエンジン（モデル）とそのステータスを取得する。 """ all_models = [] current_master = app_model_router.get_master_model() current_apprentice = app_model_router.get_apprentice_model() # Assuming app_model_router can provide info about all apprentices (even inactive ones) # This might need a new method in app_model_router if not available all_managed_engines = app_model_router.get_all_managed_engines() # This method needs to be implemented in ModelRouter for model_id, model_config in app_config_manager.models.items(): status = 'available' unique_id = None if current_master and model_id == current_master.model_id: status = 'master' elif current_apprentice and model_id == current_apprentice.model_id: status = 'apprentice' # Assuming app_model_router keeps track of unique_id for apprentices apprentice_info = next((e for e in all_managed_engines if e['config']['model_id'] == model_id and e['unique_id']), None) if apprentice_info: unique_id = apprentice_info['unique_id'] # else: default to 'available' or 'retired' if explicitly managed by router but not active # Consider models that were once master and now retired # This logic needs to be enhanced if app_model_router tracks retired models all_models.append( EngineStatusResponse( **model_config.dict(), # Unpack PydanticModelConfig fields status=status, unique_id=unique_id ) ) return all_models class ActiveEngines(BaseModel): master_engine_id: str apprentice_engine_id: Optional[str] = None class ActiveEnginesResponse(BaseModel): master_engine_id: str apprentice_engine_id: Optional[str] = None @router.get("/engines/active", response_model=ActiveEnginesResponse) async def get_active_engines(): """ 現在設定されているマスターエンジンとアプレンティスエンジンのIDを取得する。 """ master_model = app_model_router.get_master_model() apprentice_model = app_model_router.get_apprentice_model() return ActiveEnginesResponse( master_engine_id=master_model.model_id if master_model else None, apprentice_engine_id=apprentice_model.model_id if apprentice_model else None ) @router.post("/engines/active") async def set_active_engines(engines: ActiveEngines): """ マスターエンジンとアプレンティスエンジンを設定する。 """ if not app_model_router.set_master_model(engines.master_engine_id): raise HTTPException(status_code=400, detail=f"Master model '{engines.master_engine_id}' not found.") if engines.apprentice_engine_id: if not app_model_router.set_apprentice_model(engines.apprentice_engine_id): raise HTTPException(status_code=400, detail=f"Apprentice model '{engines.apprentice_engine_id}' not found.") else: app_model_router.set_apprentice_model(None) # Clear apprentice return {"status": "success", "message": f"Active engines set to {engines.master_engine_id} (Master) and {engines.apprentice_engine_id or 'None'} (Apprentice)."} @router.post("/engines/swap", response_model=ActiveEnginesResponse) async def swap_engines(request: EngineSwapRequest): """ 現在の師匠と指定した弟子を入れ替える。 """ if not app_model_router.swap_engines(request.apprentice_model_id): raise HTTPException(status_code=400, detail=f"Failed to swap engines with apprentice '{request.apprentice_model_id}'. Check logs.") master_model = app_model_router.get_master_model() apprentice_model = app_model_router.get_apprentice_model() return ActiveEnginesResponse( master_engine_id=master_model.model_id if master_model else None, apprentice_engine_id=apprentice_model.model_id if apprentice_model else None ) @router.post("/engines/promote", response_model=ActiveEnginesResponse) async def promote_apprentice(request: EnginePromoteRequest): """ 指定した弟子を師匠に昇格させ、現在の師匠を引退させる。 """ if not app_model_router.promote_apprentice(request.apprentice_model_id): raise HTTPException(status_code=400, detail=f"Failed to promote apprentice '{request.apprentice_model_id}'. Check logs.") master_model = app_model_router.get_master_model() apprentice_model = app_model_router.get_apprentice_model() return ActiveEnginesResponse( master_engine_id=master_model.model_id if master_model else None, apprentice_engine_id=apprentice_model.model_id if apprentice_model else None ) @router.post("/engines/apprentice/new", response_model=NewApprenticeResponse) async def create_new_apprentice_endpoint(): """ 新しい「空っぽの弟子」推論エンジンを生成し、登録する。 """ new_apprentice_data = app_model_router.create_new_apprentice() if not new_apprentice_data: raise HTTPException(status_code=500, detail="Failed to create new apprentice engine.") return NewApprenticeResponse( new_apprentice_id=new_apprentice_data["config"]["model_id"], message=f"New apprentice '{new_apprentice_data['config']['display_name']}' created successfully." ) class ActiveDomain(BaseModel): domain_id: str @router.post("/domains/active") async def set_active_domain(domain: ActiveDomain): """ 推論に使用するアクティブなドメイン（知識DB）を設定する。 """ if not app_config_manager.set_active_domain(domain.domain_id): raise HTTPException(status_code=400, detail=f"Domain '{domain.domain_id}' not found or failed to set.") # ModelRouterにもアクティブドメインの変更を通知 app_model_router.set_active_domain_id(domain.domain_id) return {"status": "success", "message": f"Active domain set to {domain.domain_id}."} @router.post("/upload-gguf") async def upload_gguf_model(file: UploadFile = File(...)): """ GGUFモデルファイルをアップロードする。 """ file_path = os.path.join(GGUF_MODELS_DIR, file.filename) try: with open(file_path, "wb") as buffer: while True: chunk = await file.read(1024 * 1024) # 1MBずつ読み込み if not chunk: break buffer.write(chunk) logger.info(f"Uploaded GGUF file saved to: {file_path}") return {"filename": file.filename, "path": file_path, "message": "GGUF model uploaded successfully."} except Exception as e: logger.error(f"Failed to upload GGUF file: {e}") raise HTTPException(status_code=500, detail=f"Failed to upload file: {e}")