04 · 多轮对话与流式输出

🎯 难度：初级⏱ 预计：60 分钟# messages# stream# CLI

多轮对话就是维护历史

模型不会自动记住上一轮，因此每次都带上历史 messages。本节依赖 config.py。

from openai import OpenAIError
from config import get_client, model_name

client = get_client()
messages = [{"role": "system", "content": "你是耐心的中文助手。"}]

while True:
    try:
        text = input("你：").strip()
    except (EOFError, KeyboardInterrupt):
        break
    if text == "/exit":
        break
    if text == "/clear":
        messages = messages[:1]
        print("已清空历史")
        continue
    if not text:
        continue

    # 使用副本，失败时不把未完成的一轮提交到历史
    pending = messages + [{"role": "user", "content": text}]
    try:
        response = client.chat.completions.create(
            model=model_name(), messages=pending, max_tokens=600
        )
    except OpenAIError as error:
        print(f"请求失败：{type(error).__name__}")
        continue
    answer = response.choices[0].message.content or ""
    print("助手：", answer)
    messages = pending + [{"role": "assistant", "content": answer}]

运行 uv run chat.py。说「我叫小林」，再问「我叫什么？」。运行 /clear 后再问，模型应无法从历史知道你的名字。

历史越长越贵

历史会反复发送并计入输入用量。上面的程序仅用于学习；长对话需要 Token 预算、裁剪或摘要策略，见 记忆系统。

流式：不等整段完成再展示

流式不是模型突然更快了，而是把已生成的小片段尽早送到你手里，改善首字响应时间。

from config import get_client, model_name

client = get_client()
parts = []
stream = client.chat.completions.create(
    model=model_name(),
    messages=[{"role": "user", "content": "用三段话解释什么是 Agent"}],
    stream=True,
    max_tokens=600,
)
try:
    for chunk in stream:
        if not chunk.choices:
            continue
        delta = chunk.choices[0].delta.content
        if delta:
            print(delta, end="", flush=True)
            parts.append(delta)
finally:
    stream.close()
print()
answer = "".join(parts)

delta 是增量，不能把每块当作完整回答。flush=True 让终端立刻显示；拼好的 answer 才适合加入历史。

如果流式中途失败，不要把残缺回答当作成功的一轮。产品上应标注中断、提供重试，并关闭上游连接。

工具调用也可能流式到达

工具的名称与 arguments 也可能分块输出，不能在收到半个 JSON 时执行工具。初学阶段，先用非流式完成工具决策，再流式输出最终回答；全栈项目 就采用这个策略。

练习

给 chat.py 加 /clear 以外的 /help 命令，然后把普通回答改成流式输出。

验收：最终回答只追加一次；网络失败不会留下半轮历史；Ctrl+C 可以退出，不再发送请求。