code/03-llm-apps/errors.py

"""几种常见错误长什么样，以及一个带重试、限流、记账的调用函数。"""
import json
import os
import random
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import openai
from openai import OpenAI

sys.path.insert(0, str(Path(__file__).parent.parent / "01-llm-basics"))
from cost import cost_usd  # 01 模块第 4 课写的计费函数

BASE_URL = os.environ.get("LLM_BASE_URL", "https://api.deepseek.com")
MODEL = os.environ.get("LLM_MODEL", "deepseek-flash")
NO_THINKING = {"thinking": {"type": "disabled"}}
HELLO = [{"role": "user", "content": "你好"}]

print("== 1. 密钥错误")
try:
    OpenAI(api_key="sk-wrong", base_url=BASE_URL).chat.completions.create(model=MODEL, messages=HELLO)
except openai.AuthenticationError as e:
    print(f"{type(e).__name__}，状态码 {e.status_code}")

print("== 2. 模型名写错")
try:
    OpenAI(api_key=os.environ["LLM_API_KEY"], base_url=BASE_URL).chat.completions.create(model="deepseek-flsh", messages=HELLO)
except openai.APIStatusError as e:
    print(f"{type(e).__name__}，状态码 {e.status_code}，{e.message[:100]}")

print("== 3. 超时（故意把超时设成 0.5 秒，并关掉 SDK 自带的重试）")
start = time.time()
try:
    OpenAI(api_key=os.environ["LLM_API_KEY"], base_url=BASE_URL, timeout=0.5, max_retries=0).chat.completions.create(
        model=MODEL, messages=[{"role": "user", "content": "写一篇 800 字的文章"}], extra_body=NO_THINKING)
except openai.APITimeoutError as e:
    print(f"{type(e).__name__}，用了 {time.time() - start:.1f} 秒")

# ---------- 一个能放进项目里用的调用函数 ----------

client = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=BASE_URL,
    timeout=60,  # 单次请求最多等 60 秒
    max_retries=0,  # 关掉 SDK 自带的重试，由下面的函数统一处理，方便记录
)
RETRYABLE = (openai.RateLimitError, openai.APITimeoutError, openai.APIConnectionError, openai.InternalServerError)
limiter = threading.Semaphore(5)  # 同一时刻最多 5 个请求在路上
log_lock = threading.Lock()
LOG = Path("calls.jsonl")


def call_llm(messages, max_attempts=4, **kwargs):
    for attempt in range(1, max_attempts + 1):
        start = time.time()
        try:
            with limiter:
                response = client.chat.completions.create(model=MODEL, messages=messages, **kwargs)
        except RETRYABLE as e:
            if attempt == max_attempts:
                raise
            # 指数退避：1 秒、2 秒、4 秒……再加一点随机，避免大家同时重试
            wait = 2 ** (attempt - 1) + random.random()
            print(f"  第 {attempt} 次失败（{type(e).__name__}），{wait:.1f} 秒后重试")
            time.sleep(wait)
            continue
        record = {
            "time": time.strftime("%Y-%m-%d %H:%M:%S"),
            "model": response.model,
            "seconds": round(time.time() - start, 2),
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
            "cost_usd": round(cost_usd(response.usage, MODEL), 6),
            "attempts": attempt,
        }
        with log_lock:
            with LOG.open("a") as f:
                f.write(json.dumps(record, ensure_ascii=False) + "\n")
        return response


print("== 4. 并发 20 个请求，最多同时 5 个，每次调用记账")
LOG.unlink(missing_ok=True)
questions = [f"用一句话解释 HTTP 状态码 {code} 的含义。" for code in [200, 201, 204, 301, 302, 304, 400, 401, 403, 404,
                                                                  405, 408, 409, 418, 429, 500, 502, 503, 504, 505]]
start = time.time()
with ThreadPoolExecutor(20) as pool:
    answers = list(pool.map(lambda q: call_llm([{"role": "user", "content": q}], extra_body=NO_THINKING), questions))
print(f"20 个请求用了 {time.time() - start:.1f} 秒")
print("第一条回答：", answers[0].choices[0].message.content)
records = [json.loads(line) for line in LOG.read_text().splitlines()]
print(f"日志共 {len(records)} 条，总花费 {sum(r['cost_usd'] for r in records):.6f} 美元，第一条：{records[0]}")