如何停止Ollama模型流式输出？解决流无法终止问题

问题背景 #

实现了用于流式获取模型响应的LLMAssistant类，搭配FastAPI的EventSourceResponse接口提供流式接口，但前端监听时发现模型输出完成后会收到空消息，流无法终止并出现重启情况。

原LLMAssistant类代码 #

from langchain_community.llms.ollama import Ollama
from app.config import (
    LLM_MODEL_NAME,
    MAX_LLM_INPUT_LENGTH,
    MAX_LLM_INPUT_SENTENCES,
    LLAMA_BASE_URL,
)
from app.utils.text_processing import TextProcessing
from app.utils.persona_mapper import PersonaPromptMapper
from app.enums import Personas


class LLMAssistant:
    def __init__(
        self, persona: Personas = Personas.PROFESSOR, model_name: str = LLM_MODEL_NAME
    ):
        self.persona = persona
        self.persona_prompt = PersonaPromptMapper.get_persona_prompt(persona)
        self.model = Ollama(name=model_name, base_url=LLAMA_BASE_URL)

    def process(self, question: str):
        processed_question = TextProcessing.truncate_text(
            question,
            max_words=MAX_LLM_INPUT_LENGTH,
            max_sentences=MAX_LLM_INPUT_SENTENCES,
        )
        input_text = f"Persona:'{self.persona_prompt}',Prompt: '{processed_question}'"

        accumulated_text = ""  # Initialize accumulated text

        for chunk in self.model.stream(input_text):
            # Append the chunk to accumulated text
            accumulated_text += chunk

            # Check if we have formed complete sentences
            sentences = accumulated_text.split(". ")  # Split by sentences

            # Join sentences with proper punctuation
            complete_sentences = []
            for i in range(len(sentences) - 1):
                if i == 0:
                    complete_sentences.append(sentences[i] + ". ")  # Include the period
                else:
                    complete_sentences.append(sentences[i] + ". ")  # Include the period

            # Send the current state of accumulated text (up to complete sentences)
            yield "".join(complete_sentences)

        # After finishing all chunks, yield any remaining accumulated text
        yield accumulated_text


DefaultAssistant = LLMAssistant()

原FastAPI流式接口代码 #

@router.get("/assistant/ask/stream")
async def ask_stream(question: str = Query(...)):
    try:
        # Process the text using LLMAssistant
        return EventSourceResponse(DefaultAssistant.process(question))

    except Exception as e:
        print(e)
        raise HTTPException(status_code=500, detail=f"Failed to process text: {str(e)}")

问题原因 #

1. 重复发送内容：原process方法每次循环都会yield所有已完成的句子，导致前端重复接收内容，最后还会再次yield全部文本，可能产生空消息。
2. 缺少结束标记：没有明确的流结束信号，前端无法判断输出是否完成，进而触发重连逻辑。
3. 同步生成器适配问题：直接将同步生成器传给EventSourceResponse，可能导致处理异常，产生空输出。

解决方案 #

1. 修正LLMAssistant的process方法 #

修改为只发送新增的完整句子，避免重复，并确保最后只发送非空剩余文本：

from langchain_community.llms.ollama import Ollama
from app.config import (
    LLM_MODEL_NAME,
    MAX_LLM_INPUT_LENGTH,
    MAX_LLM_INPUT_SENTENCES,
    LLAMA_BASE_URL,
)
from app.utils.text_processing import TextProcessing
from app.utils.persona_mapper import PersonaPromptMapper
from app.enums import Personas


class LLMAssistant:
    def __init__(
        self, persona: Personas = Personas.PROFESSOR, model_name: str = LLM_MODEL_NAME
    ):
        self.persona = persona
        self.persona_prompt = PersonaPromptMapper.get_persona_prompt(persona)
        self.model = Ollama(name=model_name, base_url=LLAMA_BASE_URL)

    def process(self, question: str):
        processed_question = TextProcessing.truncate_text(
            question,
            max_words=MAX_LLM_INPUT_LENGTH,
            max_sentences=MAX_LLM_INPUT_SENTENCES,
        )
        input_text = f"Persona:'{self.persona_prompt}',Prompt: '{processed_question}'"

        accumulated_text = ""
        last_sentence_end = 0  # 记录上次已发送到的位置

        for chunk in self.model.stream(input_text):
            accumulated_text += chunk
            # 循环提取所有新增的完整句子
            while True:
                # 查找下一个完整句子的结束位置（包含". "）
                sentence_end = accumulated_text.find(". ", last_sentence_end)
                if sentence_end == -1:
                    break
                # 提取并发送新增的完整句子
                new_content = accumulated_text[last_sentence_end:sentence_end+2]
                yield new_content
                last_sentence_end = sentence_end + 2

        # 发送剩余的非空文本（未完成的句子）
        remaining_text = accumulated_text[last_sentence_end:].strip()
        if remaining_text:
            yield remaining_text


DefaultAssistant = LLMAssistant()

2. 修正FastAPI接口 #

将同步生成器转为异步生成器，按SSE标准格式发送消息，并添加结束标记：

from fastapi import APIRouter, Query, HTTPException
from fastapi.responses import EventSourceResponse

router = APIRouter()

@router.get("/assistant/ask/stream")
async def ask_stream(question: str = Query(...)):
    async def async_sse_generator():
        # 包装同步生成器为异步，过滤空内容
        for chunk in DefaultAssistant.process(question):
            if chunk.strip():
                yield f"data: {chunk}\n\n"
        # 发送明确的结束标记，通知前端关闭流
        yield "data: [DONE]\n\n"

    try:
        return EventSourceResponse(async_sse_generator())
    except Exception as e:
        print(e)
        raise HTTPException(status_code=500, detail=f"Failed to process text: {str(e)}")

3. 前端处理示例 #

监听流时识别结束标记并关闭连接：

const question = "你的问题内容";
const eventSource = new EventSource(`/assistant/ask/stream?question=${encodeURIComponent(question)}`);

eventSource.onmessage = function(event) {
    if (event.data === "[DONE]") {
        eventSource.close();
        return;
    }
    // 处理收到的流式文本
    console.log("收到内容:", event.data);
};

eventSource.onerror = function(error) {
    console.error("流错误:", error);
    eventSource.close();
};

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