Bloom_Ware / core /ai_client.py
LawrenceBai's picture
feat: overhaul MCP architecture with structured tool schemas, comprehensive care-mode skill definitions, and enhanced test coverage for pipelines and service integration.
79df050
Raw
History Blame Contribute Delete
2.39 kB
"""
統一 OpenAI 客戶端管理
單一真理來源,避免重複初始化
"""
import logging
from typing import Optional
from core.config import settings
from core.logging import get_logger
logger = get_logger("core.ai_client")
# 全域 OpenAI 客戶端
_openai_client = None
_initialized = False
def _normalize_openai_base_url(base_url: Optional[str]) -> Optional[str]:
"""Normalize custom OpenAI-compatible base URLs for the Python SDK."""
if not base_url:
return None
normalized = base_url.rstrip("/")
if normalized.endswith("/v1"):
return normalized
return f"{normalized}/v1"
def get_openai_client():
"""
取得 OpenAI 客戶端(單例模式)
Returns:
OpenAI 客戶端實例,若初始化失敗則返回 None
"""
global _openai_client, _initialized
if _initialized:
return _openai_client
try:
from openai import OpenAI
api_key = settings.OPENAI_API_KEY
if not api_key:
logger.error("❌ OpenAI API Key 未設定")
_initialized = True
return None
client_kwargs = {
"api_key": api_key,
"timeout": float(settings.OPENAI_TIMEOUT),
"max_retries": 3,
}
normalized_base_url = _normalize_openai_base_url(settings.OPENAI_BASE_URL)
if normalized_base_url:
client_kwargs["base_url"] = normalized_base_url
_openai_client = OpenAI(**client_kwargs)
_initialized = True
logger.info("✅ OpenAI 客戶端初始化成功")
return _openai_client
except ImportError:
logger.error("❌ 無法導入 OpenAI SDK")
_initialized = True
return None
except Exception as e:
logger.error(f"❌ OpenAI 客戶端初始化失敗: {e}")
_initialized = True
return None
def reset_client() -> None:
"""
重置客戶端(用於測試或重新初始化)
"""
global _openai_client, _initialized
_openai_client = None
_initialized = False
logger.info("OpenAI 客戶端已重置")
def is_available() -> bool:
"""
檢查 OpenAI 服務是否可用
Returns:
True 如果客戶端已初始化且可用
"""
client = get_openai_client()
return client is not None
# 便捷別名
client = property(lambda self: get_openai_client())