模型 API：消息、流式输出、异常处理与 Token 用量

python -m pip install httpx

export MODEL_API_BASE="https://provider.example/v1"
export MODEL_API_KEY="replace-with-your-key"
export MODEL_NAME="provider-model-name"

import os
import time
import httpx


BASE_URL = os.environ["MODEL_API_BASE"].rstrip("/")
API_KEY = os.environ["MODEL_API_KEY"]
MODEL = os.environ["MODEL_NAME"]


def chat(messages: list[dict], retries: int = 2) -> dict:
    payload = {
        "model": MODEL,
        "messages": messages,
        "temperature": 0.2,
        "max_tokens": 200,
        "stream": False,
    }

    for attempt in range(retries + 1):
        try:
            started = time.perf_counter()
            response = httpx.post(
                f"{BASE_URL}/chat/completions",
                headers={"Authorization": f"Bearer {API_KEY}"},
                json=payload,
                timeout=httpx.Timeout(30.0, connect=5.0),
            )
            response.raise_for_status()
            data = response.json()
            return {
                "text": data["choices"][0]["message"]["content"],
                "usage": data.get("usage", {}),
                "request_id": response.headers.get("x-request-id"),
                "elapsed_seconds": round(time.perf_counter() - started, 3),
            }
        except (httpx.TimeoutException, httpx.NetworkError):
            if attempt == retries:
                raise
            time.sleep(2 ** attempt)
        except httpx.HTTPStatusError as error:
            if error.response.status_code not in {429, 500, 502, 503, 504} or attempt == retries:
                raise
            time.sleep(2 ** attempt)


result = chat([
    {"role": "system", "content": "只整理输入中的事实，不补充处理建议。"},
    {"role": "user", "content": "节点 NX-8 网络超时，切换线路后恢复。"},
])
print(result)

import json


def stream_chat(messages: list[dict]):
    payload = {
        "model": MODEL,
        "messages": messages,
        "temperature": 0.2,
        "max_tokens": 200,
        "stream": True,
    }

    with httpx.stream(
        "POST",
        f"{BASE_URL}/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json=payload,
        timeout=httpx.Timeout(60.0, connect=5.0),
    ) as response:
        response.raise_for_status()
        for line in response.iter_lines():
            if not line.startswith("data:"):
                continue
            data = line.removeprefix("data:").strip()
            if data == "[DONE]":
                break
            event = json.loads(data)
            delta = event["choices"][0].get("delta", {})
            content = delta.get("content")
            if content:
                yield content


for chunk in stream_chat([{"role": "user", "content": "用一句话总结网络状态。"}]):
    print(chunk, end="", flush=True)