Provider 抽象与可取消的流式对话

先看一次失败的请求

最初的 /chat 通常这样写：

answer = client.chat.completions.create(model=MODEL, messages=messages)
return {'answer': answer.choices[0].message.content}

当模型需要 12 秒时，浏览器 12 秒内什么都看不到；反向代理还可能在 60 秒后直接断开。把 stream=True 加上也不够：你仍然需要定义事件格式、会话 ID、取消路径和断线重连行为。

Provider 是稳定边界

Preview

业务代码只依赖接口，两个实现可以互换

from collections.abc import AsyncIterator
from dataclasses import dataclass

@dataclass
class Delta:
    text: str = ''
    finish_reason: str | None = None
    usage: dict | None = None

class LLMProvider(Protocol):
    async def stream(self, messages: list[dict], *, temperature: float = 0.2) -> AsyncIterator[Delta]: ...

class OpenAICompatibleProvider:
    def __init__(self, client, model: str):
        self.client, self.model = client, model

    async def stream(self, messages, *, temperature=0.2):
        response = await self.client.chat.completions.create(
            model=self.model, messages=messages, temperature=temperature,
            stream=True, stream_options={'include_usage': True},
        )
        async for chunk in response:
            choice = chunk.choices[0] if chunk.choices else None
            delta = choice.delta.content if choice and choice.delta else ''
            yield Delta(text=delta or '', finish_reason=choice.finish_reason if choice else None,
                        usage=getattr(chunk, 'usage', None))

MockProvider 不是玩具

Mock 必须“决定性”，不能随机生成文字：

class MockProvider:
    def __init__(self, script: list[str], delay: float = 0):
        self.script, self.delay = script, delay

    async def stream(self, messages, *, temperature=0.2):
        for token in self.script:
            if self.delay:
                await asyncio.sleep(self.delay)
            yield Delta(text=token)
        yield Delta(finish_reason='stop')

测试可以断言每一帧，不需要等待真实模型，也不会因为供应商改了措辞而抖动。

SSE 事件协议

首帧必须包含 conversation_id，否则客户端无法调用 /cancel 或 /approve：

event: status
data: {"conversation_id":"c_123","state":"started"}

event: token
data: {"conversation_id":"c_123","seq":1,"text":"可以"}

: ping

event: done
data: {"conversation_id":"c_123","seq":9,"usage":{"input":42,"output":18}}

把这套协议按时间画开，取消为什么必须依赖首帧就很清楚了：

Preview

首帧下发 ID，取消走的是另一条连接

服务端实现：

async def chat_stream(request: ChatRequest, provider: LLMProvider):
    conversation_id = request.conversation_id or secrets.token_urlsafe(12)
    yield sse('status', {'conversation_id': conversation_id, 'state': 'started'})
    seq = 0
    async for delta in provider.stream(build_messages(request)):
        if await cancellation_store.is_cancelled(conversation_id):
            yield sse('cancelled', {'conversation_id': conversation_id})
            return
        if delta.text:
            seq += 1
            yield sse('token', {'conversation_id': conversation_id, 'seq': seq, 'text': delta.text})
        yield ': ping\n\n'
    yield sse('done', {'conversation_id': conversation_id, 'seq': seq})

/cancel 是另一条 HTTP 连接，不能依赖请求上下文里的 asyncio.Event。单进程可以用内存字典，多进程必须把取消标记放 Redis，并设置过期时间避免泄漏。

上下文裁剪不是简单切片

把最近 20 条消息全部塞给模型，会让历史挤掉当前问题。用 token 预算明确分配：

Preview

四块预算，只有历史那块能裁

def fit_context(system: str, history: list[dict], question: str, limit: int = 8192):
    reserved_output = 1200
    budget = limit - estimate(system) - estimate(question) - reserved_output
    kept: list[dict] = []
    for message in reversed(history):
        cost = estimate(message['content'])
        if cost > budget:
            break
        kept.append(message)
        budget -= cost
    return [{'role': 'system', 'content': system}, *reversed(kept),
            {'role': 'user', 'content': question}]

真实项目中用 tiktoken 估算，供应商不支持时按字符数保守兜底，并在日志里记录裁剪前后的 token 数。

浏览器调试页和断线处理

最小客户端不要先上 React。下面这一页 HTML 直接存成 debug.html 用浏览器打开就能用，它是本篇最值回票价的东西——你能看见 token 一个个冒出来，而不是盯着 curl 猜协议对不对：

