别再重复造轮子了！一个优雅的模型调用SDK是如何拯救重复代码的

功能需求：
- 支持同步/异步调用
- 支持流式输出（SSE）
- 支持多模态输入（文本、图片）
- 支持函数调用（Function Calling）
- 自动重试与降级
- 请求超时控制

非功能需求：
- 易用性：5行代码内完成一次调用
- 扩展性：新增模型供应商不影响现有代码
- 可观测性：内置日志、指标、链路追踪
- 健壮性：优雅处理各种异常场景

┌─────────────────────────────────────────┐
│           ModelSDK (门面)                │
│  - chat()  - stream()  - function()    │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│         ModelFactory (工厂)              │
│    根据配置创建对应的模型客户端           │
└─────────────────┬───────────────────────┘
                  │
      ┌───────────┼───────────┐
      ▼           ▼           ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│OpenAI    │ │Claude    │ │Qianwen   │
│Adapter   │ │Adapter   │ │Adapter   │
└──────────┘ └──────────┘ └──────────┘
      │           │           │
      └───────────┼───────────┘
                  ▼
        ┌─────────────────┐
        │  BaseHTTPClient │ (统一HTTP层)
        │ - 重试机制       │
        │ - 熔断器         │
        │ - 日志拦截器     │
        └─────────────────┘

# 统一的消息格式
@dataclass
class Message:
    role: str  # system, user, assistant
    content: Union[str, List[ContentPart]]
    
@dataclass  
class ContentPart:
    type: str  # text, image_url
    text: Optional[str] = None
    image_url: Optional[str] = None

# 统一的请求参数
@dataclass
class ChatRequest:
    messages: List[Message]
    model: str
    temperature: float = 0.7
    max_tokens: int = 2000
    stream: bool = False
    tools: Optional[List[Tool]] = None
    
# 统一的响应格式
@dataclass
class ChatResponse:
    id: str
    content: str
    usage: TokenUsage
    finish_reason: str
    tool_calls: Optional[List[ToolCall]] = None

class BaseAdapter(ABC):
    @abstractmethod
    def to_provider_request(self, request: ChatRequest) -> dict:
        """将统一请求转换为供应商API格式"""
        pass
    
    @abstractmethod
    def to_unified_response(self, raw_response: dict) -> ChatResponse:
        """将供应商响应转换为统一格式"""
        pass

class OpenAIAdapter(BaseAdapter):
    def to_provider_request(self, request: ChatRequest) -> dict:
        return {
            "model": request.model,
            "messages": [m.__dict__ for m in request.messages],
            "temperature": request.temperature,
            "max_tokens": request.max_tokens,
            "stream": request.stream
        }
    
    def to_unified_response(self, raw_response: dict) -> ChatResponse:
        return ChatResponse(
            id=raw_response["id"],
            content=raw_response["choices"][0]["message"]["content"],
            usage=TokenUsage(**raw_response["usage"])
        )

class ResilientHTTPClient:
    def __init__(self, config: Config):
        self.retry_config = config.retry
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60
        )
        self.session = self._create_session()
        
    def _create_session(self):
        session = requests.Session()
        # 连接池配置
        adapter = HTTPAdapter(
            pool_connections=100,
            pool_maxsize=100,
            max_retries=0  # 我们自己管理重试
        )
        session.mount('https://', adapter)
        return session
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((
            TimeoutError, 
            ConnectionError,
            TooManyRequests
        ))
    )
    def post(self, url, data, headers):
        with self.circuit_breaker:
            response = self.session.post(
                url, 
                json=data, 
                headers=headers,
                timeout=(5, 30)  # (连接超时, 读取超时)
            )
            response.raise_for_status()
            return response.json()

def stream_chat(self, request: ChatRequest) -> Generator[str, None, None]:
    """流式调用，逐字返回"""
    provider_request = self.adapter.to_provider_request(request)
    provider_request["stream"] = True
    
    response = self.http_client.post_stream(
        url=self.config.api_url,
        data=provider_request
    )
    
    for line in response.iter_lines():
        if line:
            # 处理SSE格式
            if line.startswith(b'data: '):
                data = line[6:]
                if data != b'[DONE]':
                    chunk = json.loads(data)
                    # 实时解析并yield
                    yield self.adapter.extract_stream_content(chunk)

class ObservableModelClient:
    def chat(self, request: ChatRequest) -> ChatResponse:
        # 生成trace_id用于链路追踪
        trace_id = str(uuid.uuid4())
        
        # 记录开始时间
        start_time = time.time()
        
        # 结构化日志
        logger.info({
            "event": "model_call_start",
            "trace_id": trace_id,
            "model": request.model,
            "message_count": len(request.messages)
        })
        
        try:
            response = self._do_chat(request)
            
            # 记录指标
            duration = time.time() - start_time
            metrics.record_latency(
                model=request.model,
                duration=duration,
                tokens=response.usage.total_tokens
            )
            
            return response
        except Exception as e:
            # 记录错误
            metrics.record_error(model=request.model, error_type=type(e).__name__)
            raise

class ModelSDK:
    def __init__(self, config_path: str = None):
        self.config = Config.from_file(config_path) if config_path else Config()
        self.factory = ModelFactory(self.config)
        
    def chat(self, messages: List[Dict], model: str = None, **kwargs):
        """最简调用方式"""
        request = ChatRequest(
            messages=[Message(**m) for m in messages],
            model=model or self.config.default_model,
            **kwargs
        )
        
        client = self.factory.get_client(request.model)
        return client.chat(request)
    
    # 上下文管理器支持
    def __enter__(self):
        return self
    
    def __exit__(self, *args):
        self.close()

# 使用示例
with ModelSDK() as sdk:
    response = sdk.chat([
        {"role": "user", "content": "介绍一下自己"}
    ])
    print(response.content)

class FallbackClient:
    def chat(self, request: ChatRequest):
        for model in self.fallback_chain:
            try:
                client = self.factory.get_client(model)
                return client.chat(request)
            except Exception as e:
                logger.warning(f"Model {model} failed: {e}")
                continue
        raise AllModelsFailed()

class CachingClient:
    def chat(self, request: ChatRequest):
        cache_key = self._generate_cache_key(request)
        
        # 相同请求直接返回缓存
        if cache_key in self.cache:
            return self.cache[cache_key]
        
        response = self.client.chat(request)
        
        # 只缓存幂等请求
        if request.temperature == 0:
            self.cache[cache_key] = response
            
        return response

# 使用pytest进行测试
def test_openai_adapter():
    adapter = OpenAIAdapter()
    request = ChatRequest(messages=[Message(role="user", content="hi")])
    
    provider_req = adapter.to_provider_request(request)
    assert "messages" in provider_req
    assert provider_req["messages"][0]["content"] == "hi"
    
# 编写README和API文档
# 使用Sphinx或MkDocs自动生成文档