Streaming AI answers in Laravel: the hard parts#

Getting the first token of an AI answer onto the screen takes an afternoon. Everything after that took me months. People reload the page mid-answer. They press Stop. Their phone drops the connection and the app retries. nginx buffers the whole thing. The model starts writing <think> tags. And one busy chat quietly eats every PHP-FPM worker you have, so the dashboard stops loading for everyone else.

This is how I handle all of that in Chatvia, with the actual code. Chatvia is a self-hosted AI agent platform I built on Laravel 13, Inertia and Vue 3. No Node sidecar, no streaming microservice. One HTTP request runs the whole turn: guardrails, retrieval, the model, its tool calls, and the save. All of it is plain PHP, and it works better than I expected it to.

Alright, let's start with what goes over the wire.

Don't stream text. Stream events.#

My first version streamed raw text. It worked for about a week. Then I needed to send sources, then tool activity, then errors, and I ended up with a mess of magic prefixes that the frontend had to regex apart. So I threw it away.

Now a reply is a sequence of typed events, one JSON line each:

enum EventType: string
{
    case Start = 'start';
    case Delta = 'delta';
    case Reasoning = 'reasoning';
    case Meta = 'meta';
    case Sources = 'sources';
    case Tool = 'tool';
    case Replace = 'replace';
    case Error = 'error';
    case Done = 'done';
}

A small value object builds them. It knows which assistant message it belongs to and, once sent, its sequence number:

final readonly class StreamEvent
{
    public function __construct(
        public EventType $type,
        public string $id,
        public array $data = [],
        public ?int $seq = null,
    ) {}

    public static function delta(string $assistantMessageId, string $text): self
    {
        return new self(EventType::Delta, $assistantMessageId, ['text' => $text]);
    }

    public static function error(string $assistantMessageId, ErrorCode $code, string $message): self
    {
        return new self(EventType::Error, $assistantMessageId, [
            'code' => $code->value,
            'message' => $message,
            'retryable' => $code->isRetryable(),
        ]);
    }

    public function toLine(): string
    {
        $payload = [
            'v' => 1,
            'seq' => $this->seq ?? 0,
            'id' => $this->id,
            'type' => $this->type->value,
            'data' => (object) $this->data,
        ];

        return 'data: '.json_encode(
            $payload,
            JSON_THROW_ON_ERROR | JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES
        )."\n\n";
    }
}

So a turn looks like this in the network tab:

data: {"v":1,"seq":0,"id":"msg_01J…","type":"start","data":{"assistant_message_id":"msg_01J…","request_id":"req_…","language":"en","replayed":false}}

data: {"v":1,"seq":1,"id":"msg_01J…","type":"delta","data":{"text":"Your order "}}

data: {"v":1,"seq":2,"id":"msg_01J…","type":"delta","data":{"text":"ships on Monday."}}

data: {"v":1,"seq":3,"id":"msg_01J…","type":"sources","data":{"sources":[…]}}

data: {"v":1,"seq":4,"id":"msg_01J…","type":"done","data":{"reason":"complete","response_time_ms":1840,"finished_at":"2026-10-29T10:12:04+00:00"}}

Errors are events too, which sounds obvious but matters: the browser never has to guess whether a chunk of text is the answer or a failure. Two of these events do the heavy lifting later. replace tells the browser "throw away what you rendered, here is the corrected text", which is how output guardrails can redact an answer after it already streamed. tool carries a tool call's life, including the pause when a human has to approve it.

Generators, all the way down#

The domain layer has no idea HTTP exists. A chat turn is a PHP Generator that yields StreamEvents, and only the controller knows those events will become server-sent events.

readonly class ChatStreamController
{
    public function __invoke(
        ChatStreamRequest $request,
        string $conversationId,
        #[HostedChatAgent] Agent $agent,
    ): StreamedResponse {
        $data = $this->factory->forChat($request, $agent, $this->session->findOrStart($agent), $conversationId);

        return $this->sse->respondUntilFinished($this->runningTurns->track(
            $conversationId,
            fn (): Generator => $this->chatTurnStreamer->stream($data),
        ));
    }
}

