"""LLM 라우터 — 파이프라인 단계별 모델 선택 통합 인터페이스. 각 파이프라인 단계(step)에 대해 프로젝트별로 다른 provider/model을 지정할 수 있으며, GeminiTextClient와 OpenAIClient를 동일한 인터페이스로 호출한다. """ import json import logging from typing import Any, Dict, List, Optional from app.core.config import settings logger = logging.getLogger(__name__) # ── 파이프라인 단계 정의 ── # key: 코드에서 사용하는 step 이름 # label: UI 표시용 # default_provider: 기본 provider # default_model: 기본 model ID PIPELINE_STEPS = { "entity_style": { "label": "요소 추출 — 스타일 분석", "default_provider": "gemini", "default_model": "gemini-3.1-pro-preview", "category": "analysis", }, "entity_names": { "label": "요소 추출 — 이름 목록", "default_provider": "gemini", "default_model": "gemini-3.1-pro-preview", "category": "analysis", }, "entity_review": { "label": "요소 추출 — GPT 리뷰", "default_provider": "openai", "default_model": "gpt-5.4", "category": "analysis", }, "entity_detail": { "label": "요소 추출 — 상세 정보", "default_provider": "gemini", "default_model": "gemini-3.1-pro-preview", "category": "analysis", }, "scene_segmentation": { "label": "씬 세그먼테이션", "default_provider": "gemini", "default_model": "gemini-3.1-flash-lite-preview", "category": "analysis", }, "scene_split": { "label": "큰 씬 분할", "default_provider": "gemini", "default_model": "gemini-3-flash-preview", "category": "analysis", }, "scene_dependency": { "label": "씬 연관 분석", "default_provider": "gemini", "default_model": "gemini-3.1-pro-preview", "category": "analysis", }, "outlook_extraction": { "label": "아웃룩 추출", "default_provider": "gemini", "default_model": "gemini-3.1-pro-preview", "category": "analysis", }, "scene_detail": { "label": "씬 상세 분석", "default_provider": "openai", "default_model": "gpt-5.4", "category": "analysis", }, "world_guide": { "label": "월드 가이드 생성", "default_provider": "openai", "default_model": "gpt-5.4", "category": "image", }, "prompt_translation": { "label": "T2I 프롬프트 번역", "default_provider": "gemini", "default_model": "gemini-3-flash-preview", "category": "image", }, "prompt_sanitize": { "label": "프롬프트 안전화", "default_provider": "openai", "default_model": "gpt-5.4", "category": "image", }, "project_summary": { "label": "프로젝트 요약", "default_provider": "gemini", "default_model": "gemini-3.1-pro-preview", "category": "analysis", }, } # ── 사용 가능한 모델 목록 ── AVAILABLE_MODELS = { "openai": [ {"id": "gpt-5.4", "label": "GPT-5.4", "tier": "frontier", "context": "1M"}, {"id": "gpt-5.4-mini", "label": "GPT-5.4 Mini", "tier": "mid", "context": "400K"}, {"id": "gpt-5.4-nano", "label": "GPT-5.4 Nano", "tier": "lite", "context": "400K"}, {"id": "gpt-4.1", "label": "GPT-4.1", "tier": "frontier", "context": "1M"}, {"id": "gpt-4.1-mini", "label": "GPT-4.1 Mini", "tier": "mid", "context": "1M"}, {"id": "gpt-4.1-nano", "label": "GPT-4.1 Nano", "tier": "lite", "context": "1M"}, {"id": "o3", "label": "o3 (Reasoning)", "tier": "reasoning", "context": "200K"}, {"id": "o4-mini", "label": "o4-mini (Reasoning)", "tier": "reasoning", "context": "200K"}, ], "gemini": [ {"id": "gemini-3.1-pro-preview", "label": "Gemini 3.1 Pro", "tier": "frontier", "context": "1M"}, {"id": "gemini-3-flash-preview", "label": "Gemini 3 Flash", "tier": "mid", "context": "1M"}, {"id": "gemini-3.1-flash-lite-preview", "label": "Gemini 3.1 Flash Lite", "tier": "lite", "context": "1M"}, {"id": "gemini-2.5-pro", "label": "Gemini 2.5 Pro", "tier": "frontier", "context": "1M"}, {"id": "gemini-2.5-flash", "label": "Gemini 2.5 Flash", "tier": "mid", "context": "1M"}, {"id": "gemini-2.5-flash-lite", "label": "Gemini 2.5 Flash Lite", "tier": "lite", "context": "1M"}, ], } def get_step_config(step: str, project_llm_config: Optional[Dict] = None) -> Dict[str, str]: """단계별 provider + model 반환. project_llm_config가 있으면 우선, 없으면 PIPELINE_STEPS 기본값. """ defaults = PIPELINE_STEPS.get(step, {}) if project_llm_config and step in project_llm_config: override = project_llm_config[step] return { "provider": override.get("provider", defaults.get("default_provider", "gemini")), "model": override.get("model", defaults.get("default_model", "")), } return { "provider": defaults.get("default_provider", "gemini"), "model": defaults.get("default_model", ""), } def get_llm_client(step: str, project_llm_config: Optional[Dict] = None): """단계에 맞는 LLM 클라이언트 인스턴스 반환. Returns: (client, provider, model) tuple - client: GeminiTextClient or OpenAIClient instance - provider: "gemini" or "openai" - model: model ID string """ config = get_step_config(step, project_llm_config) provider = config["provider"] model = config["model"] if provider == "openai": from app.modules.llm.openai_client import OpenAIClient client = OpenAIClient(model=model) return client, provider, model else: from app.modules.llm.gemini_text_client import GeminiTextClient client = GeminiTextClient(model=model) return client, provider, model def call_structured( step: str, system_prompt: str, user_prompt: str, response_schema: Dict[str, Any], project_llm_config: Optional[Dict] = None, schema_name: str = "response", **kwargs, ) -> Dict[str, Any]: """통합 structured output 호출. provider에 따라 적절한 클라이언트로 라우팅. GPT strict schema는 additionalProperties: false가 필요하므로 자동 추가. """ client, provider, model = get_llm_client(step, project_llm_config) logger.info("LLM call: step=%s provider=%s model=%s", step, provider, model) if provider == "openai": # GPT strict schema: additionalProperties 자동 추가 schema = _ensure_additional_properties(response_schema) return client.generate_structured( system_prompt=system_prompt, user_prompt=user_prompt, response_schema=schema, schema_name=schema_name, **kwargs, ) else: return client.send_structured( user_message=user_prompt, response_schema=response_schema, system_instruction=system_prompt, **kwargs, ) def call_text( step: str, system_prompt: str, user_prompt: str, project_llm_config: Optional[Dict] = None, temperature: float = 0.2, ) -> str: """통합 free-text 호출.""" client, provider, model = get_llm_client(step, project_llm_config) logger.info("LLM text call: step=%s provider=%s model=%s", step, provider, model) if provider == "openai": # OpenAI free text — generate_structured with no schema # Fallback: use Responses API directly result = client.generate_structured( system_prompt=system_prompt, user_prompt=user_prompt, response_schema={"type": "object", "properties": {"text": {"type": "string"}}, "required": ["text"], "additionalProperties": False}, schema_name="free_text", ) return result.get("text", "") else: return client.send( user_message=user_prompt, system_instruction=system_prompt, temperature=temperature, ) def _ensure_additional_properties(schema: Dict[str, Any]) -> Dict[str, Any]: """GPT strict schema 호환: 모든 object에 additionalProperties: false 추가.""" import copy s = copy.deepcopy(schema) _add_ap(s) return s def _add_ap(obj: Any) -> None: """재귀적으로 additionalProperties: false 추가.""" if not isinstance(obj, dict): return if obj.get("type") == "object" and "properties" in obj: obj.setdefault("additionalProperties", False) for v in obj["properties"].values(): _add_ap(v) if obj.get("type") == "array" and "items" in obj: _add_ap(obj["items"])