在Django Channels中实现类ChatGPT的流式WebSocket消息发送

你报错的核心是用错了对象：StreamingHttpResponse是Django用于HTTP请求的流式响应对象，不能直接传给Websocket的send()方法。Websocket的send()只接受字符串类型的文本数据，而你把整个HTTP响应对象传进去，导致底层尝试调用它的encode()方法时失败。

在AsyncWebsocketConsumer里实现流式消息推送，不需要用StreamingHttpResponse，直接通过循环逐个发送分片消息即可。如果是调用异步生成的内容（比如调用AI接口的流式返回），用async for遍历；如果是同步生成的内容，直接循环加await发送即可。

示例代码：

import json
import asyncio
from channels.generic.websocket import AsyncWebsocketConsumer

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.accept()

    async def receive(self, text_data):
        message = json.loads(text_data)
        print('Received message:', message)

        # 模拟ChatGPT式的流式返回（异步生成器示例）
        async def async_data_stream():
            for i in range(10):
                await asyncio.sleep(0.5)  # 模拟AI思考/返回延迟
                yield json.dumps({'message': f'Message {i}'})

        # 逐个发送流式消息
        async for chunk in async_data_stream():
            await self.send(text_data=chunk)

如果你的流式数据来自同步生成器（比如本地处理的内容），可以直接循环发送：

def sync_data_stream():
    for i in range(10):
        yield json.dumps({'message': f'Message {i}'})

# 在receive方法里
for chunk in sync_data_stream():
    await asyncio.sleep(0.5)
    await self.send(text_data=chunk)

非常推荐。类ChatGPT的聊天应用核心就是实时增量返回内容，Websocket本身就是为双向实时通信设计的，相比HTTP流式响应：

• 支持双向交互（用户可以在接收消息时继续发新请求）
• 连接状态保持更稳定，适合长时间对话场景
• 前端处理分片消息更灵活，能实时渲染打字效果

内容的提问来源于stack exchange，提问作者Tamerlan Aliyev