ChatTurnStreamer::stream() runs every check that can refuse a turn before the response even starts: a deactivated workspace, a blocked visitor, the monthly message limit, the rate limits. Each refusal is itself a tiny generator that yields one error and one done. If the turn is allowed, it picks a source: an exact Q&A match, a guardrail refusal, the PDF model for attached files, or the agent's own model with its tools.

I did not plan this part, but it turned out to be the best decision in the whole thing. Because a turn is just a generator, the public API's JSON mode drains the same generator and folds it into one payload. Same gates, same guardrails, same save, zero duplicated logic:

class StreamedTurnCollector
{
    public function collect(iterable $events): CompletedTurnData
    {
        set_time_limit(0);

        $content = '';

        foreach ($events as $event) {
            switch ($event->type) {
                case EventType::Delta:
                    $content .= $event->data['text'] ?? '';
                    break;
                case EventType::Replace:
                    $content = $event->data['text'] ?? '';
                    break;
                // start, sources, error and done fill the other fields…
            }
        }

        return new CompletedTurnData(content: $content, /* … */);
    }
}

One class owns the transport#

SseResponder is the only file in the app that knows what text/event-stream is. Headers, output buffers, flushing, sequence numbers, and noticing that the browser is gone. Here is the part that matters:

class SseResponder
{
    public function respond(iterable $events): StreamedResponse
    {
        return new StreamedResponse(fn () => $this->pump($events), 200, $this->headers());
    }

    private function prepare(): void
    {
        // Long tool chains run past the PHP-FPM 30s default.
        if (function_exists('set_time_limit')) {
            @set_time_limit(0);
        }

        $this->keepRunningAfterDisconnect();

        while (ob_get_level() > 0) {
            ob_end_clean();
        }
    }

    private function emit(StreamEvent $event, int $seq): bool
    {
        if ($this->clientGone()) {
            return false;
        }

        echo $event->withSeq($seq)->toLine();
        flush();

        return ! $this->clientGone();
    }

    protected function clientGone(): bool
    {
        return connection_status() !== CONNECTION_NORMAL || connection_aborted();
    }

    private function headers(): array
    {
        return [
            'Content-Type' => 'text/event-stream',
            'Cache-Control' => 'no-cache, no-store, must-revalidate',
            'Connection' => 'keep-alive',
            'X-Accel-Buffering' => 'no',
        ];
    }
}

Three lines in there each cost me real debugging time, so let me save you some:

• That while (ob_get_level() > 0) loop. Laravel, Blade and a couple of extensions leave output buffers open. Close one level and flush() still does nothing. The whole answer shows up at once when the script ends, and you sit there wondering why your stream is not streaming.
• X-Accel-Buffering: no. nginx buffers FastCGI responses by default. Everything works beautifully on php artisan serve, then you deploy and it is silently broken. This header tells nginx to pass each write straight through.
• Checking the connection twice. emit() checks before the write and again after the flush. The flush is where PHP actually finds out the socket is gone, so the second check is the one that catches a closed tab on the very next event.

The tab closes. Now what?#

By default PHP kills your script the moment a write to a closed connection fails. For a chat, that is the worst possible outcome. You already paid for the model. The tools already ran. Maybe an image was generated. And none of it gets saved, so when the visitor reloads, the app asks the same question again and pays again.

So SseResponder calls ignore_user_abort(true) to keep control, and then one of two things happens depending on the surface.

For the chat widget and the API, the generator is told that nobody is listening. A generator can receive an exception at the exact point where it is paused. That is what Generator::throw() does, and it is one of my favourite things in PHP:

private function pumpGenerator(Generator $events): void
{
    $seq = 0;

    try {
        while ($events->valid()) {
            if (! $this->emit($events->current(), $seq++)) {
                $events->throw(new ClientDisconnected);

                return;
            }

            $events->next();
        }
    } catch (ClientDisconnected) {
        // The generator handled the disconnect itself; nothing else to do.
    }
}

