第 1 章讲义 · 01 LLM API 与模型调用基础

配套代码：本仓库的 provider 层（src/anna/providers/）与 LLM 统一模型调用服务（src/anna/server/、main.py）
学完本节，你应该能回答：一次 LLM 调用请求长什么样？参数怎么影响输出？ token 怎么算？怎么让模型稳定返回 JSON？出错怎么办？多个模型怎么统一调？

1. Chat Completions / Responses API

1.1 请求结构：以 OpenAI Chat Completions 为标准

主流模型 API 基本都是"HTTP POST + JSON 请求体"，OpenAI 的 POST /v1/chat/completions 是事实标准。先看完整请求：

curl -s https://api.openai.com/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -d '{
    "model": "gpt-4o-mini",
    "messages": [
      {"role": "system", "content": "你是英语陪练 Anna。"},
      {"role": "user", "content": "什么是 token？"}
    ],
    "temperature": 0.7,
    "max_tokens": 200,
    "stream": false
  }'

请求体字段（重要程度递减）：

字段	说明
`model`	模型名，决定用哪个模型、什么价格
`messages`	对话历史，`role` + `content` 的有序列表
`temperature`	采样随机性（0~2）
`top_p`	核采样阈值（0~1）
`max_tokens`	输出 token 上限
`stream`	是否流式返回
`response_format`	结构化输出声明（后文第 4 节）

消息格式：核心是 messages 数组，四种角色：

role	用途	例子
`system`	设定人格/规则，整段对话生效	“你是英语陪练 Anna”
`user`	用户输入	“什么是 token？”
`assistant`	模型历史回复（多轮对话必须回传）	“Token 是……”
`tool`	工具调用结果（配合 function calling）	查词工具返回的释义

返回结构：

{
  "id": "chatcmpl-xxxx",
  "object": "chat.completion",
  "model": "gpt-4o-mini",
  "choices": [
    {
      "index": 0,
      "message": {"role": "assistant", "content": "Token 是模型处理文本的基本单位……"},
      "finish_reason": "stop"
    }
  ],
  "usage": {"prompt_tokens": 18, "completion_tokens": 42, "total_tokens": 60}
}

三个关键点：内容在 choices[0].message.content；finish_reason 表示结束原因（stop 正常结束 / length 达到 max_tokens 被截断 / content_filter 被过滤）； usage 返回 token 用量（第 3 节细讲）。

1.2 Responses API（新接口，先认识）

OpenAI 后来又推出 POST /v1/responses（responses.create），能力与 Chat Completions 重叠但结构不同：消息平铺在 input 字段、系统提示用 instructions、工具调用更统一。对初学者：先吃透 Chat Completions 结构，Responses 只是同一套概念的另一种表达。本仓库的网关对外暴露的就是 Chat Completions 兼容接口。

官方调用例子（curl）：

curl -s https://api.openai.com/v1/responses \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -d '{
    "model": "gpt-4o-mini",
    "instructions": "你是英语陪练 Anna。",
    "input": "什么是 token？",
    "max_output_tokens": 200
  }'

返回结构（简化）：

{
  "id": "resp_xxxx",
  "object": "response",
  "model": "gpt-4o-mini",
  "output": [
    {
      "type": "message",
      "role": "assistant",
      "content": [{"type": "output_text", "text": "Token 是模型处理文本的基本单位……"}]
    }
  ],
  "usage": {"input_tokens": 15, "output_tokens": 42, "total_tokens": 57}
}

Python（openai SDK）：

from openai import OpenAI
client = OpenAI()
resp = client.responses.create(
    model="gpt-4o-mini",
    instructions="你是英语陪练 Anna。",
    input="什么是 token？",
)
print(resp.output_text)  # 便捷属性：拼好的文本

Responses ↔ Chat Completions 字段对照：

概念	Responses API	Chat Completions
系统提示	`instructions`（顶层）	`messages` 里 `role: "system"`
对话内容	`input`（字符串或消息数组）	`messages`
输出上限	`max_output_tokens`	`max_tokens`
返回文本	`output[].content[].text`	`choices[0].message.content`
用量字段	`input_tokens` / `output_tokens`	`prompt_tokens` / `completion_tokens`