<!doctype html>
<meta charset="utf-8">
<title>SSE 调试页</title>
<style>
  body { font: 15px/1.7 system-ui, sans-serif; max-width: 720px; margin: 40px auto; }
  #out { white-space: pre-wrap; border: 1px solid #ccc; padding: 12px; min-height: 160px; }
  #log { color: #888; font-size: 13px; }
</style>
<input id="q" size="50" value="点桌面图标没反应，任务管理器里有进程">
<button id="send">发送</button>
<button id="stop" disabled>取消</button>
<div id="out"></div>
<div id="log"></div>

<script>
const $ = (id) => document.getElementById(id);
let conversationId = null;

function handle(event, data) {
  if (event === 'status') { conversationId = data.conversation_id; $('stop').disabled = false; }
  if (event === 'token')  { $('out').textContent += data.text; }
  if (event === 'done')   { $('log').textContent = `完成，用量 ${JSON.stringify(data.usage)}`; }
  if (event === 'cancelled') { $('log').textContent = '已取消'; }
}

$('send').onclick = async () => {
  $('out').textContent = ''; $('log').textContent = '';
  const res = await fetch('/chat', {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify({ message: $('q').value }),
  });
  const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
  let buffer = '';
  for (;;) {
    const { value, done } = await reader.read();
    if (done) break;
    buffer += value;
    const frames = buffer.split('\n\n');
    buffer = frames.pop();                       // 最后一段可能不完整，留到下一轮
    for (const frame of frames) {
      if (!frame.trim() || frame.startsWith(':')) continue;   // 跳过 : ping 心跳
      const event = frame.match(/^event: (.+)$/m)?.[1];
      const data = frame.match(/^data: (.+)$/m)?.[1];
      if (event && data) handle(event, JSON.parse(data));
    }
  }
  $('stop').disabled = true;
};

$('stop').onclick = () => fetch('/cancel', {
  method: 'POST',
  headers: { 'content-type': 'application/json' },
  body: JSON.stringify({ conversation_id: conversationId }),
});
</script>

有三个边界细节会原样保留到第 14 篇的 React 实现中：

buffer = frames.pop() 把最后一段残帧留到下一次读取。TCP 不保证一次 read() 正好落在帧边界上，少了这行就会随机丢字。
frame.startsWith(':') 跳过心跳。不跳过的话，JSON.parse 会在 : ping 上抛异常，整个流就断了。
取消按钮用的是 status 帧里存下来的 conversationId。这就是首帧必须先发的原因——按钮在第一个 token 出现之前就得可用。

断线重连时携带 Last-Event-ID 或自定义 last_seq，服务端从 Redis/数据库重放缺失事件；第 14 篇会把这段逻辑移入 React 状态机。

测试、故障排查与验收

uv run pytest tests/llm/test_mock_provider.py tests/api/test_chat_sse.py -q
curl -N -X POST http://localhost:8000/chat \
  -H 'content-type: application/json' -d '{"message":"查一下我的工单"}'

常见故障：

浏览器只收到最后一帧：检查响应头 Content-Type: text/event-stream、Cache-Control: no-cache，并关闭代理缓冲。
中文乱码：每个 chunk 用 TextDecoderStream 增量解码，不要对单帧调用 decode()。
include_usage 报 400：兼容服务不支持时捕获错误，关闭该字段并用 tokenizer 估算。
取消无效：确认取消连接和流连接使用同一个 conversation_id，跨进程时确认 Redis 通道一致。

本篇验收标准

Mock 模式不配置 API Key 也能输出 token、done 事件。
首帧在 1 秒内返回 conversation_id，每个 token 有递增 seq。
调用 /cancel 后最多再收到一个 token，最终收到 cancelled 而不是 done。
历史超过预算时保留 system、当前问题和完整消息对，不产生孤立的 tool 消息。

下一篇会让模型真正“做事”：工具调用循环只接入只读工具，先解决 schema、伪造结果和无限循环。

#Provider 抽象与可取消的流式对话

#先看一次失败的请求

#Provider 是稳定边界

#MockProvider 不是玩具

#SSE 事件协议

#上下文裁剪不是简单切片

#浏览器调试页和断线处理

#测试、故障排查与验收

#本篇验收标准

Provider 抽象与可取消的流式对话

先看一次失败的请求

Provider 是稳定边界

MockProvider 不是玩具

SSE 事件协议

上下文裁剪不是简单切片

浏览器调试页和断线处理

测试、故障排查与验收

本篇验收标准