第 1 章 · LLM API 基础

Agent 的地基是对 LLM API 的工程化使用。这一章不假设你用过任何框架——我们从 HTTP 协议层开始，把每一个概念讲到能自己实现的程度。

1.1 Chat Completions 协议

OpenAI 的 Chat Completions 是事实上的行业标准协议，几乎所有模型供应商和 Agent 框架都以它为兼容目标。

一次请求的核心是 messages 数组，每个元素有 role 和 content：

{
    "model": "gpt-4o",
    "messages": [
        {"role": "system", "content": "你是一个严谨的工程师。"},
        {"role": "user", "content": "解释什么是幂等性"},
        {"role": "assistant", "content": "幂等性是指……"},
        {"role": "user", "content": "举个例子"}
    ],
    "temperature": 0.7,
    "max_tokens": 1024,
}

三个 role 的语义：

system：最高优先级指令，通常放身份设定、行为约束、工具使用规范。它不是"更聪明的指令"，而是模型对齐训练时被特别加权的前缀——工程上应把它当作可以缓存的常量前缀来设计（见 1.4 前缀缓存）。
user：终端用户的输入。注意：在 Agent 场景里，工具结果也会以 user 或 tool 角色回灌——role 本质上是"谁在说话"的标记，不是权限边界。
assistant：模型的历史输出。多轮对话就是把整段历史原样重传——模型是无状态的，"记忆"永远是你把历史再喂一遍。

关键参数的工程语义：

参数	作用	Agent 场景建议
`temperature`	采样随机性	工具选择/结构化输出用 0-0.2；创作类才调高
`max_tokens`	输出上限	防止死循环烧钱，但太小会截断 JSON
`stop`	停止序列	控制模型不要越权续写
`response_format`	JSON mode	结构化输出时配合 schema 使用
`tools` / `tool_choice`	工具声明与强制	见 1.2

1.2 Function Calling：模型选择，代码执行

工具调用是 Agent 的本质机制，必须精确理解它的分工：

你在请求里声明工具清单（名称 + JSON Schema 描述 + 参数定义）；
模型不执行任何东西，它只输出一个结构化决定："我要调 get_weather，参数是 {"city": "北京"}"；
执行发生在你的代码里，你把结果作为消息追加进历史，再次请求；
模型看到工具结果，继续生成（可能继续要工具，也可能给最终答案）。

一个完整的可运行示例（只用官方 SDK，不用任何 Agent 框架）：

import json
from openai import OpenAI

client = OpenAI()  # 从环境变量读 OPENAI_API_KEY / BASE_URL

def get_weather(city: str) -> dict:
    # 真实实现里这里是 HTTP 调用；示例直接返回假数据
    return {"city": city, "temp_c": 22, "condition": "多云"}

TOOLS = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查询指定城市当前天气。当用户问到天气、气温、出行建议时使用。",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名，如「北京」"}
            },
            "required": ["city"],
        },
    },
}]

messages = [
    {"role": "system", "content": "你是天气助手，回答前必须查询实时数据。"},
    {"role": "user", "content": "北京今天适合穿什么？"},
]

for _ in range(5):  # 工具循环轮数预算，防止死循环
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=TOOLS,
    )
    msg = resp.choices[0].message

    if not msg.tool_calls:            # 模型不再要工具 → 最终答案
        print(msg.content)
        break

    messages.append(msg)              # assistant 消息（含 tool_calls）必须回灌
    for tc in msg.tool_calls:
        args = json.loads(tc.function.arguments)
        result = get_weather(**args)
        messages.append({
            "role": "tool",
            "tool_call_id": tc.id,    # 用 id 关联，多工具并行时靠它配对
            "content": json.dumps(result, ensure_ascii=False),
        })

三个最常犯的错误，现在就建立肌肉记忆：

忘记回灌 assistant 消息本身。含 tool_calls 的 assistant 消息必须先进历史，tool 结果跟在后面——模型靠这个配对理解"我刚才要了什么"。
tool_call_id 配错。并行调用多个工具时，每个结果用它的 tool_call_id 关联，顺序错乱会让模型把结果张冠李戴。
把工具结果当字符串拼接。必须是独立的结构化消息，而不是"把结果附加在 user 消息里"——后者会让模型分不清用户说了什么和系统返回了什么。

描述即编程（description engineering）：工具的 description 和参数描述是写给模型看的文档，模型选择哪个工具、怎么填参数完全取决于它。经验法则：description 写"什么时候该用 + 什么时候不该用"，参数描述写具体格式与示例。工具多了之后（几十个以上），描述质量直接决定选工具的准确率。

1.3 流式输出：SSE 逐块解析

流式（stream=true）把一次补全拆成 SSE（Server-Sent Events）事件序列逐块返回。Agent 必须理解流式，原因有二：用户等待首字的体验；以及工具调用在流式下的形态——tool_calls 的参数 JSON 是分块拼起来的。

手写一个 SSE 解析器（这是理解流式最彻底的方式）：

import json
import httpx

