""" ============================================================================= 前置讲 P2 · 代码 01：一次 LLM 调用在网络上长什么样 —— HTTP、状态码、API key ============================================================================= 对应讲义：lectures/prereq2_llm_as_a_service.md §1–§2 【这个脚本要回答的问题】 1. 调用一次 LLM API，网络上实际传过去、传回来的是什么字节？（演示 1） 2. 出错时服务器怎样告诉你出了什么错？哪些错误值得重试、哪些不值得？（演示 2） 3. API key 应该放在哪里？怎样避免它出现在日志和报错信息里？（演示 3） 【怎么读输出】 演示 1 原样打印请求和响应的文字（HTTP 本来就是有格式的文字）。演示 2 是一张"情形 → 状态码 → 错误类型"的表。 【运行】python3 01_http_anatomy.py （1 秒左右；会在本机 127.0.0.1 起一个模拟服务器，不联网） 【依赖】只用 Python 标准库；import 同目录的 mock_server.py、llm_api.py，以及第 1 讲的 console.py。 """ import json import os import re import sys import tempfile sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "lecture01")) from console import print_header, print_observations, print_table # noqa: E402 from llm_api import LLMError, LLMRequest, LoggedLLM, read_log # noqa: E402 from mock_server import HttpLLM, MockLLMServer, raw_exchange, redact # noqa: E402 API_KEY = "sk-test-7f3a9c2e" # 只在本机的模拟服务器上有效的"假"密钥 def mask(text, srv): """把每次运行都会变的部分（随机端口号、Date 头里的当前时间）换成占位符，让输出每次都一样。""" port = srv.url.rsplit(":", 1)[1] text = text.replace(f":{port}", ":<端口>") return re.sub(r"(?im)^(date: ).*$", r"\1<当前时间>", text) def demo1_raw(srv, country): """演示 1：原始的请求和响应字节。""" print_header("演示 1：一次调用在网络上长什么样——HTTP 请求和响应的原始文字") body = {"model": "sim-small", "max_tokens": 50, "system": "Answer briefly.", "messages": [{"role": "user", "content": f"What is the capital of {country}?"}]} request, response = raw_exchange(srv.url, body, {"content-type": "application/json", "x-api-key": API_KEY}) shown = mask(request.decode("utf-8").replace(API_KEY, redact(API_KEY)), srv) print("—— 客户端发出的请求（API key 已打码）——") for line in shown.split("\r\n"): print(" > " + line) print("—— 服务器返回的响应 ——") head, _, payload = mask(response.decode("utf-8"), srv).partition("\r\n\r\n") for line in head.split("\r\n"): print(" < " + line) print(" <") print(" < " + json.dumps(json.loads(payload), ensure_ascii=False, indent=2).replace("\n", "\n < ")) print_observations([ "请求 = 一行请求行（POST /v1/messages HTTP/1.1）+ 若干行头部（content-type、x-api-key……）+ 一个空行 + JSON 请求体。" "官方 SDK、requests、urllib 在底下做的就是拼出这样一段文字。", "响应 = 状态行（HTTP/1.1 200 OK）+ 头部 + 空行 + JSON。回答在 content 里（一个'块'的列表，" "这里只有一个 text 块），stop_reason 说明为什么停下，usage 是计费用的 token 数。", "API 是**无状态**的：服务器不记得你上一次问了什么。多轮对话要把整段历史每次都发一遍（02 号脚本演示 1）。", ]) def demo2_errors(srv, country): """演示 2：各种错误。""" print_header("演示 2：出错时服务器怎样告诉你——状态码与错误类型") good = [{"role": "user", "content": f"What is the capital of {country}?"}] cases = [ ("一切正常", API_KEY, LLMRequest(messages=good), None), ("没带 API key", "", LLMRequest(messages=good), None), ("API key 写错了", "sk-test-wrong", LLMRequest(messages=good), None), ("模型名写错了", API_KEY, LLMRequest(messages=good, model="sim-tiny"), None), ("消息以 assistant 开头", API_KEY, LLMRequest(messages=[{"role": "assistant", "content": "hi"}]), None), ("服务器限速（计划好的 429）", API_KEY, LLMRequest(messages=good), 429), ("服务器内部错误（计划好的 500）", API_KEY, LLMRequest(messages=good), 500), ] rows = [] for name, key, req, planned in cases: if planned: srv.fail_plan.append(planned) llm = HttpLLM(srv.url, api_key=key, name=req.model) try: r = llm.complete(req) rows.append([name, "200", "-", "-", repr(r.text)]) except LLMError as e: ra = getattr(e, "retry_after", None) rows.append([name, str(e.status or "?"), type(e).__name__, "是" if e.retryable else "否", f"{e}" + (f"（retry-after: {ra:g} 秒）" if ra else "")]) print_table(["情形", "状态码", "错误类型", "值得重试？", "说明"], rows, aligns=["<", ">", "<", "<", "<"]) # 一个不是合法 JSON 的请求体（直接用 socket 发，绕过客户端的 JSON 编码） _, resp = raw_exchange(srv.url, None, {"content-type": "application/json", "x-api-key": API_KEY}) status = resp.split(b"\r\n", 1)[0].decode() print(f"\n 请求体是 'null'（合法的 JSON，但不是一个对象）→ {status}") print_observations([ "**4xx 是你的错**：没带 key / key 不对（401）、模型名不存在（404）、请求格式不对（400；" "请求体是合法的 JSON 但不是对象也算）。" "原样重试一万次也还是错，要改请求或配置。", "**429 和 5xx 是服务端的暂时状况**：限速、过载、内部错误。等一等再试通常就好了；" "429 往往还附带 retry-after 头，告诉你至少等多久（04 号脚本专门讲重试）。", "错误体也是 JSON（type + message）。把它解析出来、翻译成清楚的错误类型，" "外层的重试、记账代码就不需要关心底下是哪一家的 API。", ]) def demo3_secrets(srv, country): """演示 3：API key 放在哪里。""" print_header("演示 3：API key 放在哪里——以及怎样不把它写进日志") os.environ["MOCK_LLM_API_KEY"] = API_KEY # 真实场景里是你在终端里 export 的，不在代码里 llm = HttpLLM(srv.url) # 不传 api_key：从环境变量读 with tempfile.TemporaryDirectory() as d: path = os.path.join(d, "calls.jsonl") logged = LoggedLLM(llm, path, run_id="demo") logged.chat(f"What is the capital of {country}?") with open(path, encoding="utf-8") as f: text = f.read() rec = read_log(path)[0] naive = mask(f"POST {srv.url}/v1/messages headers={{'x-api-key': '{API_KEY}'}}", srv) rows = [ ["HttpLLM 从哪里拿到 key", "环境变量 MOCK_LLM_API_KEY（代码里没有）"], ["日志里有没有 key", "有！" if API_KEY in text else "没有（LoggedLLM 只记录请求体 payload，不记录头部）"], ["日志记录的字段", ", ".join(rec.keys())], ["天真地打印请求头", naive.replace(API_KEY, "<<这里是你的 key>>")], ["打码后再打印", naive.replace(API_KEY, redact(API_KEY))], ] print_table(["检查项", "结果"], rows, aligns=["<", "<"]) print_observations([ "API key 就是你的钱包：谁拿到它，谁就能用你的额度。**永远不要把它写进代码**——代码会被提交到 git、" "被复制给同事、被贴进聊天里。放在环境变量里（或者专门的密钥管理工具），代码只负责去读。", "调试时最容易泄露 key 的地方是**日志和报错信息**：把整个请求（含头部）打印出来，key 就留在了终端历史和日志文件里。" "记录请求时只记请求体，必须显示时先打码。", "如果 key 已经泄露（提交进了公开的仓库），删掉那一行代码是不够的——git 历史里还在。" "要去服务商的控制台**作废**这个 key，再换一个新的。", ]) del os.environ["MOCK_LLM_API_KEY"] def main(): with MockLLMServer(api_key=API_KEY) as srv: country = srv.llm.world.countries[0].name demo1_raw(srv, country) demo2_errors(srv, country) demo3_secrets(srv, country) if __name__ == "__main__": main()