temporalio.contrib.openai_agents._invoke_model_activity.ModelActivity

class documentation

class ModelActivity: (source)

Constructor: ModelActivity(model_provider)

View In Hierarchy

Class wrapper for model invocation activities to allow model customization. By default, we use an OpenAIProvider with retries disabled. Disabling retries in your model of choice is recommended to allow activity retries to define the retry model.

Method __init__ Initialize the activity with a model provider.
Async Method invoke_model_activity Activity that invokes a model with the given input.
Async Method invoke_model_activity_streaming Streaming-aware model activity.
Instance Variable _model_provider Undocumented
def __init__(self, model_provider: ModelProvider | None = None): (source)

Initialize the activity with a model provider.

async def invoke_model_activity(self, input: ActivityModelInput) -> ModelResponse: (source)

Activity that invokes a model with the given input.

async def invoke_model_activity_streaming(self, input: StreamingActivityModelInput) -> list[TResponseStreamEvent]: (source)

Streaming-aware model activity.

Warning

Streaming support is experimental and may change in future versions.

Calls model.stream_response() and returns the collected list of native OpenAI stream events. The workflow's Model.stream_response stub yields these to the agents framework, which builds the final ModelResponse from the terminal ResponseCompletedEvent.

Each event is also published to the workflow's stream on streaming_topic so external consumers (UIs, tracing, etc.) can observe events as they arrive.

Heartbeats run on a background task via _auto_heartbeater so long initial-token latency or long pauses between chunks do not trip heartbeat_timeout.

_model_provider = (source)

Undocumented