【每日一题】解析大模型 SSE 流式响应（Python 编程 · 附参考答案）

兴趣爱好交流
1

题目：AI 测试里经常要断言大模型的流式返回。请实现 parse_sse_stream(raw)：输入一段 OpenAI 兼容的 SSE（Server-Sent Events）原始文本，返回拼接后的 assistant 完整内容。

规则：

  1. 只处理以 data: 开头的行（data: 后固定有一个空格）。
  2. 遇到 data: [DONE] 立即停止解析，其后的数据全部忽略。
  3. 忽略空行、注释行（以 : 开头）、以及非 data 的事件字段（event:、id:、retry:）。
  4. 每个 JSON chunk 取 choices[0].delta.content 拼接；若 delta 里没有 content（如仅有 role）或 content 为 null，跳过。
  5. 单行 JSON 解析失败时跳过该行，不整体报错。

代码框架：

import json

def parse_sse_stream(raw: str) -> str:
    """解析 OpenAI 兼容 SSE 流式响应文本，返回拼接后的 assistant 内容。

    示例输入：
        data: {"choices":[{"delta":{"content":"你"}}]}
        data: {"choices":[{"delta":{"content":"好"}}]}
        data: [DONE]
    返回："你好"
    """
    pass

测试用例（你的实现需让以下断言全部通过）：

assert parse_sse_stream(
    'data: {"choices":[{"delta":{"content":"你"}}]}\n'
    'data: {"choices":[{"delta":{"content":"好"}}]}\n'
    'data: [DONE]\n') == "你好"

# 忽略注释/空行/非 data 事件/role-only delta
assert parse_sse_stream(
    ': ping\n'
    '\n'
    'data: {"choices":[{"delta":{"role":"assistant"}}]}\n'
    'data: {"choices":[{"delta":{"content":"A"}}]}\n'
    'event: message\n'
    'id: 42\n'
    'data: {"choices":[{"delta":{"content":"B"}}]}\n'
    'data: [DONE]\n') == "AB"

# [DONE] 之后忽略
assert parse_sse_stream(
    'data: {"choices":[{"delta":{"content":"X"}}]}\n'
    'data: [DONE]\n'
    'data: {"choices":[{"delta":{"content":"Y"}}]}\n') == "X"

# content 为 null 跳过
assert parse_sse_stream(
    'data: {"choices":[{"delta":{"content":null}}]}\n'
    'data: {"choices":[{"delta":{"content":"!"}}]}\n'
    'data: [DONE]\n') == "!"

# 空输入 / 仅 DONE
assert parse_sse_stream("") == ""
assert parse_sse_stream("data: [DONE]\n") == ""

# 非法 JSON 行不炸
assert parse_sse_stream(
    'data: not-a-json\n'
    'data: {"choices":[{"delta":{"content":"z"}}]}\n'
    'data: [DONE]\n') == "z"

难度：中级
题型：笔试（编程题）

:point_right:参考答案/答题要点：

思路一：常规逐行处理（面试首选，清晰稳）

import json

def parse_sse_stream(raw: str) -> str:
    parts = []
    for line in raw.splitlines():
        if not line.startswith("data: "):
            continue
        payload = line[len("data: "):].strip()
        if payload == "[DONE]":
            break
        try:
            obj = json.loads(payload)
        except json.JSONDecodeError:
            continue
        choices = obj.get("choices") or []
        if not choices:
            continue
        delta = choices[0].get("delta") or {}
        content = delta.get("content")
        if isinstance(content, str):
            parts.append(content)
    return "".join(parts)

思路二：生成器分离"事件流"+“提取”，主逻辑一行（体现 Python 功底）

import json
from typing import Iterable

def _iter_sse_data(raw: str) -> Iterable[dict]:
    for line in raw.splitlines():
        if not line.startswith("data: "):
            continue
        payload = line[6:].strip()
        if payload == "[DONE]":
            return
        try:
            yield json.loads(payload)
        except json.JSONDecodeError:
            continue

def parse_sse_stream(raw: str) -> str:
    return "".join(
        c
        for obj in _iter_sse_data(raw)
        for ch in (obj.get("choices") or [])
        if isinstance((c := (ch.get("delta") or {}).get("content")), str)
    )

（以上题解代码本机 Python 3 实跑，8 条断言全部 ALL PASS。）

采分点：

  • 正确处理 data: 前缀过滤与 [DONE] 提前终止（break/return），不误解析其后数据。
  • 用 dict.get + or []/or {} 做防御式取值，delta 无 content 或 content=null 不报错。
  • 单行 json.loads 失败用 try/except 跳过，不整体崩溃。
  • 判断用 isinstance(content, str) 而非 if content（避免空串与 null 语义混淆）。
  • 优雅解：生成器解耦解析与提取、walrus(:=)、splitlines 兼容末行无换行。
  • 加分：主动提"真流式"场景应逐 chunk read 而非一次性 split，以及中文按 code point 拼接不乱码。