torin
b8d3cb4644
Some checks are pending
Build and Push Docker Images / changes (push) Waiting to run
Build and Push Docker Images / build-backend (push) Blocked by required conditions
Build and Push Docker Images / build-frontend (push) Blocked by required conditions
Build and Push Docker Images / build-admin-frontend (push) Blocked by required conditions
47 lines
1.1 KiB
Python
47 lines
1.1 KiB
Python
"""适配器基类定义。"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Generic, TypeVar
|
|
|
|
from pydantic import BaseModel
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class AdapterConfig(BaseModel):
|
|
"""适配器配置基类。"""
|
|
|
|
api_key: str
|
|
api_base: str | None = None
|
|
model: str | None = None
|
|
timeout_ms: int = 60000
|
|
max_retries: int = 3
|
|
extra_config: dict = {}
|
|
|
|
|
|
class BaseAdapter(ABC, Generic[T]):
|
|
"""适配器基类，所有供应商适配器必须继承此类。"""
|
|
|
|
# 子类必须定义
|
|
adapter_type: str # text / image / tts
|
|
adapter_name: str # text_primary / image_primary / tts_primary
|
|
|
|
def __init__(self, config: AdapterConfig):
|
|
self.config = config
|
|
|
|
@abstractmethod
|
|
async def execute(self, **kwargs) -> T:
|
|
"""执行适配器逻辑，返回结果。"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def health_check(self) -> bool:
|
|
"""健康检查，返回是否可用。"""
|
|
pass
|
|
|
|
@property
|
|
@abstractmethod
|
|
def estimated_cost(self) -> float:
|
|
"""预估单次调用成本 (USD)。"""
|
|
pass
|