注意：本仓库网关目前没有 /v1/responses 端点（见附录对照表，标为未实现）。上面的例子调用的是 OpenAI 官方 API；后续若做"兼容 shim"（把 /v1/responses 转成内部 Chat Completions 再走现有 provider），这段请求就能打到自己的网关，属于不错的练习方向。

1.3 主流厂商协议对照

厂商	端点	差异点	本项目 adapter
OpenAI 系（含 DeepSeek/DashScope 兼容模式）	`POST /v1/chat/completions`	事实标准	`openai_compatible.py`
Anthropic	`POST /v1/messages`	system 独立字段、`max_tokens` 必填、content 是块数组	`anthropic.py`
Google Gemini	`POST /v1beta/models/{model}:generateContent`	角色叫 `user/model`、system 叫 `systemInstruction`	`gemini.py`

所以"统一模型调用"的本质是：定义一个统一接口，让各家 adapter 负责格式转换 （本仓库的 ChatProvider 就是这个统一接口）。

Anthropic 调用例子：

curl -s https://api.anthropic.com/v1/messages \
  -H "Content-Type: application/json" \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -d '{
    "model": "claude-sonnet-4-5",
    "max_tokens": 200,
    "system": "你是英语陪练 Anna。",
    "messages": [{"role": "user", "content": "什么是 token？"}]
  }'
# 返回：{"content":[{"type":"text","text":"..."}],"usage":{"input_tokens":15,"output_tokens":42}}

Gemini 调用例子：

curl -s "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent" \
  -H "Content-Type: application/json" \
  -H "x-goog-api-key: $GOOGLE_API_KEY" \
  -d '{
    "systemInstruction": {"parts": [{"text": "你是英语陪练 Anna。"}]},
    "contents": [{"role": "user", "parts": [{"text": "什么是 token？"}]}],
    "generationConfig": {"maxOutputTokens": 200}
  }'
# 返回：{"candidates":[{"content":{"parts":[{"text":"..."}]}}],"usageMetadata":{"promptTokenCount":15,"candidatesTokenCount":42}}

1.4 在本项目里调用

启动网关后，用 curl 走统一入口：

curl -s http://127.0.0.1:8000/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{"model":"auto","messages":[{"role":"user","content":"用一句话解释 token"}]}'

或者在 Python 里直接用 provider 层：

from anna.providers import get_provider_chain
provider = get_provider_chain()          # 读取 ANNA_* 环境变量
reply = provider.complete([
    {"role": "system", "content": "你是英语陪练 Anna。"},
    {"role": "user", "content": "什么是 token？"},
], max_tokens=200)
print(reply)

2. 模型参数：temperature、top_p、max_tokens

2.1 temperature：控制随机性

模型每次回答是从概率分布里"抽样"，temperature 决定抽样的激进程度：

temperature	行为	适用
0	几乎每次都选概率最高的词，稳定	结构化输出、查词、代码
0.7	平衡	日常对话
1.0+	更发散、更有创意	头脑风暴、文案

同一句话，temperature=0 时两次回答几乎一致；temperature=1 时两次可能不同：

问：给猫起个名字
temp=0   → "咪咪"
temp=1   → "毛球"  （第二次可能是 "年糕"）

在网关里传参：

curl -s http://127.0.0.1:8000/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{"model":"auto","temperature":0,"messages":[{"role":"user","content":"给猫起个名字"}]}'

Python 里对比两次调用：

from anna.providers import get_provider_chain
provider = get_provider_chain()
prompt = [{"role": "user", "content": "给猫起个名字"}]
print(provider.complete(prompt, temperature=0))   # 每次几乎一样
print(provider.complete(prompt, temperature=1))   # 每次可能不同

2.2 top_p：核采样

top_p 表示"只从累计概率达到 p 的最可能词里采样"。p=1 用全部词表；p=0.1 几乎只用最高概率的词，效果类似低 temperature。