Deep inside the turn, the shared tool loop catches it right where the stream was interrupted and saves what it has:

private function guarded(ChatContext $context, Generator $work, string $phase, string $failedMessage, bool $rememberOnDisconnect): Generator
{
    try {
        yield from $work;
    } catch (ClientDisconnected|TurnStopped) {
        $this->finishCutShort($context, $rememberOnDisconnect);
    } catch (RequestException|UpstreamStreamClosed $e) {
        yield from $this->failTurn($context, $e, $phase, ErrorCode::UpstreamUnavailable, $this->upstreamUnavailableMessage());
    } catch (Throwable $e) {
        yield from $this->failTurn($context, $e, $phase, ErrorCode::Internal, $failedMessage);
    }
}

finishCutShort() runs the output guardrails on the partial text, tidies the citation markers, stores the reply and remembers it for replay. If the turn got cut off before it wrote or ran anything, nothing is saved, because there is nothing worth keeping.

The hosted chat goes further. A reload there should not lose the answer at all. So that controller uses respondUntilFinished(): once the browser is gone, writing stops, but the events are read to the end. The turn finishes as if nothing happened and the full reply is saved.

private function pumpToEnd(iterable $events): void
{
    $seq = 0;
    $listening = true;

    foreach ($events as $event) {
        if ($listening) {
            $listening = $this->emit($event, $seq++);
        }
    }
}

Which creates a new problem, of course. The reloaded page shows up while the old request is still running. If it asks the model again, the visitor gets two answers and the tools run twice. So RunningTurns marks the conversation in the chat cache for the life of the turn, and the reloaded page waits for the reply instead of starting another one:

final class RunningTurns
{
    private const int TTL_SECONDS = 600;

    public function track(string $conversationId, callable $events): Generator
    {
        $this->start($conversationId);

        try {
            $stream = $events();
        } catch (Throwable $e) {
            $this->finish($conversationId);

            throw $e;
        }

        return $this->until($conversationId, $stream);
    }

    private function until(string $conversationId, Generator $events): Generator
    {
        try {
            return yield from $events;
        } finally {
            $this->finish($conversationId);
        }
    }
}

Look at that finally around yield from. It runs whether the generator completes, throws, or gets destroyed early, so the flag can never outlive the turn. The TTL is only there for a PHP process that dies without cleaning up.

Stop is a request of its own#

Here is the thing I got wrong first: if the turn keeps running after the browser leaves, then closing the connection cannot be how Stop works. Stop has to be its own endpoint. It marks the turn in the cache, and the turn checks for that mark while it reads the model's answer and before every tool call.

final class TurnStops
{
    private const float LOOK_EVERY_SECONDS = 0.5;

    public function request(string $requestId): void
    {
        Cache::driver('chat')->put($this->key($requestId), true, self::TTL_SECONDS);
    }

    public function throwIfStopped(ChatContext $context): void
    {
        $requestId = $this->requestIdOf($context);

        if ($requestId === '') {
            return;
        }

        $this->lookedAt[$requestId] = now()->getPreciseTimestamp(3) / 1000;

        if (Cache::driver('chat')->pull($this->key($requestId)) === true) {
            throw new TurnStopped;
        }
    }

    public function throwIfStoppedLately(ChatContext $context): void
    {
        $requestId = $this->requestIdOf($context);
        $now = now()->getPreciseTimestamp(3) / 1000;

        if ($requestId === '' || $now - ($this->lookedAt[$requestId] ?? 0.0) < self::LOOK_EVERY_SECONDS) {
            return;
        }

        $this->throwIfStopped($context);
    }
}

The read loop calls throwIfStoppedLately() on every chunk, which hits the cache at most twice a second. TurnStopped goes through the same guarded() catch as a disconnect, so a stopped answer keeps its partial text too.

The request id is scoped to the agent, the session and the conversation before it ever becomes a cache key. You can only stop your own turn, not someone else's.

The Stop endpoint then waits up to five seconds for the turn to actually end, so whatever the chat does next, like editing the message or asking again, happens after the stopped reply is saved:

