流式响应处理 - AI 应用开发实战

← 返回课程AI 应用开发实战 › 第 5 关

进度 0%

第 5 关：流式响应处理

Python高级

📖 本课目标

本关属于《AI 应用开发实战》第 5 课。阅读右侧代码模板，理解每一行的作用，然后动手修改并完成练习。

🎯 本课要点

流式响应处理
模拟 SSE (Server-Sent Events) 流式输出

🛠 学习方式

阅读右侧代码模板中的注释与实现
修改代码，尝试达成新的效果
点击「运行」查看实际输出
确认理解后点击「提交」标记本关完成

💡 说明

本关为 Python，浏览器内无法真正执行该语言。当前「运行」输出为演示结果，请在本地环境中编译运行以验证代码。

main.pytest.py

# 流式响应处理
# 模拟 SSE (Server-Sent Events) 流式输出

import time

class StreamingResponse:
    """模拟流式 AI 响应"""
    
    def __init__(self, full_text):
        self.full_text = full_text
        self.position = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.position >= len(self.full_text):
            raise StopIteration
        # 每次返回 1-3 个字符（模拟 token 流）
        chunk_size = min(3, len(self.full_text) - self.position)
        chunk = self.full_text[self.position:self.position + chunk_size]
        self.position += chunk_size
        return {"content": chunk, "done": self.position >= len(self.full_text)}

def stream_chat(message):
    """模拟流式对话"""
    print(f"用户: {message}")
    print("AI: ", end="", flush=True)
    
    # 模拟 AI 生成的完整回复
    response = "Python 是一门优秀的编程语言。建议从基础语法开始，然后学习数据结构和算法，最后做实际项目来巩固知识。"
    
    stream = StreamingResponse(response)
    full_response = ""
    for chunk in stream:
        full_response += chunk["content"]
        print(chunk["content"], end="", flush=True)
    
    print()  # 换行
    return full_response

# 流式输出演示
print("=== 流式响应演示 ===\n")
stream_chat("如何学习Python？")

# SSE 格式说明
print("\n--- SSE (Server-Sent Events) 格式 ---")
sse_example = """data: {"content": "你", "done": false}

data: {"content": "好", "done": false}

data: {"content": "！", "done": false}

data: {"content": "", "done": true}
"""
print(sse_example)

# 前端处理代码
print("--- 前端处理 (fetch + ReadableStream) ---")
frontend_code = """
const response = await fetch('/api/chat', { method: 'POST', body: ... });
const reader = response.body.getReader();
while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const text = new TextDecoder().decode(value);
    // 解析 SSE 数据并更新 UI
    displayText += parseSSE(text);
}
"""
print(frontend_code)

点击「运行」或「提交代码」查看结果

📖 查看本课说明

第 5 关：流式响应处理

Python · 高级 · 《AI 应用开发实战》

流式响应处理
模拟 SSE (Server-Sent Events) 流式输出

本关为 Python，浏览器无法真正编译运行，请在本地环境验证代码。