OpenAI 的建议：temperature 和 top_p 改一个就好，不要同时调——两者都在控制随机性，同时调会互相干扰。

在原始 OpenAI API 里传参：

curl -s https://api.openai.com/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"写一句关于秋天的诗"}],"top_p":0.1}'

（本项目网关当前暴露 temperature / max_tokens；top_p 作为扩展字段可透传到 OpenAI 兼容厂商，属于课程后续练习。）

2.3 max_tokens：输出预算

控制单次回复的长度上限，也直接控制成本（按 token 计费）；
如果回答超长被截断，返回的 finish_reason 会是 length，而不是 stop；
设置过小（如 10）会看到"话没说完"：

{"choices": [{"message": {"content": "Token 是模型处理文本的基本单"}, "finish_reason": "length"}]}

在网关上用 max_tokens=10 复现截断（返回内容很短、可能话没说完）：

curl -s http://127.0.0.1:8000/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{"model":"auto","max_tokens":10,"messages":[{"role":"user","content":"用 200 字介绍 token"}]}'

排查技巧：看到 finish_reason: length，先加 max_tokens，再考虑换更大 context window 的模型。

说明：网关组装响应时目前固定返回 finish_reason: "stop"（openai_format.py），上游的 length / content_filter 结束原因暂未透传，属于可扩展项；上文的 length 示例描述的是各家原始 API 的行为。

3. Token 与 Context Window

3.1 Token 是什么

模型不按"字"理解文本，而是按 token（词元）：把文本切成一串基本单位。

英文：大约 4 个字符 ≈ 1 token（“token” ≈ 1 token，“streaming” ≈ 2~3 token）
中文：一般 1 个汉字 ≈ 1~2 token
数字/标点/空格也占 token

输入:  "Anna is a reading assistant。"
分词:  ["Anna", " is", " a", " reading", " assistant", "。"]
约 6 token

用 Python 估算（精确计数用 tiktoken，首次使用需联网下载编码表）：

def estimate_tokens(text: str) -> int:
    """粗略估算：英文按 4 字符/token，中文按 1 字/token。"""
    ascii_chars = sum(1 for ch in text if ord(ch) < 128)
    cjk_chars = len(text) - ascii_chars
    return ascii_chars // 4 + cjk_chars
print(estimate_tokens("Anna is a reading assistant。"))   # 例如 6~7
# 精确版（cl100k_base 与 GPT-4o 系列同族）：
# import tiktoken
# enc = tiktoken.get_encoding("cl100k_base")
# print(len(enc.encode("Anna is a reading assistant。")))

3.2 Context Window（上下文窗口）

context window = 模型一次能"看"的输入 + 输出总预算。例如 32K 的模型：

输入 12K token + 输出 8K token = 20K ≤ 32K  ✓
输入 30K token + 想输出 5K   = 35K > 32K  ✗（需要压缩输入或减小输出）

代码里检查预算：

def fits_context(input_tokens: int, output_tokens: int, window: int) -> bool:
    return input_tokens + output_tokens <= window
assert fits_context(12_000, 8_000, 32_000) is True
assert fits_context(30_000, 5_000, 32_000) is False

超限的常见表现：请求被拒（context_length 类错误）、或回答被截断。应对手段：

分块：长文按段落/标题切块，每次只喂相关块（注：Anna 的文章导入与分块属于 Phase 0 后续任务，当前仓库尚未实现，这里只讲思路）；
摘要：把旧对话压成 summary；
缓存：固定前缀（system prompt）命中 prompt cache 降本。

3.3 在本项目里看 token

每次调用后网关会记录 usage：

{"usage": {"prompt_tokens": 18, "completion_tokens": 42, "total_tokens": 60}}

页面"用量统计"面板或 GET /v1/usage/summary 都能看到累计 token 与成本估算（价格表在 src/anna/usage.py 的 PRICING）。

Python 里读单次用量并估算成本：

