让AI对话像流水般自然：深入大模型Streaming技术核心源码

HTTP/1.1 200 OK
Transfer-Encoding: chunked
1a
{"data": "This is the first chunk"}
23
{"data": "and this is the second"}
0

# 伪代码：流式生成的核心循环
def generate_stream():
tokens = []
while not is_generation_complete():
next_token = model.predict(tokens) # 预测下一个token
tokens.append(next_token)
yield format_token(next_token) # 转换为可传输格式
flush_buffer() # 立即发送网络缓冲区

2.1 典型架构分层+-----------------+
| Client (前端/APP) |
+--------+--------+
| SSE/WebSocket
+--------v--------+
| 流式代理网关 |
|（负载均衡/鉴权） |
+--------+--------+
| gRPC流
+--------v--------+
| 模型推理服务 |
|（分批生成token） |
+-----------------+

def sse_format(data):
return f"data: {json.dumps(data)}\n\n"

sequenceDiagram
Client->>Gateway: 发起SSE连接
Gateway->>Model: 创建gRPC流
Model->>Gateway: 持续返回token块
Gateway->>Client: 实时转发数据
alt 网络中断
Client->>Gateway: 自动重连(含last_id)
Gateway->>Model: 恢复生成位置
end

from langchain.callbacks import StreamingStdOutCallbackHandler
chat = ChatOpenAI(
streaming=True,
callbacks=[StreamingStdOutCallbackHandler()]
)
chat.predict("解释量子力学") # 结果实时打印

from sse_starlette import EventSourceResponse
@app.get('/stream')
async def message_stream():
async def generator():
while True:
yield sse_format(await get_next_token())
return EventSourceResponse(generator())

// 分块传输方案
{"state": "partial", "data": "片段1"}
{"state": "partial", "data": "片段2"}
{"state": "complete", "data": "完整JSON"}

# 混合传输文本与图片token
def multi_modal_stream():
if content_type == "text":
yield text_token
else:
yield image_fragment_base64