← 返回首页
🧠

SDK设计:构建LLM应用SDK的设计原则

📂 llm ⏱ 2 min 361 words

--- title: "SDK设计:构建LLM应用SDK的设计原则" description: "探讨构建LLM应用SDK的核心设计原则,涵盖接口设计、配置管理、中间件机制、可扩展性等关键架构要素" tags: ["SDK设计", "架构设计", "LLM SDK", "接口设计"] category: "llm" icon: "🧠"

SDK设计:构建LLM应用SDK的设计原则

为什么需要LLM SDK

直接使用HTTP请求调用LLM API虽然可行,但缺乏类型安全、错误处理、重试机制等基础能力。一个好的SDK应封装这些复杂性,让开发者专注于业务逻辑。

核心设计原则

1. 简洁的API入口

SDK应提供最少的概念和最直观的调用方式:

# 简洁的客户端初始化
client = LLMClient(api_key="your-key", provider="openai")

# 一行完成调用
response = client.chat("你好,请介绍一下自己")

# 支持流式输出
for chunk in client.stream_chat("写一首诗"):
    print(chunk.content, end="")

2. 统一的Provider抽象

不同LLM提供商应通过统一接口调用:

from abc import ABC, abstractmethod

class BaseProvider(ABC):
    @abstractmethod
    def chat(self, messages: list[Message], **kwargs) -> Response:
        pass

    @abstractmethod
    def stream_chat(self, messages: list[Message], **kwargs):
        pass

class OpenAIProvider(BaseProvider):
    def chat(self, messages, **kwargs):
        return self.client.chat.completions.create(
            model=kwargs.get("model", "gpt-4o"),
            messages=[m.to_dict() for m in messages],
            **kwargs
        )

class AnthropicProvider(BaseProvider):
    def chat(self, messages, **kwargs):
        return self.client.messages.create(
            model=kwargs.get("model", "claude-sonnet-4-20250514"),
            messages=[m.to_dict() for m in messages],
            **kwargs
        )

3. 灵活的配置系统

支持多层级配置,优先级从高到低:环境变量 → 配置文件 → 默认值:

from pydantic_settings import BaseSettings

class LLMConfig(BaseSettings):
    api_key: str = ""
    base_url: str = "https://api.openai.com/v1"
    model: str = "gpt-4o"
    temperature: float = 0.7
    max_tokens: int = 4096
    timeout: int = 30
    retry_count: int = 3

    model_config = {
        "env_prefix": "LLM_",
        "env_file": ".env",
        "env_file_encoding": "utf-8",
    }

# 使用时自动从环境变量 LLM_API_KEY 等读取
config = LLMConfig()
client = LLMClient(config=config)

4. 中间件机制

通过中间件实现横切关注点的解耦:

class Middleware(ABC):
    @abstractmethod
    async def before_request(self, request: Request) -> Request:
        return request

    @abstractmethod
    async def after_response(self, response: Response) -> Response:
        return response

class LoggingMiddleware(Middleware):
    async def before_request(self, request):
        logger.info(f"LLM请求: {request.model}, tokens: ~{len(request.messages)}")
        return request

    async def after_response(self, response):
        logger.info(f"LLM响应: {response.usage}")
        return response

class RateLimitMiddleware(Middleware):
    def __init__(self, max_rpm: int = 60):
        self.limiter = RateLimiter(max_rpm)

    async def before_request(self, request):
        await self.limiter.acquire()
        return request

client = LLMClient(config)
client.use(LoggingMiddleware())
client.use(RateLimitMiddleware(max_rpm=60))

5. 类型安全

使用类型系统确保API调用的正确性:

from dataclasses import dataclass
from enum import Enum

class Role(str, Enum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"

@dataclass
class Message:
    role: Role
    content: str

@dataclass
class ChatResponse:
    content: str
    model: str
    usage: Usage
    finish_reason: str

    @property
    def is_complete(self) -> bool:
        return self.finish_reason == "stop"

6. 异步优先

现代SDK应默认支持异步,同时提供同步封装:

import asyncio

class AsyncLLMClient:
    async def chat(self, messages: list[Message], **kwargs) -> ChatResponse:
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/chat/completions",
                json=self._build_payload(messages, **kwargs),
                headers=self._headers(),
                timeout=self.timeout,
            ) as resp:
                data = await resp.json()
                return self._parse_response(data)

class LLMClient:
    """同步客户端,内部调用异步实现"""
    def __init__(self, *args, **kwargs):
        self._async_client = AsyncLLMClient(*args, **kwargs)
        self._loop = asyncio.new_event_loop()

    def chat(self, messages, **kwargs):
        return self._loop.run_until_complete(
            self._async_client.chat(messages, **kwargs)
        )

错误处理设计

SDK应定义清晰的异常层次:

class LLMError(Exception):
    """SDK基础异常"""
    pass

class AuthenticationError(LLMError):
    """API密钥无效"""
    pass

class RateLimitError(LLMError):
    """请求频率超限"""
    pass

class ModelNotFoundError(LLMError):
    """模型不存在"""
    pass

class ContentFilterError(LLMError):
    """内容被安全过滤"""
    pass

总结

优秀的LLM SDK应做到:简洁统一的API设计、灵活的配置管理、可扩展的中间件机制、完整的类型安全、异步优先的执行模型,以及清晰的错误处理体系。