"""几种常见错误长什么样，以及一个带重试、限流、记账的调用函数。"""
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]}")