from anna.providers import get_provider_chain
from anna.usage import estimate_cost
provider = get_provider_chain()
provider.complete([{"role": "user", "content": "你好"}])
usage = provider.last_usage                 # 各家原始 usage 形状
print(provider.model, usage)
cost = estimate_cost(
    provider.model,
    usage.get("prompt_tokens", 0),
    usage.get("completion_tokens", 0),
)
print(f"cost ≈ ${cost:.6f}")

4. Structured Output

4.1 为什么需要

自由文本不可靠：解析模型回答做后续逻辑时，一个多出来的标点就可能让程序崩。 结构化输出 = 让模型"必须"返回可解析、可校验的数据。

三种常见方式（由弱到强）：

方式	说明	可靠性
Prompt 约定	“只输出 JSON”	低（模型可能加注释/文字）
JSON Schema	用 schema 约束字段与类型，模型原生遵循	高
Pydantic 校验	结果出来后用 Pydantic 模型解析校验	兜底

4.2 在本项目里的实现

网关的 /v1/chat/completions 支持 response_format：

curl -s http://127.0.0.1:8000/v1/chat/completions \
  -H 'Content-Type: application/json' \
  -d '{
    "model": "auto",
    "messages": [{"role": "user", "content": "查词：serendipity"}],
    "response_format": {"type": "json_object"}
  }'

三家厂商的落地方式（见 tests/test_structured_output.py）：

OpenAI 兼容：response_format 原样透传（模型原生支持）；
Gemini：映射成 generationConfig.responseMimeType: "application/json"；
Anthropic：用隐藏工具强制 JSON（模型没有 response_format，就用 tool 协议约束）。

更严格的 JSON Schema 版本：

-d '{
  "model": "auto",
  "messages": [{"role": "user", "content": "查词：serendipity"}],
  "response_format": {
    "type": "json_schema",
    "json_schema": {
      "type": "object",
      "properties": {
        "word": {"type": "string"},
        "meaning": {"type": "string"},
        "example": {"type": "string"}
      },
      "required": ["word", "meaning", "example"]
    }
  }
}'

Python 里直接用 provider 拿结构化输出：

import json
from anna.providers import get_provider_chain
provider = get_provider_chain()
raw = provider.complete(
    [{"role": "user", "content": "查词：serendipity"}],
    response_format={"type": "json_object"},
)
data = json.loads(raw)          # 有了 response_format，raw 一定是合法 JSON
print(data["word"], data["meaning"])

4.3 拿到 JSON 之后：用 Pydantic 校验

from pydantic import BaseModel, ValidationError
class Vocab(BaseModel):
    word: str
    meaning: str
    example: str = ""
raw = '{"word": "serendipity", "meaning": "机缘巧合", "example": "What a serendipity!"}'
try:
    item = Vocab.model_validate_json(raw)
    print(item.word, item.meaning)
except ValidationError as exc:
    print("格式不符，重试或报错：", exc)

工程原则：模型输出先过 Pydantic，校验失败就重试或按错误处理（见第 5 节），不要把未校验的字符串直接塞进业务逻辑。

5. 错误类型与异常处理

5.1 错误类型速查

状态码	类型	含义	处理建议
400	invalid_request	参数/格式错误	修请求，别重试
401/403	authentication	key 无效/无权限	检查 key
404	not_found	模型不存在/已下线	换模型
408 / 网络超时	timeout	请求超时	退避重试
429	rate_limit	限流/额度耗尽	退避重试或切换模型
500/502/503/504	server_error	服务端故障	退避重试
响应不合法	output_format	JSON 解析失败/结构缺失	重新生成或校验兜底

5.2 在本项目里怎么处理

provider 层把所有失败归一成 ProviderError，并分类（classify_provider_error，见 src/anna/providers/base.py）：

from anna.providers import ProviderError
from anna.providers.base import classify_provider_error
try:
    reply = provider.complete([{"role": "user", "content": "hi"}])
except ProviderError as exc:
    kind = classify_provider_error(exc)   # "rate_limit" / "authentication" / ...
    if kind == "rate_limit":
        # 退避后重试，或切到下一个候选模型
        pass
    elif kind == "authentication":
        raise  # 改 key，重试没用

自己写重试也可以这么干（框架层已内置同样思路，这里演示原理）：

