AI模型服务按输入输出流式传输

AI模型服务做流式传输时，先分别约定输入与输出。常见文本问答可以一次POST完整JSON问题，响应端再用事件流逐段返回答案；这不要求客户端把输入逐字上传，也不等于双向实时会话。音频连续输入、实时打断等场景需要另外设计会话协议。

如果原有客户端一直等待一个完整JSON答案，我会先迁移接口契约，再调整显示组件。仅把服务端返回值换成生成器，客户端仍调用json()或等待完整响应，页面往往还是最后一次性显示。下面给出整包响应迁移到POST加SSE格式响应的原创适配方案，不假设任何厂商版本已经替你完成这些变更。

Starlette官方响应文档：https://starlette.dev/responses/

HTTPX官方异步文档：https://www.python-httpx.org/async/

确定输入输出的迁移范围

本文保留完整JSON输入，只改变响应消费方式。调用方先提交prompt；服务端在开始响应前完成授权、参数检查、请求体大小限制与模型容量分配。示例展示prompt字段检查，其余入口控制应接入现有服务中间件。4000字符只是演示限制，不能取代请求体字节限制或模型token上限。

如果要连续上传输入，接口还需要规定输入分帧、结束标记、会话归属和中断规则。响应流与请求流是两个方向；不能因为HTTPX有stream方法，就认为json参数也变成流式上传。本文不改变模型推理算法，而是在既有模型文本迭代器之外加协议层。

定义事件格式和兼容边界

本例定义delta、done和error三个应用事件名。delta载荷只有新增text，不能每次都发送累计全文后让客户端再追加一次；done说明本次文本迭代正常结束；error说明答案不完整。事件名是本项目契约，不是所有模型服务统一采用的标准名称。

SSE格式使用UTF-8文本、event字段和data字段，空行分隔事件。正文中的换行由JSON编码转义，不直接插进协议字段。一个网络数据块可能切开半个事件，也可能包含多个事件，因此不能把每次网络读取结果直接当一个完整JSON对象解析。

浏览器原生EventSource的通常入口是URL驱动的接收方式。本例需要POST正文，客户端采用能发送POST并读取响应流的HTTP客户端；浏览器可用fetch配合事件解析器。不要把原生EventSource的自动重连行为直接等同于POST生成请求可安全重放。

实现模型服务的流式响应适配

以下是Starlette路由处理函数片段，需要以POST方法绑定到项目路由，例如/generate-stream。Python需支持asyncio.timeout。应用启动时把已有的模型流适配器设置到app.state.model_stream；该适配器接收问题，返回异步文本生成器，并在aclose或取消时释放请求资源。这里不包含模型下载、服务启动或真实推理输出。

import asyncio
import json
from contextlib import aclosing
from starlette.responses import JSONResponse, StreamingResponse

def frame(kind, payload):
    # 事件名来自服务端常量；JSON会转义正文中的换行。
    data = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
    return f"event: {kind}\ndata: {data}\n\n"

async def generate_stream(request):
    try:
        body = await request.json()
    except ValueError:
        return JSONResponse({"error": "invalid_json"}, status_code=400)
    prompt = body.get("prompt") if isinstance(body, dict) else None
    if not isinstance(prompt, str) or not prompt.strip() or len(prompt) > 4000:
        return JSONResponse({"error": "invalid_prompt"}, status_code=400)

    async def events():
        try:
            # 项目注入的适配器必须返回可aclose的异步文本生成器。
            async with aclosing(request.app.state.model_stream(prompt)) as stream:
                async with asyncio.timeout(60):
                    async for piece in stream:
                        if not isinstance(piece, str):
                            raise TypeError("模型适配器必须输出文本")
                        if piece:
                            yield frame("delta", {"text": piece})
            yield frame("done", {"finish": "complete"})
        except asyncio.CancelledError:
            # 断连取消要继续向上传递，不把它转换成成功。
            raise
        except Exception:
            # 流已开始后用协议内事件报错，不暴露模型路径或堆栈。
            yield frame("error", {"code": "generation_failed"})

    return StreamingResponse(
        events(), media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"},
    )

模型适配器必须真正异步等待或主动让出执行权。把同步长时间计算直接塞进async函数，仍会阻塞事件循环；这种情况下asyncio.timeout也不能强制打断底层计算。应使用合适的推理服务或工作进程，把截止、取消与容量控制落实到计算执行层。

