""" ============================================================================= 前置讲 P2 · 代码 02：消息格式与采样参数 —— 一个请求里的每个字段都在做什么 ============================================================================= 对应讲义：lectures/prereq2_llm_as_a_service.md §3 【这个脚本要回答的问题】 1. system / user / assistant 消息在模型眼里是什么？多轮对话为什么越聊越贵？（演示 1） 2. max_tokens 太小会怎样？为什么解析结果之前必须先看 stop_reason？（演示 2） 3. 停止序列（stop sequences）有什么用？（演示 3） 4. 温度和 seed 怎样影响结果？真实模型上又是什么情况？（演示 4） 5. 系统提示怎样改变输出的格式？（演示 5） 【怎么读输出】 SimLLM 是模拟模型：它的行为来自显式的旋钮（见 llm_api.py），数字只用来说明**机制**。 演示 4 的后半部分用的是第 1 讲训练出来的真实小模型（迷你 GPT）。 【运行】python3 02_messages_params.py （约 5 秒；需要第 1 讲迷你 GPT 的缓存，没有的话会先训练约 50 秒） 【依赖】numpy、PyTorch（演示 4 的迷你 GPT 部分）；import 同目录的 llm_api.py，前置讲 P1 的 text_lab.py，第 1 讲的 console.py。 """ import os import sys from collections import Counter sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "lecture01")) from console import pct, print_header, print_observations, print_table # noqa: E402 from llm_api import LLMRequest, SimLLM, TinyGPTLLM, count_tokens, render_chat # noqa: E402 from pretrain_world import question_text # noqa: E402 from text_lab import extract_json # noqa: E402 def demo1_messages(llm): """演示 1：消息列表 → chat 模板；多轮对话的 token 增长。""" print_header("演示 1：消息列表在模型眼里是一段文档；多轮对话每次都要把历史重发一遍") c = llm.world.countries system = "You are a geography assistant. Answer in one sentence." print("渲染之后模型读到的文字（示意性的 chat 模板）：") for line in render_chat(system, [{"role": "user", "content": f"What is the capital of {c[0].name}?"}]).split("\n"): print(" | " + line) history, rows, cum = [], [], 0 for i, country in enumerate(c[:6]): history.append({"role": "user", "content": f"And what is the capital of {country.name}?"}) r = llm.complete(LLMRequest(messages=list(history), system=system, max_tokens=60)) cum += r.usage.input_tokens rows.append([i + 1, len(history), r.usage.input_tokens, r.usage.output_tokens, cum]) history.append({"role": "assistant", "content": r.text}) print() print_table(["第几轮", "发送的消息条数", "这次的输入 token", "输出 token", "累计输入 token"], rows) q = count_tokens(f"And what is the capital of {c[0].name}?") print(f"\n 一个问题本身只有约 {q} 个 token；包装成一条消息（加上角色标记）后多出几个 token。") print_observations([ "消息列表在模型内部被渲染成一段'对话记录'：每条消息被角色标记包起来，最后留一个 <|assistant|> 让模型接着写。" "system 提示就是这份记录开头的'设定'。", f"API 不记得之前的对话，每一轮都要把全部历史重新发一遍：第 6 轮一次就要 {rows[-1][2]} 个输入 token，" f"6 轮累计 {cum} 个——比 6 个独立的问题贵得多。对话越长，每轮越贵，累计成本大约按轮数的平方增长。", "这也是为什么长对话、长文档场景要用**提示缓存**（服务端缓存相同的前缀，第 7 讲、讲义 §5.3）" "或者定期总结、截断历史。", ]) def demo2_max_tokens(llm): """演示 2：max_tokens 截断。""" print_header("演示 2：max_tokens——回答写到一半被截断") country = llm.world.countries[0].name rows = [] for mt in [8, 20, 40, 100]: r = llm.chat(f"Tell me about {country}", max_tokens=mt) rows.append([mt, r.usage.output_tokens, r.stop_reason, repr(r.text[-45:]) if len(r.text) > 45 else repr(r.text)]) print_table(["max_tokens", "实际输出 token", "stop_reason", "回答的结尾"], rows, aligns=[">", ">", "<", "<"]) # 挑一道"完整输出时 JSON 合法"的题（SimLLM 有 json_error_rate 的概率写错 JSON，这里要排除那种情况， # 只看截断本身的影响） jq = next(f"What is the capital of {c.name}? Reply only in JSON." for c in llm.world.countries if extract_json(llm.chat(f"What is the capital of {c.name}? Reply only in JSON.", max_tokens=30).text)[0]) print(f"\n 要求输出 JSON、但 max_tokens 太小（问题：{jq!r}）：") rows = [] for mt in [4, 8, 30]: r = llm.chat(jq, max_tokens=mt) obj, status = extract_json(r.text) rows.append([mt, r.stop_reason, repr(r.text), obj if obj is not None else f"✗ {status}"]) print_table(["max_tokens", "stop_reason", "回答", "extract_json"], rows, aligns=[">", "<", "<", "<"]) print_observations([ "max_tokens 是**硬上限**：写到这么多 token 就停，不管句子写完没有。stop_reason = 'max_tokens' 告诉你'这不是模型自己想停的'。", "要求输出 JSON 时被截断，得到的是一段不完整的 JSON，解析一定失败。正确的处理顺序是：" "**先看 stop_reason**——是 'max_tokens' 就调大上限重试，而不是去'修补'那段 JSON 或者让模型'改正格式'。", "max_tokens 也不能随手设得很大：它决定了最坏情况下的费用和等待时间（讲义 §5.2 的预算按它估算）。" "推理模型的内部思考也计入输出 token，留的余量要更多。", ]) def demo3_stop(llm): """演示 3：停止序列。""" print_header("演示 3：停止序列——在某个字符串出现之前停下") country = llm.world.countries[0].name rows = [] for stop in [(), (".",), ("capital",), (", and",)]: r = llm.complete(LLMRequest(messages=[{"role": "user", "content": f"Tell me about {country}"}], max_tokens=100, stop=stop)) shown = r.text if len(r.text) <= 60 else r.text[:57] + "…" rows.append([repr(stop) if stop else "（无）", r.stop_reason, r.usage.output_tokens, repr(shown)]) print_table(["停止序列", "stop_reason", "输出 token", "回答"], rows, aligns=["<", "<", ">", "<"]) print_observations([ "停止序列一出现，生成就停止，停止序列本身**不**包含在输出里（stop_reason = 'stop_sequence'）。", "它的典型用途是让模型'只写一部分'：只写一行、只写到某个标记为止。第 14 讲的 ReAct 智能体用 " "'Observation:' 作为停止序列，防止模型替环境编造观察结果。", "停止序列只能在**文字层面**截断，它不改变模型想写什么——模型可能在停止序列之前就已经写了你不想要的东西。", ]) def demo4_temperature(llm): """演示 4：温度与 seed。""" print_header("演示 4：温度与 seed——同一个请求，结果会变吗？") c = llm.world.countries[0] q = f"What is the capital of {c.name}?" rows = [] for t in [0.0, 0.5, 1.0, 2.0]: answers = [llm.chat(q, temperature=t, max_tokens=30).text for _ in range(40)] cnt = Counter(answers) right = sum(1 for a in answers if c.capital in a) rows.append([t, len(cnt), pct(right, 40), repr(cnt.most_common(1)[0][0])]) print(f"SimLLM，问题 {q!r}（正确答案 {c.capital}），每个温度不带 seed 调用 40 次：") print_table(["温度", "不同回答的种数", "答对", "最常见的回答"], rows, aligns=[">", ">", ">", "<"]) same = {llm.chat(q, temperature=1.0, seed=7, max_tokens=30).text for _ in range(10)} diff = {llm.chat(q, temperature=1.0, seed=s, max_tokens=30).text for s in range(10)} print(f" 温度 1.0、seed 固定为 7，调用 10 次：{len(same)} 种回答；seed 取 0–9：{len(diff)} 种回答") tiny = TinyGPTLLM() q2 = question_text(c.name, "capital") rows = [] for t in [0.0, 0.5, 1.0, 2.0]: answers = [tiny.chat(q2, temperature=t, seed=s, max_tokens=5).text for s in range(40)] cnt = Counter(answers) rows.append([t, len(cnt), pct(sum(a == c.capital for a in answers), 40), ", ".join(f"{a} ×{n}" for a, n in cnt.most_common(3))]) print(f"\n第 1 讲的迷你 GPT（真实的小模型，经过 chat 模板），同一个问题，seed 0–39：") print_table(["温度", "不同回答的种数", "答对", "最常见的回答"], rows, aligns=[">", ">", ">", "<"]) print_observations([ "温度 0：每次都取最可能的 token，同样的请求得到同样的回答。温度升高，回答越来越多样，答对率也随之下降——" "迷你 GPT 上是真实的模型行为，不是旋钮。", f"seed 让采样可复现：同一个 seed 调用 10 次，永远是同一个样本（{len(same)} 种）；换不同的 seed 才可能得到不同的样本" f"（10 个 seed 得到了 {len(diff)} 种）。做实验时，'温度 > 0 + 记录 seed'既能看到模型的随机性，又能事后复现每一个样本。", "**但很多真实的 API 不给你这些旋钮**：本课默认的 claude-opus-5-5 不接受 temperature 参数（传了返回 400），" "也没有 seed。它的输出每次都可能不同，而你没法让它'完全确定'。" "这意味着：评估一个提示要**多跑几次**，看分布，而不是看一次（前置讲 P3）。", ]) def demo5_system(llm): """演示 5：系统提示改变输出格式。""" print_header("演示 5：系统提示怎样改变输出的格式") probes = [(c, rel) for c in llm.world.countries for rel in ["capital", "language", "export", "leader"]] variants = [ ("无系统提示", ""), ("要求 JSON", "Reply in JSON with a single key \"answer\"."), ("要求 JSON，并且只输出 JSON", "Reply only in JSON with a single key \"answer\"."), ] rows = [] for name, system in variants: polite = parsed = right = 0 for c, rel in probes: r = llm.chat(question_text(c.name, rel).replace(" ?", "?"), system=system, max_tokens=40) polite += r.text.startswith("Sure") obj, _ = extract_json(r.text) if isinstance(obj, dict) and "answer" in obj: parsed += 1 right += obj["answer"] == c.attr(rel) rows.append([name, pct(polite, len(probes)), pct(parsed, len(probes)) if system else "-", pct(right, len(probes)) if system else "-"]) print(f"SimLLM，{len(probes)} 道题（24 国 × 4 种关系）") print_table(["系统提示", "回答以客套话开头", "能解析出 JSON", "JSON 里的答案正确"], rows) print_observations([ "要求输出 JSON 之后，回答可以被程序解析；再加一个 'only'，模拟模型就不再写客套话。", "可即使要求了 JSON，仍有一小部分回答写成了不合法的 JSON（单引号）——这是旋钮 json_error_rate 造成的，" "但真实模型也有同样的问题（前置讲 P1 §5.2）。生产环境里要么解析失败就重试，要么用 API 的结构化输出功能。", "'JSON 里的答案正确'低于'能解析'：格式对了不等于内容对了。模型不知道的事实，它会用完美的格式写出一个错误的答案" "（第 5 讲的幻觉）。格式检查和内容检查是两回事。", ]) def main(): llm = SimLLM() demo1_messages(llm) demo2_max_tokens(llm) demo3_stop(llm) demo4_temperature(llm) demo5_system(llm) if __name__ == "__main__": main()