import time
from anna.providers import ProviderError
from anna.providers.base import (
    ERROR_KIND_NETWORK,
    ERROR_KIND_RATE_LIMIT,
    ERROR_KIND_SERVER,
    classify_provider_error,
)
def call_with_retry(provider, messages, max_retries=3):
    for attempt in range(max_retries + 1):
        try:
            return provider.complete(messages)
        except ProviderError as exc:
            kind = classify_provider_error(exc)
            if kind not in (ERROR_KIND_NETWORK, ERROR_KIND_RATE_LIMIT, ERROR_KIND_SERVER) or attempt == max_retries:
                raise
            time.sleep(0.5 * (2 ** attempt))   # 0.5s → 1s → 2s

重试策略（内置在 _request_json）：对 429/5xx 做指数退避（0.5s → 1s → 2s…），ANNA_MAX_RETRIES 可配；不可回退错误（鉴权/内容过滤/ 无效请求）不重试、不切模型，直接抛给上层。

回退策略（FallbackProvider）：主模型失败且错误类型可回退时，按顺序切 fallback；主模型连续失败 3 次熔断 60 秒，避免一直打一个坏端点：

export ANNA_FALLBACKS="dashscope:qwen3.7-max,anthropic"

限流：网关按客户端令牌桶限流（ANNA_RATE_LIMIT_RPM），超限返回 429 + Retry-After：

{"error": {"message": "rate limit exceeded", "type": "rate_limit", "code": 429}}

6. 多模型调用基础

6.1 为什么要"统一"

直接对接各家 SDK 的问题：

协议不同：Chat Completions / Messages / GenerateContent 结构不一样；
切换成本高：换模型要改调用代码；
没有回退：一个模型挂了，整个功能就挂了；
成本难统计：不同模型计费规则不同。

解法 = 统一接口 + 适配器 + 注册表 + 配置驱动（本仓库 provider 层就是这个结构的练习实现）。

6.2 统一接口

class ChatProvider:
    def complete(self, messages, *, temperature=None, max_tokens=None,
                 response_format=None) -> str: ...
    def stream(self, messages, ...) -> Iterator[str]: ...
    def supports(self, feature: str) -> bool: ...

上层代码只认这个接口，不管背后是 DashScope 还是 Ollama。

6.3 模型路由：三种写法

网关的 model 字段支持三种路由方式（见 src/anna/router.py）：

# 1) auto：用 models.json 清单里"最近到期"的候选（免费额度优先用完）
-d '{"model": "auto", ...}'
# 2) 显式路由：厂商:模型
-d '{"model": "dashscope:qwen3.7-max", ...}'
# 3) 纯模型名：按关键词自动识别（qwen*→dashscope，claude*→anthropic，llama*→ollama）
-d '{"model": "qwen3.7-flash", ...}'

from anna.router import ModelRouter
router = ModelRouter(settings)
assert router.resolve("auto") == ("dashscope", "qwen3.7-max")
assert router.resolve("dashscope:qwen3.7-max") == ("dashscope", "qwen3.7-max")
assert router.resolve("claude-sonnet-4-5")[0] == "anthropic"

6.4 厂商元数据（ProviderSpec）

每个厂商的默认端点、key 环境变量、是否本地、thinking 注入方式都是声明式数据（src/anna/providers/specs.py），新增厂商不用写代码：

SPECS["dashscope"] = ProviderSpec(
    "dashscope", keywords=("qwen", "dashscope"),
    env_key="DASHSCOPE_API_KEY", backend="openai",
    default_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
    thinking_style="enable_thinking",
)

6.5 配置示例（本地 + 云端）

# 云端（DashScope 免费模型）
export ANNA_PROVIDER=openai
export ANNA_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
export ANNA_API_KEY=$DASHSCOPE_API_KEY
# 本地（Ollama，无需 key）
export ANNA_PROVIDER=ollama
export OLLAMA_BASE_URL=http://localhost:11434/v1
export OLLAMA_MODEL=llama3.1

小结

