Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 69 additions & 0 deletions google/genai/_interactions/types/interaction.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,9 @@
from .video_content import VideoContent
from .._legacy_lyria import is_legacy_lyria_response_body
from .webhook_config import WebhookConfig
from .user_input_step import UserInputStep
from .document_content import DocumentContent
from .model_output_step import ModelOutputStep
from .dynamic_agent_config import DynamicAgentConfig
from .text_response_format import TextResponseFormat
from .audio_response_format import AudioResponseFormat
Expand Down Expand Up @@ -212,3 +214,70 @@ def _coerce_outputs_to_steps(cls, values: Any) -> Any:
def _coerce_outputs_to_steps(cls, data: Any) -> Any:
coerced, _ = cls._maybe_coerce_outputs(data)
return coerced

@property
def _last_model_output_steps(self) -> List[ModelOutputStep]:
if not self.steps:
return []
trailing: List[ModelOutputStep] = []
for step in reversed(self.steps):
if isinstance(step, ModelOutputStep):
trailing.append(step)
else:
break
trailing.reverse()
return trailing

@property
def output_text(self) -> str:
"""Concatenated last consecutive text from the last consecutive model output steps."""
parts: List[str] = []
done = False
for step in reversed(self._last_model_output_steps):
if done:
break
if step.content:
for content in reversed(step.content):
if isinstance(content, TextContent):
parts.append(content.text)
else:
done = True
break
parts.reverse()
return "".join(parts)

@property
def output_image(self) -> Optional[ImageContent]:
"""The last image generated by the model in response to the current request."""
for step in reversed(self.steps):
if isinstance(step, UserInputStep):
break
if isinstance(step, ModelOutputStep) and step.content:
for content in reversed(step.content):
if isinstance(content, ImageContent):
return content
return None

@property
def output_audio(self) -> Optional[AudioContent]:
"""The last audio generated by the model in response to the current request."""
for step in reversed(self.steps):
if isinstance(step, UserInputStep):
break
if isinstance(step, ModelOutputStep) and step.content:
for content in reversed(step.content):
if isinstance(content, AudioContent):
return content
return None

@property
def output_video(self) -> Optional[VideoContent]:
"""The last video generated by the model in response to the current request."""
for step in reversed(self.steps):
if isinstance(step, UserInputStep):
break
if isinstance(step, ModelOutputStep) and step.content:
for content in reversed(step.content):
if isinstance(content, VideoContent):
return content
return None
Loading