if ($requestId !== null) {
    $stops->request($requestId);
}

for ($waited = 0; $waited < self::WAIT_MS && $runningTurns->isRunning($conversationId); $waited += self::CHECK_EVERY_MS) {
    Sleep::for(self::CHECK_EVERY_MS)->milliseconds();
}

return response()->noContent();

Retries should not pay twice#

Phones drop connections. A chat client that retries is a good client. But a model call costs money and tool calls have side effects, so a retry must not run the turn again. Every request carries a client-generated id, and a finished answer is remembered under it for a few minutes, encrypted:

class RequestDedupeStore
{
    public function remember(string $requestId, string $assistantMessageId, string $content, ?string $language, array $sources = []): void
    {
        $snapshot = new ReplaySnapshotData($assistantMessageId, $content, $language, $sources);

        $this->cache()->put($this->key($requestId), CacheCipher::encrypt($snapshot->toArray()), $this->ttl());
    }
}

When a request with a known id comes in, ChatTurnStreamer hands back a ReplaySource instead of calling the model. It yields the same event types as a live turn, with replayed: true on the start event so the UI can skip the typing animation:

class ReplaySource
{
    public function events(string $requestId, ReplaySnapshotData $snapshot): Generator
    {
        $assistantId = $snapshot->assistant_message_id;

        yield StreamEvent::start($assistantId, [
            'assistant_message_id' => $assistantId,
            'user_message_id' => null,
            'request_id' => $requestId,
            'language' => $snapshot->language,
            'replayed' => true,
        ]);

        if ($snapshot->content !== '') {
            yield StreamEvent::delta($assistantId, $snapshot->content);
        }

        if ($snapshot->sources !== []) {
            yield StreamEvent::sources($assistantId, $snapshot->sources);
        }

        yield StreamEvent::done($assistantId, 'complete');
    }
}

And remember, a disconnected turn remembers its partial answer as well. The retry after a dropped connection gets the text that was already written, not a second bill.

Reading the model's stream#

On the other end of the process, Chatvia is itself a client of someone else's event stream. Every OpenAI-compatible API, from OpenAI to vLLM to Ollama, sends data: lines and ends with data: [DONE]. I read bytes from the PSR-7 body, keep a buffer, and yield one payload per complete line:

private function payloads(StreamInterface $body, ChatContext $context): Generator
{
    $buffer = '';

    while (! $body->eof()) {
        $this->stops->throwIfStoppedLately($context);

        try {
            $buffer .= $body->read(self::READ_BYTES);
        } catch (RuntimeException $e) {
            throw new UpstreamStreamClosed(previous: $e);
        }

        while (($newline = strpos($buffer, "\n")) !== false) {
            $line = substr($buffer, 0, $newline);
            $buffer = substr($buffer, $newline + 1);

            if (filled($line) && Str::startsWith($line, self::DATA_PREFIX)) {
                yield substr($line, strlen(self::DATA_PREFIX));
            }
        }
    }
}

Bytes come in 1 KB reads, not lines, so a JSON payload is often cut in half between two reads. The buffer handles that. Nothing clever. But the same problem shows up one level higher, and there it is sneaky.

Open-weight reasoning models like Qwen3 and DeepSeek often write their thinking inline, as a <think>…</think> block at the start of the reply. That text must never reach the visitor as part of the answer. The annoying part is that the tag itself gets split across chunks: one chunk ends with </th, the next one starts with ink>. Naive string matching lets half a tag leak into the answer. So the transformer holds back the longest tail of the buffer that could be the start of the closing tag:

private function partialTagLength(string $text, string $tag): int
{
    for ($length = min(strlen($tag) - 1, strlen($text)); $length > 0; $length--) {
        if (Str::endsWith($text, substr($tag, 0, $length))) {
            return $length;
        }
    }

    return 0;
}

Everything before that tail goes out as a reasoning event. The chat shows it in a collapsible block and never stores it. The real answer only starts streaming once the closing tag is whole.

