会话管理与 JSON 输出
第 5 章把模型跑通了：本地推理服务就位，system / user / assistant 的消息体系也建立起来。但离"用好"还差两块拼图。其一是跨轮次的记忆管理——所谓"连续对话"，其实是你的代码在维护一个不断增长的…
第 5 章把模型跑通了：本地推理服务就位，system / user / assistant 的消息体系也建立起来。但离"用好"还差两块拼图。其一是跨轮次的记忆管理——所谓"连续对话"，其实是你的代码在维护一个不断增长的 messages 数组，数组每长一分，费用涨一分、延迟高一分，直到撞上上下文窗口的天花板；其二是让输出可被程序消费——模型天生输出自由文本，人能读，程序没法稳定解析。本章把这两块补齐：前半部分把会话的"持续经营"工程化，封装 Session 类、算清 token 账、掌握三种历史裁剪策略、吃透缓存命中的原理；后半部分给出结构化输出从脆弱到稳固的完整方案链。这是"第二章 怎么用好它"的收官——从此模型不只是"聊得来"，还要"管得住、读得出"。
1. 最小会话：20 行的"记忆幻觉"
先封装一个能用的 Session，把第 5 章的数组维护逻辑收进类里：
from openai import OpenAI
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
class Session:
def __init__(self, system: str, model: str = "qwen3:4b"):
self.model = model
self.messages = [{"role": "system", "content": system}]
def chat(self, user_input: str) -> str:
self.messages.append({"role": "user", "content": user_input})
resp = client.chat.completions.create(
model=self.model,
messages=self.messages, # 全量历史随请求重发
)
answer = resp.choices[0].message.content
self.messages.append({"role": "assistant", "content": answer})
return answer
# 使用
s = Session(system="你是一个简洁的 Python 助教")
print(s.chat("什么是递归？"))
print(s.chat("给我一个例子")) # 它"记得"上文在聊递归
print(len(s.messages)) # 5 条：system + 2轮 × (user+assistant)
这个类是本章前半部分的主角，后面逐步给它加能力。先注意它的成本结构：第 N 轮请求时，之前所有轮次的 user+assistant 消息都要作为输入 token 重新计费（第 4 章 token 经济学）。对话越长越贵，这不是 API 的设计缺陷，是自回归架构的物理现实。
2. 算清 token 账：事前估算，事后校准
管理上下文的前提是计量。两个手段，一个事后、一个事前：
事后（精确）：从响应的 usage 里读真实值（第 5 章见过），累计起来就是会话的真实开销：
def chat(self, user_input):
...
self.total_tokens += resp.usage.total_tokens
事前（估算）：发送前预判会不会超窗，只能估算。第 1 章的换算率在这里落地：
def estimate_tokens(text: str) -> int:
"""粗略估算中英混合文本的 token 数（±15% 误差）"""
# 英文：约 4 字符/token；中文：约 1.5 字符/token
cn = sum('\u4e00' <= c <= '\u9fff' for c in text)
en = len(text) - cn
return int(cn / 1.5 + en / 4)
def total_messages_tokens(messages: list[dict]) -> int:
# 每条消息有少量固定开销（role 标记等），约 4 token
return sum(estimate_tokens(m["content"]) + 4 for m in messages)
本地开发可以做得更准——用模型自己的分词器。Ollama 的响应里带有精确计数，只让它生成 1 个 token 就能拿到输入侧的账单：
# uv add ollama
import ollama
def count_precise(text: str, model: str = "qwen3:4b") -> int:
# 只生成 1 个 token 即停：prompt_eval_count 就是输入文本的精确 token 数
resp = ollama.generate(
model=model, prompt=text,
options={"num_predict": 1},
)
return resp.prompt_eval_count or 0
生产中推荐组合拳：发送前用估算值做粗筛，发送后用 usage 真值校准水位线。
3. 历史裁剪：三种策略
窗口是有限的（第 4 章），历史无限增长必然出事。三种策略，复杂度和效果递增：
3.1 截断（保头保尾）
最朴素：优先保留 system（system + 最近 K 轮），丢弃中间旧轮次。
def truncate(self, max_msgs: int = 10):
"""保留 system 和最近 max_msgs 条"""
system, rest = self.messages[0], self.messages[1:]
self.messages = [system] + rest[-max_msgs:]
def chat(self, user_input: str):
...
if total_messages_tokens(self.messages) > self.window_budget:
self.truncate()
优点：一行逻辑，O(1) 开销。缺点：硬切断会丢关键信息（用户在第一轮说的偏好、早前布置的任务）。适合闲聊类短会话。
3.2 滑动窗口 + 摘要压缩
丢之前先"榨汁"：把将被丢弃的历史让模型压缩成摘要，替换原文，常驻 system 附近：
def summarize_old(self) -> str:
"""把旧历史压缩为一段摘要"""
resp = client.chat.completions.create(
model=self.model,
messages=[
{"role": "system",
"content": "将以下对话压缩为不超过150字的关键信息摘要，保留：用户目标、已确认的事实、尚未完成的任务。"},
{"role": "user", "content": repr(self.messages)},
],
temperature=0.2,
)
return resp.choices[0].message.content
def compact(self, keep_recent: int = 4):
"""压缩旧历史，保留最近几轮原文"""
system = self.messages[0]
old, recent = self.messages[1:-keep_recent], self.messages[-keep_recent:]
if not old:
return
summary = self.summarize_old()
self.messages = [
system,
{"role": "system", "content": f"[会话历史摘要]\n{summary}"},
*recent,
]
这是性价比最高的主流方案：信息以压缩形式保留，近期交互保持原文精度。注意摘要本身也是一次 LLM 调用（有损、有成本），而且压缩比要克制——摘要塞太满，模型容易把摘要当圣旨。
3.3 检索式（记忆外挂）
长会话的正确终局：历史存进向量库（第 1 章的 embedding + 余弦相似度），每轮只检索相关片段注入上下文。这已经是"长期记忆"的范畴，第 15 章 Agent 记忆专门展开，此处按下不表。
3.4 策略选择
| 策略 | 成本 | 信息保留 | 适用 |
|---|---|---|---|
| 截断 | 零 | 差 | 短会话、闲聊 |
| 摘要压缩 | 每次压缩一次调用 | 中（有损） | 多轮任务会话 |
| 检索式 | 向量库 + 每轮检索 | 按相关性 | 跨天/长期记忆 |
一个容易忽略的细节：裁剪必须在"成对边界"上进行。从一对 user/assistant 的中间切开，会留下"有问无答"或"有答无问"的畸形历史——多数模型能容忍，但第 7 章之后的历史里会出现 tool 调用对，那种历史被拦腰截断会直接报错。
4. 缓存命中：省钱的隐形杠杆
第 4 章埋的线在此展开。回顾原理：注意力计算需要全部历史位置的 K、V 向量（第 3 章），如果两次请求的前缀完全相同，这些 K/V 可以直接复用——省掉整个 prefill 阶段的重算。这就是提示词缓存（prompt caching）。
由此推出黄金法则——静态内容在前，动态内容在后：
❌ 差的排列（每轮前缀都变，缓存全失效）:
[含当前时间戳的system] [历史...] [新输入]
✅ 好的排列（前缀稳定，命中到最新一条）:
[固定system] [历史(追加式)] [新输入]
各家的实现形态：
| 厂商 | 机制 | 要点 |
|---|---|---|
| Ollama | 自动前缀缓存 | 同一客户端连续请求自动命中；keep_alive 参数控制模型驻留内存，避免反复加载 |
| DeepSeek | 自动上下文缓存 | 无需任何代码改动，命中部分按约 1/10 计价 |
| OpenAI / Anthropic | 显式标记缓存 | 前缀稳定过阈值（如 1024 token）自动缓存，命中打折 |
对代码的影响只有一条：系统提示词和工具定义（第 7 章）保持字节级稳定——包括消息顺序。把"当前时间"这类每轮变化的信息放到最后一条 user 消息里，而不是 system 里（第 5 章的模板化要为此让步）。
# 时间戳放 user 消息尾部，保住前缀缓存
messages = [
{"role": "system", "content": FIXED_SYSTEM}, # 字节级稳定
*history, # 只追加，不改动
{"role": "user", "content": f"{q}\n\n[当前时间 {now}]"},
]
5. 生产级 Session 的完整形态
把本章前半部分的所有能力组装起来，这就是第三章 Agent 实战的会话地基：
from openai import OpenAI
class Session:
def __init__(self, system: str, model: str = "qwen3:4b",
window_budget: int = 6000, keep_recent: int = 4):
self.model = model
self.window_budget = window_budget
self.keep_recent = keep_recent
self.messages = [{"role": "system", "content": system}]
self.total_tokens = 0
def chat(self, user_input: str) -> str:
self.messages.append({"role": "user", "content": user_input})
self._enforce_budget() # 每轮发送前检查
resp = client.chat.completions.create(
model=self.model, messages=self.messages,
)
answer = resp.choices[0].message.content
self.messages.append({"role": "assistant", "content": answer})
self.total_tokens += resp.usage.total_tokens
return answer
def _enforce_budget(self):
if total_messages_tokens(self.messages) <= self.window_budget:
return
self.compact(keep_recent=self.keep_recent) # 摘要压缩
# 压缩后仍超（极端情况），退化为硬截断
if total_messages_tokens(self.messages) > self.window_budget:
self.truncate(max_msgs=self.keep_recent * 2)
注意 chat 里 finish_reason 没检查——留给你：参照第 5 章补上 length 截断处理，这是很好的练习。
至此，你已经能让模型"持续对话"了。但有个问题越来越刺眼：模型所有输出都是自由文本——人能读，程序没法稳定解析。本章后半部分就补上这块拼图。
6. 问题：自由文本是程序的地雷
模型天生输出自由文本，但程序需要的是机器可解析的数据。这个"翻译层"是 Agent 的命门：第 7 章的 tool calling、第 8 章的 Agent 循环、第 15 章的评测打分，全都建立在"模型输出能被代码可靠消费"之上。
你要做"用户情绪分类"功能，最直觉的做法是提示词里写"请返回 JSON"：
resp = client.chat.completions.create(
model="qwen3:4b",
messages=[{"role": "user",
"content": '分析这句话的情绪，返回JSON：{"sentiment": "正面|负面|中性", "score": 0到1}\n用户输入：这个产品用起来太糟心了'}],
)
print(resp.choices[0].message.content)
你会收获五花八门的"惊喜"：
好的！以下是分析结果： ← 有人在说话
{"sentiment": "负面", "score": 0.85} ← 有前缀废话
```json
{"sentiment": "负面", "score": 0.85} ← markdown 代码块包裹
{"sentiment": "负面", "score": 85} ← score 忘了是 0~1，给成百分制
{"sentiment": "负向", "score": 0.85} ← 枚举值自由发挥
直接 json.loads 一把梭，程序当场崩溃。根因在第 4 章：模型是续写统计机器，"格式正确"只是高概率倾向，不是保证。要工程级可靠，必须用约束 + 校验 + 重试的组合拳。
7. 第一道防线：JSON 模式
多数推理服务提供 JSON 模式——通过 response_format 开启：
resp = client.chat.completions.create(
model="qwen3:4b",
messages=[
{"role": "system", "content": "你是情感分析引擎，只输出JSON。"},
{"role": "user", "content": "分析：这个产品用起来太糟心了"},
],
response_format={"type": "json_object"}, # JSON 模式
)
它保证的是语法合法的 JSON（引号闭合、括号配对）。但注意它不保证内容符合你的 schema——字段名可能变、枚举可能自由发挥、数值范围可能漂移。语法关过了，语义关还没过。
8. 第二道防线：Pydantic 校验
用 Pydantic 把期望的结构声明成类型，输出必须过审才算数：
# uv add pydantic
from pydantic import BaseModel, Field, ValidationError
from typing import Literal
class Sentiment(BaseModel):
sentiment: Literal["正面", "负面", "中性"] # 枚举锁死
score: float = Field(ge=0, le=1) # 范围锁死
keywords: list[str] = Field(default_factory=list) # 可选字段给默认值
data = Sentiment.model_validate_json(resp.choices[0].message.content)
任何不符合约定的输出（枚举外值、score=85、缺字段）都会抛 ValidationError，把"看起来对"变成"可证明对"。
但校验只负责"发现问题"，问题出现后怎么办？——把错误喂回去重试。
9. 第三道防线：校验-重试循环
这个模式是结构化输出的核心工程模式（也是第 7 章工具调用可靠性的雏形）：
import json
from openai import OpenAI
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
def extract_json(text: str) -> dict:
"""从模型输出中鲁棒地提取 JSON 对象"""
# 剥掉 markdown 代码块围栏
if "```" in text:
text = text.split("```")[1]
if text.startswith("json"):
text = text[4:]
# 取第一个 { 到最后一个 } 之间的内容（跳过前后废话）
start, end = text.find("{"), text.rfind("}")
return json.loads(text[start:end + 1])
def ask_structured(question: str, retries: int = 2) -> Sentiment:
messages = [
{"role": "system",
"content": "你是情感分析引擎。只输出一个JSON对象，结构："
'{"sentiment":"正面|负面|中性","score":0到1的小数,"keywords":[]}'},
{"role": "user", "content": question},
]
for attempt in range(retries + 1):
resp = client.chat.completions.create(
model="qwen3:4b",
messages=messages,
temperature=0.1, # 结构化任务用低温（第4章）
)
try:
return Sentiment.model_validate(extract_json(resp.choices[0].message.content))
except (json.JSONDecodeError, ValidationError) as e:
# 关键：把错误信息作为新消息喂回去，让模型自我修正
messages += [
{"role": "assistant", "content": resp.choices[0].message.content},
{"role": "user", "content": f"输出不符合要求：{e}\n请严格按要求重新输出，只输出JSON。"},
]
raise RuntimeError(f"{retries + 1} 次尝试均失败")
三个细节值得琢磨：
- 错误信息是最有效的提示词。
ValidationError的报文（"score: Input should be less than or equal to 1"）对模型是精确的修正指令 extract_json的鲁棒解析是无奈但必要的防御——即使有 JSON 模式，也常遇到模型在 JSON 前后"加戏"- 重试次数要设上限：本地小模型偶尔会陷入"错-重试-还是错"的循环，两三次失败就该换策略（换模型/降低 schema 复杂度），而不是无限重试
10. 终极方案：约束解码（constrained decoding）
前面都是"生成后补救"。存在一种釜底抽薪的手段：在采样阶段就屏蔽非法 token。
回忆第 4 章：模型每步在词表上产出一个概率分布，然后采样选出一个。约束解码的做法：在采样前，把所有不符合 schema 的候选 token 的概率直接置零——比如已经输出了 {"sentiment": "负，下一步只允许 面 这一个 token，其他全部禁止。
效果是100% 语法+结构合规（代价是牺牲一点内容自由度，且对解码速度有少量影响）。工程接入方式：
| 层级 | 方案 |
|---|---|
| API 层 | OpenAI 的 response_format={"type":"json_schema", strict:true}，严格模式 |
| 框架层 | Outlines / xgrammar / llama.cpp 的 grammar 模式（本地模型可用） |
| Ollama | format 参数可传 JSON Schema（原生 API） |
# 原生 Ollama API 传入 schema（OpenAI 兼容层之外的方式）
import ollama
schema = Sentiment.model_json_schema() # Pydantic 直接生成 JSON Schema
resp = ollama.chat(
model="qwen3:4b",
messages=[{"role": "user", "content": "分析：这个产品用起来太糟心了"}],
format=schema, # 约束解码：输出必合法
)
注意分工：约束解码保证结构，Pydantic 校验保证语义（枚举、范围、业务规则）。生产中两者叠加，再加校验-重试兜底，就是完整方案。
11. Schema 设计的实战建议
同样的约束目标，schema 写法不同，稳定性差异巨大：
| 做法 | 建议 |
|---|---|
| 字段越少越好 | 每个字段都是模型的一次"发挥机会"，能删则删 |
| 用 Literal 锁枚举 | 开放字符串是自由发挥的重灾区 |
| 数值给范围 | Field(ge=0, le=1) 比"0到1"的口头约定硬一万倍 |
| 可选字段给默认值 | 减少模型"必须编一个值"的压力 |
| 字段描述写进 schema | Field(description=...) 会进入提示，模型看得见 |
| 拆大 schema | 需要十几个字段时，拆成多次调用各取所需 |
最后一条是架构级经验：别让一次调用输出整个系统需要的所有数据。多个小而清晰的 schema 串联（分析→分类→生成），比一个巨型 schema 一次到位稳定得多——这也为第 8 章的 Agent 多步执行埋下伏笔。
本章小结
会话与上下文管理：
- 会话 = 客户端维护的 messages 数组；第 N 轮输入包含全部历史，成本随轮次线性上升
- token 计量：事前估算（字符比例法）+ 事后精确（usage 真值校准）
- 三种裁剪策略：截断（零成本有损）、摘要压缩（主流选择）、检索式（第 15 章的长期记忆）
- 裁剪要切在轮次边界上，否则后续 tool 调用对会被截断报错
- 缓存命中的黄金法则：静态在前动态在后；时间戳放 user 消息尾部；system 保持字节级稳定
结构化输出：
- 自由文本对程序是雷区；可靠输出 = 约束 + 校验 + 重试的三层组合
- JSON 模式只保证语法合法；Pydantic 的 Literal/Field 才锁得住语义
- 校验-重试循环：把 ValidationError 喂回给模型是最有效的修正提示
- 约束解码在采样阶段屏蔽非法 token（第 4 章分布的改写），结构 100% 合规
- Schema 设计原则：少字段、锁枚举、给范围、拆大 schema
到这里，"第二章 怎么用好它"正式收官。你已经掌握：调用模型与组织输入（第 5 章）、经营会话与锁定输出（本章）。万事俱备，只欠东风——让模型动起手来。