返回StreamingResponse后，响应头可能已经发送。此后模型失败，不能再把当前200响应改写成普通500 JSON，所以协议保留error事件。连接已经断开时，失败事件也未必能送达；客户端仍要处理网络异常和缺失done。这个示例只给出统一generation_failed，不泄露内部堆栈，具体错误原因可用请求标识在服务日志中关联。

按事件读取而非按网络块拼JSON

客户端进入HTTPX的stream上下文，先判断状态码与Content-Type，再按行积攒事件数据，遇到空行解析。下面只实现本接口使用的event和data子集，忽略注释行；它不实现事件ID恢复、自动重试或完整通用SSE库。

import json
import httpx

async def read_answer(client, url, prompt, emit):
    # client由应用复用；read超时是等下一段数据的时间，不是总截止时间。
    timeout = httpx.Timeout(65.0, connect=5.0)
    async with client.stream(
        "POST", url, json={"prompt": prompt},
        headers={"Accept": "text/event-stream"}, timeout=timeout,
    ) as response:
        response.raise_for_status()
        if response.headers.get("content-type", "").split(";")[0] != "text/event-stream":
            raise RuntimeError("服务未返回约定事件流")
        kind, data, chunks = "message", [], []
        async for line in response.aiter_lines():
            if line == "":
                if data:
                    obj = json.loads("\n".join(data))
                    if kind == "delta":
                        text = obj["text"]
                        if not isinstance(text, str):
                            raise ValueError("delta.text必须是字符串")
                        emit(text)
                        chunks.append(text)
                    elif kind == "done":
                        return "".join(chunks)
                    elif kind == "error":
                        raise RuntimeError("模型服务返回失败事件")
                kind, data = "message", []
                continue
            if line.startswith(":"):
                continue
            field, sep, value = line.partition(":")
            if sep and value.startswith(" "):
                value = value[1:]
            if field == "event":
                kind = value
            elif field == "data":
                data.append(value)
        # TCP结束不是业务完成；缺少done时保留为部分结果。
        raise RuntimeError("事件流提前结束，未收到done")

emit是同步展示回调，调用方应确保它不会长时间阻塞。read_answer接收应用复用的AsyncClient，不在每个片段里重建连接池。离开stream上下文会关闭响应；调用方在自己的生命周期结束时关闭client。若要限制完整调用总时长，还应在上层增加总截止，不能把HTTPX的read等待超时当成总时长。

这个读取器直到done才返回完整答案。网络提前结束时抛错，即使之前已经通过emit显示了部分文字，也不会把它当作完成结果。正式接入时还应对事件长度、累计输出量及未知事件制定限制，避免无界缓存；示例中的chunks仅适合已经限制输出长度的文本问答。

处理断连回退与迁移检查

我会把兼容迁移分成两个入口：保留旧JSON接口，新客户端使用独立流式路径。这样旧客户端不会突然把事件正文交给JSON解析器。回退时让新客户端选择旧路径，并恢复整包读取方式；不要在已经输出半段答案后自动再请求一次，造成重复文本和重复模型费用。

代理缓冲、压缩组件、发送端聚合和客户端积攒都会影响可见的首段时间。Cache-Control: no-cache不是让所有代理立即刷新数据的统一开关。需要按实际代理配置流式路径，并分别观察模型首段就绪、服务首次发送和客户端首次显示，不能只看服务端已经yield。

输入和输出都要流式吗？不必。短文本问答通常完整提交问题即可；连续语音或可打断会话才需要考虑双向通信与输入帧。若需要客户端不断向服务端发事件，应评估WebSocket等适合会话交互的协议。

收到HTTP 200就能记成功吗？只能说明响应已经开始。本文的成功还要求收到done，未收到done的网络结束、error事件和客户端取消都应单列。done只代表本接口定义的正常生成结束，不是模型答案事实正确的保证。

断开后能从上次位置继续吗？本示例不能。恢复需要服务端保留事件序号与可重放内容，还要明确模型请求是继续原任务还是重新生成。仅发送id字段不会自动实现这些能力。

迁移前应覆盖哪些场景？包括中文换行和引号、空片段、多事件合并传输、事件跨块、流前参数错误、流中模型失败、连接中断和慢消费。这里给出的是读者工程接入时的检查清单，本文未运行这些场景，也不编造延迟收益。

SSE格式参考：https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events