|
| 1 | +# Streaming OpenAI Agents |
| 2 | + |
| 3 | +> **Experimental.** These samples use the streaming support in |
| 4 | +> `temporalio.contrib.openai_agents` together with |
| 5 | +> `temporalio.contrib.workflow_streams`. Both are experimental and their APIs |
| 6 | +> may change in future versions. |
| 7 | +
|
| 8 | +*Adapted from the [OpenAI Agents SDK basic examples](https://github.com/openai/openai-agents-python/tree/main/examples/basic)* |
| 9 | + |
| 10 | +Before running these examples, be sure to review the [prerequisites and background on the integration](../README.md). |
| 11 | + |
| 12 | +The OpenAI Agents SDK streams model output via `Runner.run_streamed`, which |
| 13 | +yields events as the model produces them. Inside a Temporal workflow the model |
| 14 | +call runs in an activity, so the workflow cannot iterate the live HTTP stream |
| 15 | +directly. Instead the plugin runs `model.stream_response()` in a streaming |
| 16 | +activity, and that activity publishes each event to the workflow's |
| 17 | +[`WorkflowStream`](../../workflow_streams/README.md) so external subscribers |
| 18 | +see events as they are produced. |
| 19 | + |
| 20 | +Publishing is batched: the activity coalesces events over |
| 21 | +`ModelActivityParameters.streaming_batch_interval` (default 100ms) before |
| 22 | +signalling the workflow. Call this **buffered token streaming** — deltas reach |
| 23 | +subscribers within a batch window of being produced, not on every byte. At |
| 24 | +typical model speeds one batch carries several tokens, so output arrives in |
| 25 | +small bursts rather than glyph-by-glyph. Lower the interval for smoother |
| 26 | +output at the cost of more signals. |
| 27 | + |
| 28 | +Two things to know before reading the samples: |
| 29 | + |
| 30 | +* `streaming_topic` is **required** for `Runner.run_streamed`. If it is unset, |
| 31 | + `run_streamed` raises before scheduling any activity. |
| 32 | +* The workflow must host a `WorkflowStream`. It has to be constructed from a |
| 33 | + method named `__init__` — `WorkflowStream` inspects its caller's frame and |
| 34 | + raises otherwise — and `@workflow.init` is what makes the workflow's run |
| 35 | + argument (carrying `stream_state` for continue-as-new) available there. |
| 36 | + |
| 37 | +## Running the Examples |
| 38 | + |
| 39 | +First, start the worker (supports both examples): |
| 40 | + |
| 41 | +```bash |
| 42 | +uv run openai_agents/streaming/run_worker.py |
| 43 | +``` |
| 44 | + |
| 45 | +Then run either example in another terminal. |
| 46 | + |
| 47 | +### `stream_text` — buffered text deltas |
| 48 | + |
| 49 | +Adapted from [`examples/basic/stream_text.py`][upstream-text]. The workflow |
| 50 | +just calls `Runner.run_streamed`; the subscriber renders the |
| 51 | +`ResponseTextDeltaEvent`s the streaming activity publishes on the `events` |
| 52 | +topic. |
| 53 | + |
| 54 | +Subscribers receive **native OpenAI events** (`TResponseStreamEvent`), because |
| 55 | +the activity publishes them straight from `Model.stream_response`. That differs |
| 56 | +from `stream_events()` inside the workflow, which yields the agents-SDK |
| 57 | +`StreamEvent` union — raw model events arrive there wrapped as |
| 58 | +`RawResponsesStreamEvent.data`. |
| 59 | + |
| 60 | +[upstream-text]: https://github.com/openai/openai-agents-python/blob/main/examples/basic/stream_text.py |
| 61 | + |
| 62 | +```bash |
| 63 | +uv run openai_agents/streaming/run_stream_text_workflow.py |
| 64 | +``` |
| 65 | + |
| 66 | +### `stream_items` — agent-level events with a tool call |
| 67 | + |
| 68 | +Adapted from [`examples/basic/stream_items.py`][upstream-items]. Renders agent |
| 69 | +updates, tool calls, tool outputs, and message outputs as a play-by-play. |
| 70 | + |
| 71 | +The agents SDK builds those higher-level events from the model output, so they |
| 72 | +exist only inside the workflow — the streaming activity never sees them. This |
| 73 | +workflow therefore does its own publishing: it iterates |
| 74 | +`result.stream_events()` and forwards each event of interest to an `items` |
| 75 | +topic as a small serializable `ItemEvent`. (The agents-SDK event types carry |
| 76 | +the originating `Agent`, which holds tool callables and so cannot be |
| 77 | +serialized.) `stream_events()` resolves a turn at a time — each model call is |
| 78 | +one activity — so a multi-turn run like this one reaches the subscriber |
| 79 | +progressively rather than in one lump. |
| 80 | + |
| 81 | +[upstream-items]: https://github.com/openai/openai-agents-python/blob/main/examples/basic/stream_items.py |
| 82 | + |
| 83 | +```bash |
| 84 | +uv run openai_agents/streaming/run_stream_items_workflow.py |
| 85 | +``` |
| 86 | + |
| 87 | +## How it works |
| 88 | + |
| 89 | +1. The workflow constructs a `WorkflowStream` in `@workflow.init`. |
| 90 | +2. `OpenAIAgentsPlugin` is configured with `streaming_topic="events"`, which |
| 91 | + routes `Runner.run_streamed` to `invoke_model_activity_streaming`. |
| 92 | +3. Inside that activity each event from the live HTTP stream is both collected |
| 93 | + (returned to the workflow when the activity completes) and published to the |
| 94 | + stream via `WorkflowStreamClient.from_within_activity()`. |
| 95 | +4. Just before returning, the workflow publishes a terminator on a separate |
| 96 | + `done` topic, then sleeps briefly so the subscriber's next poll can drain |
| 97 | + the tail of the stream — the log lives in workflow memory and disappears |
| 98 | + when the run completes. |
| 99 | +5. External code subscribes with |
| 100 | + `WorkflowStreamClient.create(...).subscribe([...], result_type=RawValue)` |
| 101 | + and breaks on the terminator. `RawValue` keeps the payloads undecoded so |
| 102 | + each topic can be decoded against its own type. If the workflow reaches a |
| 103 | + terminal state without publishing a terminator (a failure, say), the |
| 104 | + iterator exhausts on its own and the following `handle.result()` raises. |
| 105 | + |
| 106 | +In the workflow, `stream_events()` resolves only after the model activity |
| 107 | +returns, so the workflow itself does not see deltas as they arrive — the |
| 108 | +streaming benefit is for external observers. |
| 109 | + |
| 110 | +## Notes |
| 111 | + |
| 112 | +* Streaming is incompatible with `use_local_activity=True`: local activities |
| 113 | + support neither heartbeats nor the workflow stream signal channel. |
| 114 | +* The streaming activity heartbeats on a background task, so set |
| 115 | + `heartbeat_timeout` well below `start_to_close_timeout` to detect a stuck |
| 116 | + model call early. |
| 117 | +* Delivery is at-least-once per activity attempt. An attempt that fails |
| 118 | + mid-response leaves its partial events on the stream — they are flushed |
| 119 | + before the failure is reported — and the retry publishes a whole new |
| 120 | + response. `stream_events()` in the workflow only sees the successful attempt, |
| 121 | + so the workflow's return value stays correct while a naive subscriber renders |
| 122 | + the truncated attempt followed by the full one. |
| 123 | + |
| 124 | + The plugin's streaming activity publishes no retry marker, so subscribers |
| 125 | + detect this in band: every OpenAI stream event carries a `sequence_number` |
| 126 | + that starts at 0 per response, and a number that fails to advance means a new |
| 127 | + attempt. `run_stream_text_workflow.py` prints a notice at that seam; |
| 128 | + `workflow_streams/run_llm.py` shows the fuller treatment, where an activity |
| 129 | + you own publishes an explicit `RetryEvent` from `activity.info().attempt` and |
| 130 | + the consumer erases the failed attempt's output. |
0 commit comments