流式输出与多轮对话
生成长文本时，等模型全部算完再显示会让用户盯着空白屏幕发呆。流式输出（Streaming） 让 token 一边生成一边显示；多轮对话则是把历史消息追加进请求，让模型「记住」前面说的话。这篇把两者都讲清楚，附可直接运行的代码。
为什么要流式输出
非流式调用的问题：
- 延迟不可接受：生成 2000 token 的回答，即使 50 token/s，也需要 40 秒才能返回第一个字。
- 超时风险：HTTP 连接长时间无数据，网关或客户端可能主动断开。
流式的好处：首 token 延迟（Time-to-First-Token）只需 0.5~2 秒，用户立刻看到内容在流淌，体验与打字机相似。
流式调用
Claude SDK 提供流式上下文管理器，用 stream() 替换 create()：
from anthropic import Anthropic
client = Anthropic()
with client.messages.stream(
model="claude-opus-4-8",
max_tokens=2048,
system="你是一个简洁专业的技术助手，用中文回答。",
messages=[{"role": "user", "content": "解释一下什么是事件驱动架构，举一个实际案例。"}],
) as stream:
# 逐 token 打印，end="" 不换行，flush=True 立即刷新缓冲区
for text in stream.text_stream:
print(text, end="", flush=True)
# 流结束后获取完整 Message 对象（含 usage 统计）
message = stream.get_final_message()
print(f"\n\n[usage: {message.usage.input_tokens} in / {message.usage.output_tokens} out]").get_final_message() 在流结束后返回完整的 Message 对象，包含 usage（token 计费数据）和 stop_reason。不需要手动拼接 token。多轮对话
Claude API 本身是无状态的——它不记得上一次调用说了什么。多轮对话的实现方法是：把历史消息追加进 messages 数组一起发送。
from anthropic import Anthropic
client = Anthropic()
def chat():
history = []
system = "你是一个有帮助的技术助手，用中文回答。"
print("开始对话（输入 exit 退出）\n")
while True:
user_input = input("你：").strip()
if user_input.lower() == "exit":
break
history.append({"role": "user", "content": user_input})
with client.messages.stream(
model="claude-opus-4-8",
max_tokens=1024,
system=system,
messages=history, # 每次都把完整历史发过去
) as stream:
print("Claude：", end="", flush=True)
full_reply = ""
for text in stream.text_stream:
print(text, end="", flush=True)
full_reply += text
print()
# 把模型的回复追加进历史，供下一轮使用
history.append({"role": "assistant", "content": full_reply})
chat()上下文窗口管理
每轮都把完整历史发过去意味着：对话越长，每次请求的 token 越多，成本越高，最终还会超出模型的上下文窗口（Claude Opus 4.8 为 1M token）。
常见管理策略：
| 策略 | 适用场景 | 实现方式 |
|---|---|---|
| 滑动窗口 | 只需记住近几轮 | 只保留最后 N 条消息，超出时丢弃最早的 |
| 摘要压缩 | 长会话且历史重要 | 每隔 K 轮，把旧消息总结成一段文字，替换掉原消息 |
| Token 预算控制 | 精确控制成本 | 用 client.messages.count_tokens() 估算，超出阈值时触发压缩 |
滑动窗口示例：
MAX_TURNS = 10 # 保留最近 10 轮（user+assistant 各算一条）
# 在追加之前检查
if len(history) > MAX_TURNS * 2:
# 如果有 system 消息，保留之（通过 system 参数传的不在 history 里，无需处理）
history = history[-(MAX_TURNS * 2):]流式 + 工具调用
流式模式与工具调用可以结合。stream.text_stream 只会生成文本 token，工具调用事件通过其他事件类型触发。更简单的方式是用 .get_final_message() 在流结束后再检查 stop_reason：
with client.messages.stream(model="claude-opus-4-8", max_tokens=1024,
tools=tools, messages=messages) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
final = stream.get_final_message()
if final.stop_reason == "tool_use":
# 处理工具调用（同非流式逻辑）
...提示词缓存
当 system 提示很长（如带了大量文档）、且每轮对话都重复发送时，开启**提示词缓存（Prompt Caching）**可以大幅节省成本：
import anthropic
client = Anthropic()
# 在需要缓存的内容块末尾加 cache_control
response = client.messages.create(
model="claude-opus-4-8",
max_tokens=1024,
system=[
{
"type": "text",
"text": "你是技术助手。" + long_knowledge_base, # 假设有长文档
"cache_control": {"type": "ephemeral"}, # 标记为可缓存
}
],
messages=[{"role": "user", "content": user_question}],
betas=["prompt-caching-2024-07-31"],
)缓存命中时，被缓存部分的 input token 费用降至 10%（读取缓存比重新处理便宜约 90%）。适合 system prompt 超过 1K token 且请求频繁的场景。
一句话小结
流式 = 用 stream() 替换 create()，逐 token 打印；多轮 = 把历史消息完整追加进 messages 数组。两者组合是大多数对话产品的标准实现，加上窗口管理和提示词缓存，就能把成本和体验都控制好。