基于HuggingFace的微调LLM在Django中实现流式输出的问题咨询

你当前使用的TextStreamer默认将内容输出到终端，要实现向Django用户流式传输文本，需要自定义Streamer捕获输出，再通过Django的流式响应返回给前端。

步骤1：自定义Streamer类 #

继承TextStreamer，重写输出逻辑，把生成的token文本存入队列，而非打印到终端：

from transformers import TextStreamer
import queue

class DjangoTextStreamer(TextStreamer):
    def __init__(self, tokenizer, skip_prompt=True, **decode_kwargs):
        super().__init__(tokenizer, skip_prompt=skip_prompt, **decode_kwargs)
        self.output_queue = queue.Queue()
        self.stop_signal = object()  # 自定义生成结束标记

    def put(self, value):
        # 解码token为可读文本
        text = self.tokenizer.decode(value, **self.decode_kwargs)
        # 跳过提示词部分（开启skip_prompt时生效）
        if self.skip_prompt and self.next_tokens_are_prompt:
            self.next_tokens_are_prompt = False
            return
        # 将文本块存入队列
        self.output_queue.put(text)

    def end(self):
        # 生成结束后发送终止信号
        self.output_queue.put(self.stop_signal)

步骤2：编写Django流式视图 #

使用StreamingHttpResponse配合生成器，从队列中逐块取出内容返回给前端。用线程后台处理模型生成，避免阻塞请求：

from django.http import StreamingHttpResponse, HttpResponseBadRequest
from django.views.decorators.http import require_POST
import threading

# 建议将model、tokenizer、pipeline全局初始化（放在Django启动时执行，比如apps.py的ready方法）
# 不要每次请求都重新加载模型，否则性能极差

@require_POST
def stream_llm(request):
    prompt = request.POST.get('prompt', '')
    if not prompt:
        return HttpResponseBadRequest("缺少提示词")

    # 初始化自定义流式处理器
    streamer = DjangoTextStreamer(tokenizer)

    # 后台执行模型生成任务
    def generate_task():
        pipeline(
            prompt,
            max_new_tokens=300,
            streamer=streamer,
            do_sample=True,
            temperature=0.8,
            pad_token_id=tokenizer.eos_token_id
        )
        streamer.end()  # 生成完成后发送终止信号

    # 启动守护线程执行生成
    threading.Thread(target=generate_task, daemon=True).start()

    # 生成器：从队列取内容并返回
    def content_generator():
        while True:
            chunk = streamer.output_queue.get()
            if chunk is streamer.stop_signal:
                break
            # 用Server-Sent Events格式，前端易解析
            yield f"data: {chunk}\n\n"

    # 返回流式响应，设置正确的Content-Type
    return StreamingHttpResponse(
        content_generator(),
        content_type="text/event-stream"
    )

步骤3：前端接收流式内容 #

用Fetch API读取响应流，实时更新页面内容：

<div>
    <input type="text" id="prompt-input" placeholder="输入你的问题...">
    <button id="send-btn">发送</button>
    <div id="response-area"></div>
</div>

<script>
const sendBtn = document.getElementById('send-btn');
const promptInput = document.getElementById('prompt-input');
const responseArea = document.getElementById('response-area');

sendBtn.addEventListener('click', async () => {
    const prompt = promptInput.value.trim();
    if (!prompt) return;

    responseArea.textContent = '';

    // 发送POST请求获取流式响应
    const response = await fetch('/stream-llm/', {
        method: 'POST',
        headers: {
            'Content-Type': 'application/x-www-form-urlencoded',
        },
        body: `prompt=${encodeURIComponent(prompt)}`
    });

    // 读取流内容并渲染
    const reader = response.body.getReader();
    const decoder = new TextDecoder();

    while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        // 解析SSE格式的内容块
        const chunks = decoder.decode(value).split('\n\n');
        chunks.forEach(chunk => {
            if (chunk.startsWith('data: ')) {
                responseArea.textContent += chunk.slice(6);
            }
        });
    }
});
</script>

关键注意事项 #

• 全局初始化模型：将模型加载逻辑放在Django启动阶段，避免重复加载浪费资源。
• 异步优化：生产环境建议用Django异步视图（async def）配合asyncio替代线程，降低资源占用。
• 错误处理：添加队列超时、模型异常捕获逻辑，防止请求无限挂起。
• 跨域配置：若前后端跨域，需在Django中配置CORS允许SSE请求。

内容的提问来源于stack exchange，提问作者John mick