Tool calls get a similar hold. Some models write a tool call as JSON in the reply text instead of in the tool_calls field. If tools were offered and the reply starts with {, [ or a code fence, I hold the text until the stream ends. If it parses as a call to a tool that was actually offered, it becomes a tool call. Otherwise it streams as the answer and nobody notices the pause.

Give streaming its own PHP-FPM pool#

A streaming request holds a PHP-FPM worker for the whole answer. Ten seconds is normal. A minute with a chain of tool calls is not rare. If chat answers share a pool with the dashboard, a busy afternoon of chats means the dashboard stops loading, and that is a support ticket you do not want.

Chatvia's Docker image runs a second pool for the streaming routes only:

[chatvia-streams]
listen = 127.0.0.1:9001
pm = dynamic
pm.max_children = ${CHATVIA_STREAM_WORKERS}
pm.start_servers = 1
pm.min_spare_servers = 1
pm.max_spare_servers = 1

nginx sends the streaming routes there and leaves everything else on the default pool:

location ~ ^/(?:chat/[^/]+/(?:stream/[^/]+|tools/[^/]+/resolve)|widget/stream|api/v1/agents/[^/]+/conversations/[^/]+/messages)/?$ {
    include fastcgi_params;
    fastcgi_param SCRIPT_FILENAME $document_root/index.php;
    fastcgi_pass 127.0.0.1:9001;
    fastcgi_read_timeout 300s;
}

The pool is dynamic, not ondemand, and that is deliberate. An empty on-demand pool starts one worker when two chats arrive in the same moment, and the second visitor waits for the entire first answer. With one spare worker always idle, the second chat starts immediately, and FPM adds more as they come. There is also a test in the repo that checks this nginx pattern against the app's route list, because I know I will add a streaming route one day and forget the pool. The worker count is one setting in the Docker install.

The browser side#

I do not use EventSource. It cannot send a POST body or custom headers, and it reconnects on its own, which would restart a turn. So it is fetch() and a body reader, splitting on the blank line between events:

const reader = response.body!.getReader();
const decoder = new TextDecoder();
let buffer = '';

while (true) {
    const chunk = await reader.read();

    if (chunk.done) {
        await pacer.flush();
        return;
    }

    buffer += decoder.decode(chunk.value, { stream: true });

    const events = buffer.split('\n\n');
    buffer = events.pop() ?? '';

    for (const raw of events) {
        const match = raw.match(/^data:\s?(.+)$/s);
        if (match) dispatch(JSON.parse(match[1]), callbacks, pacer);
    }
}

That pacer is a tiny thing with an outsized effect on how the chat feels. Fast models finish a paragraph in under a second, and if you render deltas the moment they land, the text appears in ugly blocks. So the composable buffers the text and drains it at a steady rate with requestAnimationFrame, then speeds up once done arrives so the backlog clears before the typing indicator disappears. A tool event drains the buffer immediately, because the text before a tool call belongs to the step above it. A replace event throws the buffer away, because a guardrail just corrected that text.

What I would tell you to do#

If you are building streaming AI features in Laravel, this is where I landed after a lot of wrong turns:

• Stream typed events, never raw text. Errors, sources and tool activity need their own events, and a sequence number makes the stream debuggable.
• Keep HTTP out of the domain. A turn is a generator of events. One class turns it into SSE, another folds it into JSON.
• Treat a closed tab as a signal, not a failure. Keep running, save the partial answer, and make the reloaded page wait for the reply that is still being written.
• Make Stop a request of its own, scoped to the visitor, and check it between reads instead of trusting the connection.
• Give retries a request id and replay the remembered answer instead of calling the model again.
• Buffer at every boundary: bytes to lines, lines to JSON, and tags that split across chunks.
• Put streaming routes in their own PHP-FPM pool, and tell nginx not to buffer them.

All of this lives in the Chatvia repository, under src/Domain/Conversation/Services/Streaming and src/App/Http/Responses/SseResponder.php. If you want to see it run, install Chatvia on a server of yours, open a chat, and watch the network tab. Then close the tab halfway through an answer and reload. That moment is the whole point of this post.