"""8512/8513 전용 AI Web Agent Console model profile registry. 이 registry는 모델 metadata만 관리한다. ``provider=oci``는 모델의 출처를 뜻하며 ``AI_WEB_AGENT_CONSOLE_MCP_PROVIDER``와 독립적이다. 따라서 기본 model profile이 GPT-5.5여도 현재 MCP 실행 경로는 계속 ``mock``일 수 있다. """ from __future__ import annotations from dataclasses import dataclass, field, replace import json import os from pathlib import Path import re from typing import Any, Mapping, Optional ROOT = Path(__file__).resolve().parents[1] REGISTRY_PATH = ROOT / "config" / "model_profiles.json" DEFAULT_MODEL_PROFILE_KEY = "gpt55_oci" MODEL_PROFILE_ENV = "AI_WEB_AGENT_CONSOLE_MODEL_PROFILE" MODEL_PROFILE_DEFAULT_ENV = "AI_WEB_AGENT_CONSOLE_MODEL_PROFILE_DEFAULT" EXISTING_MODEL_PROFILE_KEYS = ("grok43", "llama4_maverick", "llama33_70b") MODEL_PROFILE_ALIASES = { "gpt54_mini": "gpt54_mini_oci", "llama33": "llama33_70b", } EXPECTED_SOURCE_TAG = "poc_2-gpt55-oci-partial" EXPECTED_SOURCE_COMMIT = "7a3b37f175b65ed5eab1d8bf37c9bf6114e7558f" _EXPECTED_ANSWER_MODEL_ROUTES = { "gpt55_oci": ( "openai.gpt-5.5", "us-chicago-1", "OPENAI_GPT_5_5_CHAT", "OCI_REGIONAL_DEFAULT", ), "gpt54_mini_oci": ( "openai.gpt-5.4-mini", "us-chicago-1", "OPENAI_GPT_5_4_MINI_CHAT", "OCI_REGIONAL_DEFAULT", ), "grok43": ( "xai.grok-4.3", "us-chicago-1", "XAI_GROK_4_3_CHAT", "OCI_REGIONAL_DEFAULT", ), "llama4_maverick": ( "meta.llama-4-maverick-17b-128e-instruct-fp8", "us-chicago-1", "META_LLAMA_4_MAVERICK_CHAT", "OCI_REGIONAL_DEFAULT", ), "llama33_70b": ( "meta.llama-3.3-70b-instruct", "us-chicago-1", "META_LLAMA_3_3_70B_CHAT", "OCI_REGIONAL_DEFAULT", ), } _MODEL_KEY = re.compile(r"^[a-z][a-z0-9_]{1,63}$") _MODEL_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{1,127}$") _ANSWER_MODEL_ID_ALIAS = re.compile(r"^[A-Z][A-Z0-9_]{1,127}$") _OCI_REGION = re.compile(r"^[a-z]{2}-[a-z0-9-]+-[1-9][0-9]*$") _OCI_REGIONAL_ENDPOINT = re.compile( r"^https://inference\.generativeai\." r"(?P[a-z]{2}-[a-z0-9-]+-[1-9][0-9]*)\.oci\.oraclecloud\.com$" ) _ORACLE_IDENTIFIER = re.compile(r"^[A-Z][A-Z0-9_$#]{0,127}$") _SOURCE_TAG = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{1,127}$") _VERIFICATION_STATUSES = frozenset( {"VERIFIED", "PARTIAL_VERIFIED", "VERIFIED_WITH_WARNINGS"} ) _REQUIRED_PROFILE_FIELDS = ( "model_key", "display_name", "provider", "model_id", "answer_model_id_alias", "answer_model_region", "answer_model_endpoint_mode", "poc2_select_ai_profile", "poc2_native_agent_team", "verification_status", "default_for_console", "source_tag", "notes", ) _PROFILE_ROUTE_ENV_KEYS = { "gpt55_oci": ( "AI_WEB_AGENT_CONSOLE_LLM_GPT55_OCI_MODEL_ID", "AI_WEB_AGENT_CONSOLE_LLM_GPT55_OCI_REGION", "AI_WEB_AGENT_CONSOLE_LLM_GPT55_OCI_ENDPOINT", ), "gpt54_mini_oci": ( "AI_WEB_AGENT_CONSOLE_LLM_GPT54_MINI_OCI_MODEL_ID", "AI_WEB_AGENT_CONSOLE_LLM_GPT54_MINI_OCI_REGION", "AI_WEB_AGENT_CONSOLE_LLM_GPT54_MINI_OCI_ENDPOINT", ), "grok43": ( "AI_WEB_AGENT_CONSOLE_LLM_GROK43_MODEL_ID", "AI_WEB_AGENT_CONSOLE_LLM_GROK43_REGION", "AI_WEB_AGENT_CONSOLE_LLM_GROK43_ENDPOINT", ), "llama4_maverick": ( "AI_WEB_AGENT_CONSOLE_LLM_LLAMA4_MAVERICK_MODEL_ID", "AI_WEB_AGENT_CONSOLE_LLM_LLAMA4_MAVERICK_REGION", "AI_WEB_AGENT_CONSOLE_LLM_LLAMA4_MAVERICK_ENDPOINT", ), "llama33_70b": ( "AI_WEB_AGENT_CONSOLE_LLM_LLAMA33_70B_MODEL_ID", "AI_WEB_AGENT_CONSOLE_LLM_LLAMA33_70B_REGION", "AI_WEB_AGENT_CONSOLE_LLM_LLAMA33_70B_ENDPOINT", ), } def _regional_endpoint(region: str) -> str: return "https://inference.generativeai.%s.oci.oraclecloud.com" % region def _endpoint_host_alias(region: str) -> str: return "OCI_GENAI_INFERENCE_%s" % region.upper().replace("-", "_") def _env_override( environ: Mapping[str, str], key: str, default: str, ) -> str: if key not in environ: return default value = environ.get(key) if not isinstance(value, str) or not value.strip(): raise ValueError("answer model route override is invalid") return value.strip() @dataclass(frozen=True) class ModelProfile: """UI와 workflow가 공유하는 비밀값 없는 model metadata.""" model_key: str display_name: str provider: str model_id: str answer_model_id_alias: str answer_model_region: str answer_model_endpoint_mode: str answer_model_endpoint: str = field(repr=False) poc2_select_ai_profile: str poc2_native_agent_team: str verification_status: str default_for_console: bool source_tag: str notes: str display_order: int = 999 @classmethod def from_mapping(cls, value: Mapping[str, Any]) -> "ModelProfile": missing = [name for name in _REQUIRED_PROFILE_FIELDS if name not in value] if missing: raise ValueError("AI Web Agent Console model profile fields are missing") if not isinstance(value.get("default_for_console"), bool): raise ValueError("default_for_console must be boolean") order = value.get("display_order", 999) if isinstance(order, bool) or not isinstance(order, int) or order < 0: raise ValueError("model profile display_order is invalid") answer_model_region = str(value["answer_model_region"]).strip().lower() profile = cls( model_key=str(value["model_key"]).strip().lower(), display_name=str(value["display_name"]).strip(), provider=str(value["provider"]).strip().lower(), model_id=str(value["model_id"]).strip(), answer_model_id_alias=str(value["answer_model_id_alias"]) .strip() .upper(), answer_model_region=answer_model_region, answer_model_endpoint_mode=str( value["answer_model_endpoint_mode"] ).strip().upper(), answer_model_endpoint=_regional_endpoint(answer_model_region), poc2_select_ai_profile=str(value["poc2_select_ai_profile"]) .strip() .upper(), poc2_native_agent_team=str(value["poc2_native_agent_team"]) .strip() .upper(), verification_status=str(value["verification_status"]).strip().upper(), default_for_console=value["default_for_console"], source_tag=str(value["source_tag"]).strip(), notes=str(value["notes"]).strip(), display_order=order, ) if not _MODEL_KEY.fullmatch(profile.model_key): raise ValueError("model profile key is invalid") if not profile.display_name or len(profile.display_name) > 128: raise ValueError("model profile display name is invalid") if profile.provider != "oci": raise ValueError("unsupported model provider") if not _MODEL_ID.fullmatch(profile.model_id): raise ValueError("model id is invalid") if not _ANSWER_MODEL_ID_ALIAS.fullmatch(profile.answer_model_id_alias): raise ValueError("answer model id alias is invalid") if not _OCI_REGION.fullmatch(profile.answer_model_region): raise ValueError("answer model region is invalid") if profile.answer_model_endpoint_mode != "OCI_REGIONAL_DEFAULT": raise ValueError("answer model endpoint mode is invalid") endpoint_match = _OCI_REGIONAL_ENDPOINT.fullmatch( profile.answer_model_endpoint ) if ( endpoint_match is None or endpoint_match.group("region") != profile.answer_model_region ): raise ValueError("answer model endpoint is invalid") if not _ORACLE_IDENTIFIER.fullmatch(profile.poc2_select_ai_profile): raise ValueError("PoC_2 Select AI profile mapping is invalid") if not _ORACLE_IDENTIFIER.fullmatch(profile.poc2_native_agent_team): raise ValueError("PoC_2 Native Agent team mapping is invalid") if profile.verification_status not in _VERIFICATION_STATUSES: raise ValueError("model verification status is invalid") if not _SOURCE_TAG.fullmatch(profile.source_tag): raise ValueError("model profile source tag is invalid") if not profile.notes or len(profile.notes) > 1_000: raise ValueError("model profile notes are invalid") return profile def with_answer_route_overrides( self, environ: Mapping[str, str], ) -> "ModelProfile": """Apply only this profile's validated, non-secret OCI route settings.""" keys = _PROFILE_ROUTE_ENV_KEYS.get(self.model_key) if keys is None: raise ValueError("answer model route is not registered") model_id = _env_override(environ, keys[0], self.model_id) region = _env_override(environ, keys[1], self.answer_model_region).lower() endpoint = _env_override( environ, keys[2], _regional_endpoint(region), ) if not _MODEL_ID.fullmatch(model_id): raise ValueError("answer model route override is invalid") if not _OCI_REGION.fullmatch(region): raise ValueError("answer model route override is invalid") endpoint_match = _OCI_REGIONAL_ENDPOINT.fullmatch(endpoint) if endpoint_match is None or endpoint_match.group("region") != region: raise ValueError("answer model route override is invalid") return replace( self, model_id=model_id, answer_model_region=region, answer_model_endpoint=endpoint, ) def public_metadata(self) -> dict[str, object]: """System Details에 투영 가능한 비밀값 없는 metadata를 반환한다.""" return { "model_key": self.model_key, "display_name": self.display_name, "provider": self.provider, "answer_model_id_alias": self.answer_model_id_alias, "answer_model_region": self.answer_model_region, "answer_model_endpoint_mode": self.answer_model_endpoint_mode, "answer_model_endpoint_host_alias": _endpoint_host_alias( self.answer_model_region ), "poc2_select_ai_profile": self.poc2_select_ai_profile, "poc2_native_agent_team": self.poc2_native_agent_team, "verification_status": self.verification_status, "default_for_console": self.default_for_console, "source_tag": self.source_tag, "notes": self.notes, } @dataclass(frozen=True) class ModelProfileRegistry: """검증된 AI Web Agent Console model profile 집합.""" profiles: tuple[ModelProfile, ...] default_model_profile: str registry_name: str source_commit: str schema_version: int = 1 def by_key(self, model_key: object) -> ModelProfile: candidate = str(model_key or "").strip().lower() candidate = MODEL_PROFILE_ALIASES.get(candidate, candidate) for profile in self.profiles: if profile.model_key == candidate: return profile # 사용자 입력이나 환경변수 원문을 오류에 반사하지 않는다. raise ValueError("model profile is not registered") @property def default_profile(self) -> ModelProfile: return self.by_key(self.default_model_profile) def selector_options(self) -> tuple[ModelProfile, ...]: return tuple(sorted(self.profiles, key=lambda item: item.display_order)) def resolve( self, requested: object = None, *, environ: Optional[Mapping[str, str]] = None, ) -> ModelProfile: """명시 요청은 엄격히 검증하고 환경 기본값은 안전하게 fallback한다.""" source = os.environ if environ is None else environ if requested is not None and str(requested).strip(): return self.by_key(requested).with_answer_route_overrides(source) for name in (MODEL_PROFILE_ENV, MODEL_PROFILE_DEFAULT_ENV): candidate = source.get(name) if not candidate or not candidate.strip(): continue try: profile = self.by_key(candidate) except ValueError: continue return profile.with_answer_route_overrides(source) return self.default_profile.with_answer_route_overrides(source) def load_model_registry(path: Path = REGISTRY_PATH) -> ModelProfileRegistry: """JSON registry를 매 호출마다 검증해 반환한다.""" try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ValueError("AI Web Agent Console model profile registry cannot be loaded") from exc if not isinstance(payload, Mapping): raise ValueError("AI Web Agent Console model profile registry must be an object") raw_profiles = payload.get("profiles") if not isinstance(raw_profiles, list) or not raw_profiles: raise ValueError("AI Web Agent Console model profile registry has no profiles") profiles = tuple( ModelProfile.from_mapping(item) for item in raw_profiles if isinstance(item, Mapping) ) if len(profiles) != len(raw_profiles): raise ValueError("AI Web Agent Console model profile registry contains an invalid profile") keys = tuple(item.model_key for item in profiles) if len(set(keys)) != len(keys): raise ValueError("AI Web Agent Console model profile keys must be unique") if len({item.display_name for item in profiles}) != len(profiles): raise ValueError("AI Web Agent Console model profile display names must be unique") defaults = tuple(item.model_key for item in profiles if item.default_for_console) configured_default = str(payload.get("default_model_profile") or "").strip().lower() if defaults != (configured_default,): raise ValueError("AI Web Agent Console model profile default is inconsistent") if configured_default != DEFAULT_MODEL_PROFILE_KEY: raise ValueError("AI Web Agent Console GPT-5.5 default contract is not satisfied") if not set(EXISTING_MODEL_PROFILE_KEYS).issubset(keys): raise ValueError("existing AI Web Agent Console selector models are missing") actual_answer_routes = { item.model_key: ( item.model_id, item.answer_model_region, item.answer_model_id_alias, item.answer_model_endpoint_mode, ) for item in profiles } if actual_answer_routes != _EXPECTED_ANSWER_MODEL_ROUTES: raise ValueError("AI Web Agent Console answer model route mapping is inconsistent") if str(payload.get("source_commit") or "").strip() != EXPECTED_SOURCE_COMMIT: raise ValueError("PoC_2 source commit is inconsistent") if any(item.source_tag != EXPECTED_SOURCE_TAG for item in profiles): raise ValueError("PoC_2 source tag is inconsistent") if payload.get("schema_version") != 1: raise ValueError("unsupported AI Web Agent Console model profile registry schema") registry_name = str(payload.get("registry_name") or "").strip() if not registry_name: raise ValueError("AI Web Agent Console model profile registry name is missing") return ModelProfileRegistry( profiles=profiles, default_model_profile=configured_default, registry_name=registry_name, source_commit=EXPECTED_SOURCE_COMMIT, ) def resolve_model_profile( requested: object = None, *, environ: Optional[Mapping[str, str]] = None, ) -> ModelProfile: return load_model_registry().resolve(requested, environ=environ) def resolve_model_profile_key( requested: object = None, *, environ: Optional[Mapping[str, str]] = None, ) -> str: return resolve_model_profile(requested, environ=environ).model_key def is_registered_model_profile(value: object) -> bool: try: load_model_registry().by_key(value) except ValueError: return False return True __all__ = [ "DEFAULT_MODEL_PROFILE_KEY", "EXISTING_MODEL_PROFILE_KEYS", "EXPECTED_SOURCE_COMMIT", "EXPECTED_SOURCE_TAG", "MODEL_PROFILE_DEFAULT_ENV", "MODEL_PROFILE_ENV", "MODEL_PROFILE_ALIASES", "ModelProfile", "ModelProfileRegistry", "REGISTRY_PATH", "is_registered_model_profile", "load_model_registry", "resolve_model_profile", "resolve_model_profile_key", ]