How do I stop a streaming chat client from crashing on an empty delta?

I work on CloudGPU. Treat streamed chunks as events, not complete chat messages. A chunk can contain a role update, no text, or no choices. Check before indexing, append only text, and keep the completion's finish reason separate from the accumulated answer.

1. Isolate the reducer from network code

Save this helper as stream_text.py. It consumes the shape of an SDK chunk after model_dump() and only follows choice index zero:

def collect_text(chunks):
    pieces = []
    finish = None
    for chunk in chunks:
        for choice in chunk.get("choices") or []:
            if choice.get("index") != 0:
                continue
            delta = choice.get("delta") or {}
            content = delta.get("content")
            if isinstance(content, str):
                pieces.append(content)
            reason = choice.get("finish_reason")
            if reason is not None:
                finish = reason
    return "".join(pieces), finish

The index is explicit: do not concatenate different alternatives into one answer. This helper handles text only. It does not interpret extra model-specific fields or claim they have the same meaning as answer text.

2. Exercise the empty cases without calling a model

In a separate local file, use a deliberately synthetic sequence:

from stream_text import collect_text

fixture = [
    {"choices": []},
    {"choices": [{"index": 0, "delta": {"role": "assistant"}}]},
    {"choices": [{"index": 0, "delta": {"content": "hello"}}]},
    {"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]},
]
assert collect_text(fixture) == ("hello", "stop")

These events are a client-side fixture, not a captured CloudGPU response. A reducer can be tested without credentials and without inferring which optional events a particular model sends.

3. Attach the reducer to an authorized live stream

Set ${CLOUDGPU_API_KEY} and ${CLOUDGPU_BASE_URL} in the process environment. Use the documented current API base including its version prefix. Add this to a separate script when you are ready for a live request:

import os
from openai import OpenAI
from stream_text import collect_text

with OpenAI(
    api_key=os.environ["CLOUDGPU_API_KEY"],
    base_url=os.environ["CLOUDGPU_BASE_URL"],
    max_retries=0,
) as client:
    stream = client.chat.completions.create(
        model="deepseek-v4-flash",
        messages=[{"role": "user", "content": "Name one Python collection."}],
        stream=True,
    )
    try:
        text, finish = collect_text(chunk.model_dump() for chunk in stream)
    finally:
        stream.close()
    print(text)
    print({"finish_reason": finish})

An interrupted iterator raises; do not turn a partial buffer into a successful answer. Even when the iterator ends normally, a missing finish reason should remain unresolved in your application. If the reason indicates truncation, retain that state instead of labeling the answer complete. Never infer success from a nonempty text buffer alone.

Official references: Python SDK and CloudGPU's streaming contract. The model identifier was checked against today's public catalog; that is not a live-stream validation.

Related CloudGPU video: This Video Was Made by AI. Now Run AI Yourself. It provides product context, not evidence that this particular streaming sequence was observed.

The preparation checks cover syntax and an offline reducer fixture only. No API generation or GPU run was performed.