一次 LLM 调用 = POST 一个 JSON：model + messages + 采样参数 + 输出预算；
temperature/top_p 控制随机性，max_tokens 控制成本与长度；
token 是计费与 context window 的基本单位，超限靠分块/摘要/缓存解决；
结构化输出 = 原生 response_format + Pydantic 校验兜底；
错误要先分类再决定"重试 / 切模型 / 直接报错"；
多模型支持靠"统一接口 + adapter + 注册表 + 配置驱动"，而不是堆厂商代码。

动手练习：启动网关，用 curl 分别试 temperature=0 与 temperature=1 同一个问题对比两次输出；再用 response_format={"type":"json_object"} 让模型返回 JSON，并用 Pydantic 校验；最后故意把 ANNA_FALLBACKS 配一个失效模型，观察自动回退。

附录：讲义内容 × 项目实现对照

快速核对"讲义里讲的，仓库里有没有"。

讲义条目	项目实现位置	状态
统一请求/消息格式（Chat Completions）	`server/schemas.py`、`providers/openai_compatible.py`	✅ 已实现
Responses API 对比	—（认识性内容）	📘 通用说明
Anthropic / Gemini 协议转换	`providers/anthropic.py`、`providers/gemini.py`	✅ 已实现
temperature	网关 + 三家 adapter 均透传	✅ 已实现
top_p	—	⬜ 未实现（讲义已注明为扩展项）
max_tokens	网关 + 三家 adapter	✅ 已实现
finish_reason（length/content_filter）透传	—	⬜ 未实现（网关固定返回 stop）
Token 用量统计 / 成本估算	`usage.py`、`GET /v1/usage/summary`	✅ 已实现
长文分块 / 摘要 / prompt cache	—	⬜ Phase 0 后续任务（讲义仅讲思路）
结构化输出 json_object / json_schema	网关 + OpenAI 透传 + Gemini 映射 + Anthropic 隐藏工具	✅ 已实现（有测试）
Pydantic 校验模型输出	—（项目内 Pydantic 用于请求校验 `server/schemas.py`）	📘 讲义示例，待接入
ProviderError + 错误分类	`providers/base.py` `classify_provider_error`	✅ 已实现
重试（指数退避）	`providers/base.py` `_request_json`	✅ 已实现
跨模型回退 + 熔断	`providers/fallback.py`、`ANNA_FALLBACKS`	✅ 已实现
限流 429 + Retry-After	`server/rate_limit.py`	✅ 已实现
统一接口 ChatProvider	`providers/base.py`	✅ 已实现
模型路由（auto / 厂商:模型 / 关键词）	`router.py`	✅ 已实现
ProviderSpec 厂商元数据	`providers/specs.py`	✅ 已实现
云端 + 本地（Ollama）配置	环境变量 + `providers/ollama.py`	✅ 已实现

图例：✅ 已实现并有测试 / ⬜ 未实现（讲义中已注明或属后续阶段）/ 📘 通用知识或示例说明。

第 1 章讲义 · 01 LLM API 与模型调用基础

目录#

1. Chat Completions / Responses API#

1.1 请求结构：以 OpenAI Chat Completions 为标准#

1.2 Responses API（新接口，先认识）#

1.3 主流厂商协议对照#

1.4 在本项目里调用#

2. 模型参数：temperature、top_p、max_tokens#

2.1 temperature：控制随机性#

2.2 top_p：核采样#

2.3 max_tokens：输出预算#

3. Token 与 Context Window#

3.1 Token 是什么#

3.2 Context Window（上下文窗口）#

3.3 在本项目里看 token#

4. Structured Output#

4.1 为什么需要#

4.2 在本项目里的实现#

4.3 拿到 JSON 之后：用 Pydantic 校验#

5. 错误类型与异常处理#

5.1 错误类型速查#

5.2 在本项目里怎么处理#

6. 多模型调用基础#

6.1 为什么要"统一"#

6.2 统一接口#

6.3 模型路由：三种写法#

6.4 厂商元数据（ProviderSpec）#

6.5 配置示例（本地 + 云端）#

小结#

附录：讲义内容 × 项目实现对照#

目录