"""几种省钱、提速的办法，各做一个小实验，量一量效果。
在 AI-Course/code/06-production 目录下运行：python cost_latency.py
一共约 40 次调用，按 2026 年 9 月的价格花费约 0.05 美元。
"""
import hashlib
import os
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from openai import OpenAI
client = OpenAI(api_key=os.environ["LLM_API_KEY"], base_url=os.environ.get("LLM_BASE_URL", "https://api.deepseek.com"))
MODEL = "deepseek-flash"
PRICES = (0.006, 0.30, 1.20) # 缓存命中输入、未命中输入、输出，美元 / 百万词元，高峰价，截至 2026 年 9 月
NO_THINKING = {"thinking": {"type": "disabled"}}
DOCS = "\n\n".join(p.read_text() for p in sorted(Path(__file__, "../../../data/httpx-docs").resolve().glob("*.md")))
def cost(u):
hit = getattr(u, "prompt_cache_hit_tokens", 0) or 0
return (hit * PRICES[0] + (u.prompt_tokens - hit) * PRICES[1] + u.completion_tokens * PRICES[2]) / 1e6
def ask(messages, **kw):
start = time.time()
r = client.chat.completions.create(model=MODEL, messages=messages, extra_body=kw.pop("extra_body", NO_THINKING), **kw)
return r.choices[0].message.content, r.usage, time.time() - start
print("== 1. 缓存：固定的资料放在前面，还是问题放在前面")
questions = ["httpx 怎么设置超时？", "httpx 怎么开启 HTTP/2？", "httpx 怎么上传文件？"]
for label, build in [
("问题在前", lambda q: [{"role": "user", "content": f"问题：{q}\n\n参考文档：\n{DOCS}\n\n一句话回答。"}]),
("文档在前", lambda q: [{"role": "system", "content": f"参考文档：\n{DOCS}"}, {"role": "user", "content": q + "一句话回答。"}]),
]:
total, hits = 0.0, []
for q in questions:
_, u, _ = ask(build(q))
total += cost(u)
hits.append(f"{u.prompt_cache_hit_tokens}/{u.prompt_tokens}")
print(f" {label}：三次的缓存命中 {hits}，共 {total:.5f} 美元")
print("\n== 2. 输出长度：不做要求 vs 要求简短")
q = "httpx 和 requests 有什么区别？"
for label, prompt in [("不做要求", q), ("要求简短", q + "用三句话以内回答。")]:
text, u, sec = ask([{"role": "user", "content": prompt}])
print(f" {label}：输出 {u.completion_tokens} 词元，{sec:.1f} 秒，{cost(u):.5f} 美元")
print("\n== 3. 简单问题开不开思考")
q = [{"role": "user", "content": "httpx 的 get 方法怎么传查询参数？一句话回答。"}]
for label, extra in [("不思考", NO_THINKING), ("思考", {"thinking": {"type": "enabled"}})]:
text, u, sec = ask(q, extra_body=extra)
print(f" {label}：输出 {u.completion_tokens} 词元，{sec:.1f} 秒，{cost(u):.5f} 美元 | {text[:40]!r}")
print("\n== 4. 自己做结果缓存：同样的问题直接返回上次的答案")
cache = {}
def cached_ask(question):
key = hashlib.sha256(" ".join(question.lower().split()).encode()).hexdigest() # 忽略大小写和多余空格
if key in cache:
return cache[key], 0.0
text, u, _ = ask([{"role": "user", "content": question}])
cache[key] = text
return text, cost(u)
spent = sum(cached_ask(q)[1] for q in ["httpx 怎么设置代理？", "httpx 怎么设置代理？", "HTTPX 怎么设置代理？", "httpx 怎么设置代理"])
print(f" 问了 4 次（写法略有不同），实际调用 {len(cache)} 次模型，共 {spent:.5f} 美元")
print("\n== 5. 串行 vs 并发")
batch = [f"用一句话解释 HTTP 状态码 {c}。" for c in [200, 301, 302, 400, 401, 403, 404, 429, 500, 503]]
start = time.time()
for b in batch:
ask([{"role": "user", "content": b}])
serial = time.time() - start
start = time.time()
with ThreadPoolExecutor(5) as pool:
list(pool.map(lambda b: ask([{"role": "user", "content": b}]), batch))
print(f" 10 个请求：一个一个来 {serial:.1f} 秒，5 个并发 {time.time() - start:.1f} 秒")