Streaming and structured output solve different problems
Separate incremental display from a complete business result and validate output with a local Python example.
Handle partial output, final validation, refusals, and interrupted streams separately.
Streaming controls when the user sees content. Structured output controls how an application interprets a result. You can use both, but neither replaces the other.
Seeing a few words does not mean the task succeeded. Half of a JSON object must not become a database update or a tool instruction.
Separate display state from business state
Display states include waiting, receiving, completed, and interrupted. Business states can include a usable result, insufficient evidence, a refusal, or a request to execute a tool.
Event boundaries do not necessarily match network chunk boundaries. A single event may arrive in several chunks, and one chunk may contain multiple events. Use the supported SDK or a correct SSE parser.
Check the final event and status. A closed connection alone does not establish that the response is complete.
A correct structure solves only one layer
On a supported interface, structured output constrains a response to a supplied schema. Check the interface's supported schema features and its handling of refusal and incomplete output.
Schema conformance does not make content true or authorize an action. An object containing an amount can be valid JSON while still describing an unauthorized operation.
Separate schema validation, business validation, and permission checks. They answer different questions: are the fields well formed, are the values meaningful, and may this user perform the operation?
Validate a final result locally
This Python example validates a small contract. It does not call a model or parse SSE. It demonstrates a check before a completed result enters application logic.
def validate_result(value):
if not isinstance(value, dict):
raise ValueError("object required")
if set(value) != {"status", "items"}:
raise ValueError("unexpected fields")
if value["status"] not in {"ready", "no_evidence"}:
raise ValueError("invalid status")
if not isinstance(value["items"], list):
raise ValueError("items must be a list")
if not all(isinstance(item, str) for item in value["items"]):
raise ValueError("items must contain strings")
if value["status"] == "no_evidence" and value["items"]:
raise ValueError("no_evidence must have empty items")
return value
result = validate_result({"status": "ready", "items": ["Check sources"]})
assert result["items"] == ["Check sources"]
try:
validate_result({"status": "no_evidence", "items": ["Made up"]})
except ValueError:
print("Rejected inconsistent result")
else:
raise AssertionError("invalid result accepted")
This still does not establish evidence for each item. You can validate source identifiers, but factual checks require the actual material and appropriate evaluation cases.
Give interruptions an explicit behavior
| Situation | User interface | Application behavior |
|---|---|---|
| Incremental content | Show that generation is ongoing | Buffer it; do not commit the result |
| Completed result | Show completion | Validate fields, meaning, and permission |
| Connection interruption | Explain that the task is unfinished | Retain a clear state and offer a safe retry |
| Refusal or incomplete output | Explain that no completed result is available | Use a separate branch instead of forcing a parse |
Retrying a request that contains external writes can duplicate an operation. Regenerating text and re-executing an action should not be the same unconditional button.
Try it next
Add waiting, receiving, completed, and interrupted states to your interface. Prepare a valid local result, a field error, and a business contradiction.
Send all three through one validation entry point and check that they reach different branches. Connect a real streaming interface after these branches work.
Sources & further reading
Sources checked on 2026-11-10. Original engineering explanation; examples are teaching data, not production measurements.