temporalio.contrib.google_adk_agents._model

module documentation
(source)

Undocumented

Class StreamingInvokeInput Input for invoke_model_streaming.
Async Function invoke_model Activity that invokes an LLM model.
Async Function invoke_model_streaming Streaming-aware model activity.
Function _with_serializable_response_schema Return the request with a response_schema that can be serialized.
async def invoke_model(llm_request: LlmRequest) -> list[LlmResponse]: (source)

Activity that invokes an LLM model.

Parameters
llm_request:LlmRequestThe LLM request containing model name and parameters.
Returns
list[LlmResponse]List of LLM responses from the model.
Raises
ValueErrorIf model name is not provided or LLM creation fails.
async def invoke_model_streaming(input: StreamingInvokeInput) -> list[LlmResponse]: (source)

Streaming-aware model activity.

Warning

Streaming support is experimental and may change in future versions.

Calls the LLM with stream=True and returns the collected list of raw LlmResponse chunks. The workflow's TemporalModel.generate_content_async yields these to the caller.

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

def _with_serializable_response_schema(llm_request: LlmRequest) -> LlmRequest: (source)

Return the request with a response_schema that can be serialized.

ADK stores an agent's output_schema on the request as a Python type (for example a Pydantic model class), which the payload converter cannot serialize. google-genai and ADK's LiteLlm both turn such a type into its JSON schema before calling the model, so sending the JSON schema instead is equivalent. Pydantic model classes use their model_json_schema method to preserve custom schema generation. Integer-valued enums are normalized to string enums to match google-genai's enum handling.