def stream_chat(messages, model="gpt-4o"):
    """逐事件处理流式响应。真实生产代码还需要：断线重连、超时、usage 收集。"""
    payload = {"model": model, "messages": messages, "stream": True,
               "stream_options": {"include_usage": True}}
    tool_calls = {}  # index -> {id, name, arguments}，流式中分块累积

    with httpx.stream("POST", f"{client.base_url}/chat/completions",
                      json=payload, timeout=120) as resp:
        for line in resp.iter_lines():
            if not line.startswith("data: "):
                continue                 # SSE 注释行、keep-alive 直接跳过
            data = line[len("data: "):]
            if data == "[DONE]":
                break
            chunk = json.loads(data)
            if chunk.get("usage"):       # 流式的 usage 在最后一个 chunk
                print(f"\n[usage] {chunk['usage']}")
            for delta in chunk["choices"]:
                d = delta.get("delta", {})
                if d.get("content"):
                    yield {"type": "content", "text": d["content"]}
                for tc in d.get("tool_calls") or []:
                    slot = tool_calls.setdefault(tc.index, {"id": "", "name": "", "arguments": ""})
                    if tc.id: slot["id"] = tc.id
                    if tc.function and tc.function.name: slot["name"] = tc.function.name
                    if tc.function and tc.function.arguments:
                        slot["arguments"] += tc.function.arguments  # 参数是分块拼接的！
        for slot in tool_calls.values():
            slot["arguments"] = json.loads(slot["arguments"] or "{}")
            yield {"type": "tool_call", **slot}

两个协议细节：

SSE 注释行：某些供应商/网关会发 : keep-alive 注释行防止连接超时，解析时必须跳过，否则 json.loads 直接崩；
usage 在流末尾：默认流式不返回 token 统计，要开 stream_options.include_usage，且它出现在 choices 为空的最后一个 chunk——统计代码必须容忍空 choices。

1.4 Token 与前缀缓存

Tokenizer 原理：模型不处理字符，处理 token（子词单元）。中文大约 1 字 ≈ 0.6-1 token，英文 1 词 ≈ 1.3 token，代码和特殊符号方差很大。精确计算用模型对应的 tokenizer（如 tiktoken），估算用字符数 × 系数即可。

计费结构：输入 token 和输出 token 分开计价，输出通常是输入的 3-5 倍。这决定了两个工程方向：压输入（上下文工程，第 3 章）、限输出（max_tokens + 压缩指令）。

前缀缓存（prompt caching）：请求的公共前缀（system prompt、工具定义、few-shot 示例）的 KV 计算结果可以被缓存复用，命中部分大幅降价。工程含义：

把不变的内容放前缀，变化的内容放后缀。比如 system(稳定) + 工具定义(稳定) + 历史(递增) + 新输入(变) 这个顺序天然缓存友好；反之如果每次把当前时间插在 system 前面，整条前缀全部失效；
多轮对话的历史是只增不改的，天然是缓存友好的——不要在中间做无谓改写（例如"清理"历史里的口头禅），一次改写废掉全部缓存；
工具清单是常驻前缀的一部分，200 个工具的 schema 可能比对话本身还长——工具列表越稳定越好。

1.5 重试、超时与幂等

LLM API 的失败模式：网络抖动（连接重置）、限流（429，带 Retry-After 头）、上游过载（529/503）、内容过滤拦截。工程规范：

import asyncio, random

async def call_with_retry(fn, max_retries=5, base_delay=1.0):
    for attempt in range(max_retries + 1):
        try:
            return await fn()
        except RateLimitError as e:
            retry_after = float(getattr(e, "retry_after", 0) or 0)
        except (ConnectionError, TimeoutError, InternalServerError):
            retry_after = 0                       # 网络/上游错误走指数退避
        except (AuthenticationError, InvalidRequestError, ContentFilterError):
            raise                                 # 不可恢复错误，重试没有意义
        if attempt == max_retries:
            raise
        delay = retry_after or base_delay * (2 ** attempt) + random.uniform(0, 1)
        await asyncio.sleep(delay)

三条纪律：

指数退避 + 抖动：不加抖动的重试在故障恢复瞬间会形成同步惊群；
错误分类决定行为：4xx 里只有 429 可重试；鉴权错误重试一百次也是失败；
超时是预算的一部分：Agent 每轮工具循环都有总时限，单次 LLM 调用的超时必须小于总预算，否则一轮 hang 死整个任务。

1.6 多模态输入

图片进入 messages 的形式是 content 数组：

messages = [{"role": "user", "content": [
    {"type": "text", "text": "这张截图里的报错是什么？"},
    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}},
]}]

工程要点：图片 token 消耗与分辨率成正比（通常按 512×512 块计价），Agent 里传图前先降采样能省一个量级的成本；多模态解析结果应该缓存（同一文件反复解析是常见浪费）。

实现作业

实现一个零框架的多轮工具 Agent 客户端，要求：

支持 system prompt + 至少 3 个工具；
支持流式输出（内容实时打印，工具调用分块累积）；
带 token 计量（输入/输出分开累计）和轮数预算；
带指数退避重试；
全程不 import 任何 agent 框架。

完成后你会亲手踩到本章所有细节：id 配对、分块参数拼接、空 choices 的 usage chunk、前缀顺序对缓存的影响。

深入材料

OpenAI Function Calling 官方指南（platform.openai.com/docs/guides/function-calling）
Anthropic Messages API 参考（docs.anthropic.com，注意 Anthropic 的 tool_use/tool_result 消息形态与 OpenAI 不同，但思想一致）
tiktoken 源码（BPE 分词的实现级理解）

Agent 工程 · 第 1 章｜LLM API 基础：协议、工具调用、流式与重试

第 1 章 · LLM API 基础

1.1 Chat Completions 协议

1.2 Function Calling：模型选择，代码执行

1.3 流式输出：SSE 逐块解析

1.4 Token 与前缀缓存

1.5 重试、超时与幂等

1.6 多模态输入

实现作业

深入材料

读者留言

Agent 工程 · 第 1 章｜LLM API 基础：协议、工具调用、流式与重试

第 1 章 · LLM API 基础

1.1 Chat Completions 协议

1.2 Function Calling：模型选择，代码执行

1.3 流式输出：SSE 逐块解析

1.4 Token 与前缀缓存

1.5 重试、超时与幂等

1.6 多模态输入

实现作业

深入材料

读者留言

相关阅读