模型 API 实战

生产里 LLM 多通过 HTTP API 调用：鉴权、流式、重试、计费都落在你的客户端封装。本课覆盖 OpenAI、Gemini、Hugging Face Inference 的常见模式——对应 ai-engineer 的 model APIs 区域。

学习目标

• 统一封装 chat/completions、embeddings、streaming
• 处理 rate limit、超时、幂等与错误分类
• 不在仓库提交 API key；对齐团队密钥管理

通用客户端骨架

class LlmClient:
    def __init__(self, base_url, api_key_env="OPENAI_API_KEY"):
        self.key = os.environ[api_key_env]  # 勿硬编码
        self.base_url = base_url

    def chat(self, messages, model, temperature=0.2, tools=None, stream=False):
        payload = {"model": model, "messages": messages, "temperature": temperature}
        if tools:
            payload["tools"] = tools
        return self._post("/v1/chat/completions", payload, stream=stream)

    def embed(self, texts: list[str], model: str):
        return self._post("/v1/embeddings", {"model": model, "input": texts})

密钥走环境变量或密钥管理服务；PR 与日志 脱敏——ai-engineering-essentials 红线。

厂商差异速览

厂商	典型用途	注意
OpenAI	GPT 系列、embedding、function calling	org 级 quota；response_format JSON
Gemini	长上下文、多模态	safety settings；JSON mode
Hugging Face Inference	开源模型托管、embed	冷启动；模型 id 与 task 匹配

抽象层建议：messages + tools + stream 统一接口，厂商适配在 thin adapter。

Streaming

for chunk in client.chat(messages, model="gpt-4o", stream=True):
    delta = chunk.choices[0].delta.content
    if delta:
        yield delta  # SSE 推前端；Agent 可边读边展示

流式改善 首 token 延迟；仍需在服务端聚合完整输出做日志与评测。中断连接时要 cancel 上游（若 API 支持）。

重试与限流

RETRYABLE = {429, 500, 502, 503}

def with_retry(fn, max_attempts=3):
    for i in range(max_attempts):
        try:
            return fn()
        except ApiError as e:
            if e.status not in RETRYABLE or i == max_attempts - 1:
                raise
            time.sleep(2 ** i + random.random())