Stream turn output

A conversational turn takes a few seconds: router call, classifier call, catalog lookup, slot generation. A spinner for all of that feels dead. Streaming node transitions over SSE turns the same wait into a live feed. The UI shows "routing intent", "matching specialty", "finding doctors", one after another. Same latency, completely different feel.

booking_graph.py
python
def _record_event(state: BookingState, node: str, detail: Dict[str, Any]) -> None:
    events = state.get("events") or []
    events.append({"node": node, **detail})
    state["events"] = events

# Every node calls this on completion:
#   _record_event(state, "intent_router",   {"intent": intent})
#   _record_event(state, "match_specialty", {"specialty": specialty})
#   _record_event(state, "find_doctor",     {"doctor": doctor_name})
#   _record_event(state, "propose_slots",   {"slots": slots})
#   _record_event(state, "confirm_booking", {"appointment_id": appt_id})

Each node appends an event to a list on the state. The events capture what the node produced. By the end of the turn, state["events"] is an ordered trace of everything that happened.

booking_graph.py
python
async def run_turn(thread_id, user_message, patient_id=None):
    state = get_thread_state(thread_id)
    state["user_message"] = user_message
    state["patient_id"] = patient_id
    state["events"] = []
    state["assistant_reply"] = ""

    graph = get_graph()

    if state.get("status") == "ready_to_confirm":
        result = await confirm_booking(state)
    else:
        result = await graph.ainvoke(state)

    _THREADS[thread_id] = dict(result)

    # Stream each node event, then final reply
    for ev in result.get("events", []) or []:
        yield {"type": "node", **ev}

    yield {
        "type": "final",
        "thread_id": thread_id,
        "status": result.get("status"),
        "reply": result.get("assistant_reply", ""),
    }

run_turn is an async generator. It yields one event per node, then a final event with the assistant reply. The caller (the FastAPI handler) wraps each yield as an SSE frame.

One naming note before the endpoint. The graph module exposes this turn generator as run_turn. The FastAPI layer imports it through a thin service function called chat_turn, so the handler below iterates chat_turn with the exact signature you just saw. Same generator, an endpoint-friendly name.

router.py
python
@router.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    thread_id = request.thread_id or new_thread_id()

    async def event_stream():
        try:
            yield f"data: {json.dumps({'type': 'start', 'thread_id': thread_id})}\n\n"
            async for event in chat_turn(
                thread_id=thread_id,
                user_message=request.message,
                patient_id=request.patient_id,
            ):
                yield f"data: {json.dumps(event)}\n\n"
            yield f"data: {json.dumps({'type': 'done'})}\n\n"
        except Exception as e:
            logger.exception("chat_stream error")
            yield f"data: {json.dumps({'type': 'error', 'error': str(e)})}\n\n"

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )

The FastAPI handler wraps each event in the SSE frame format: data: <json>\n\n. The headers disable proxy buffering so events reach the client as they are produced, not in a single batch at the end.

Ordering exercise: Order the SSE frames the client sees for one booking turn

Loading practice…

Quiz: Quiz

Loading practice…