+
+ Returns:
+ ImageContent with extracted base64 data and media type
+ """
+ try:
+ header, b64_data = data_url.split(',', 1)
+ media_type = header.split(':')[1].split(';')[0] # e.g., "image/png"
+ return cls(image_data=b64_data, media_type=media_type)
+ except (ValueError, IndexError):
+ # Fallback: assume the whole thing is base64 data
+ return cls(image_data=data_url.split(',')[-1], media_type="image/jpeg")
+
+ @staticmethod
+ def autocast(value: Any, format: str = "PNG") -> Dict[str, Any]:
+ """Auto-detect value type and return image field values.
+
+ Args:
+ value: Can be:
+ - URL string (starting with 'http://' or 'https://')
+ - Data URL string (starting with 'data:image/')
+ - Local file path (string)
+ - Numpy array or array-like RGB image
+ - PIL Image object
+ - Raw bytes
+ - None (empty image)
+ format: Image format for numpy arrays (PNG, JPEG, etc.). Default: PNG
+
+ Returns:
+ Dictionary with keys: image_url, image_data, image_bytes, media_type
+ """
+ # Handle None or empty
+ if value is None:
+ return {"image_url": None, "image_data": None, "image_bytes": None, "media_type": "image/jpeg"}
+
+ # Handle ImageContentBlock instance
+ if isinstance(value, ImageContent):
+ return {
+ "image_url": value.image_url,
+ "image_data": value.image_data,
+ "image_bytes": value.image_bytes,
+ "media_type": value.media_type
+ }
+
+ # Handle string inputs
+ if isinstance(value, str):
+ if not value.strip():
+ return {"image_url": None, "image_data": None, "image_bytes": None, "media_type": "image/jpeg"}
+
+ # Data URL
+ if value.startswith('data:image/'):
+ try:
+ header, b64_data = value.split(',', 1)
+ media_type = header.split(':')[1].split(';')[0]
+ return {"image_url": None, "image_data": b64_data, "image_bytes": None, "media_type": media_type}
+ except (ValueError, IndexError):
+ return {"image_url": None, "image_data": value.split(',')[-1], "image_bytes": None, "media_type": "image/jpeg"}
+
+ # HTTP/HTTPS URL
+ if value.startswith('http://') or value.startswith('https://'):
+ return {"image_url": value, "image_data": None, "image_bytes": None, "media_type": "image/jpeg"}
+
+ # File path - only check if string is reasonable length (< 4096 chars)
+ # Long strings are clearly not file paths and would cause OS errors
+ if len(value) < 4096:
+ path = Path(value)
+ try:
+ if path.exists():
+ ext_to_type = {
+ '.jpg': 'image/jpeg',
+ '.jpeg': 'image/jpeg',
+ '.png': 'image/png',
+ '.gif': 'image/gif',
+ '.webp': 'image/webp'
+ }
+ media_type = ext_to_type.get(path.suffix.lower(), 'image/jpeg')
+ with open(value, 'rb') as f:
+ image_data = base64.b64encode(f.read()).decode('utf-8')
+ return {"image_url": None, "image_data": image_data, "image_bytes": None, "media_type": media_type}
+ except (OSError, IOError):
+ # Not a valid file path, continue to other checks
+ pass
+
+ # Handle bytes - store as base64 for portability
+ if isinstance(value, bytes):
+ image_data = base64.b64encode(value).decode('utf-8')
+ return {"image_url": None, "image_data": image_data, "image_bytes": None, "media_type": "image/jpeg"}
+
+ # Handle PIL Image
+ try:
+ from PIL import Image
+ if isinstance(value, Image.Image):
+ import io
+ buffer = io.BytesIO()
+ img_format = value.format or format.upper()
+ value.save(buffer, format=img_format)
+ buffer.seek(0)
+ image_data = base64.b64encode(buffer.getvalue()).decode('utf-8')
+ media_type = f"image/{img_format.lower()}"
+ return {"image_url": None, "image_data": image_data, "image_bytes": None, "media_type": media_type}
+ except ImportError:
+ pass
+
+ # Handle numpy array or array-like
+ try:
+ import numpy as np
+ if isinstance(value, np.ndarray) or hasattr(value, '__array__'):
+ try:
+ from PIL import Image
+ except ImportError:
+ raise ImportError("Pillow is required for array conversion. Install with: pip install Pillow")
+
+ import io
+
+ if not isinstance(value, np.ndarray):
+ value = np.array(value)
+
+ # Normalize to [0, 255] if needed
+ if value.dtype == np.float32 or value.dtype == np.float64:
+ if value.max() <= 1.0:
+ value = (value * 255).astype(np.uint8)
+ else:
+ value = value.astype(np.uint8)
+ elif value.dtype != np.uint8:
+ value = value.astype(np.uint8)
+
+ image = Image.fromarray(value)
+ buffer = io.BytesIO()
+ image.save(buffer, format=format.upper())
+ buffer.seek(0)
+
+ image_data = base64.b64encode(buffer.getvalue()).decode('utf-8')
+ media_type = f"image/{format.lower()}"
+ return {"image_url": None, "image_data": image_data, "image_bytes": None, "media_type": media_type}
+ except ImportError:
+ pass
+
+ return {"image_url": None, "image_data": None, "image_bytes": None, "media_type": "image/jpeg"}
+
+ @classmethod
+ def build(cls, value: Any, format: str = "PNG") -> 'ImageContent':
+ """Auto-detect format and create ImageContent from various input types.
+
+ Args:
+ value: Can be:
+ - URL string (starting with 'http://' or 'https://')
+ - Data URL string (starting with 'data:image/')
+ - Local file path (string)
+ - Numpy array or array-like RGB image
+ - PIL Image object
+ - Raw bytes
+ format: Image format for numpy arrays (PNG, JPEG, etc.). Default: PNG
+
+ Returns:
+ ImageContent or None if the value cannot be converted
+ """
+ # Handle ImageContentBlock instance directly
+ if isinstance(value, cls):
+ return value
+
+ value_dict = cls.autocast(value, format=format)
+ return cls(**value_dict)
+
+ def set_image(self, image: Any, format: str = "PNG") -> None:
+ """Set the image from various input formats (mutates self).
+
+ Args:
+ image: Can be:
+ - URL string (starting with 'http://' or 'https://')
+ - Data URL string (starting with 'data:image/')
+ - Local file path (string)
+ - Numpy array or array-like RGB image
+ - PIL Image object
+ - Raw bytes
+ format: Image format for numpy arrays (PNG, JPEG, etc.). Default: PNG
+ """
+ result = ImageContent.build(image, format=format)
+ if result:
+ self.image_url = result.image_url
+ self.image_data = result.image_data
+ # Only copy image_bytes if it was explicitly set (e.g., from Google API)
+ if result.image_bytes:
+ self.image_bytes = result.image_bytes
+ self.media_type = result.media_type
+
+ def as_image(self) -> Image.Image:
+ """Convert the image to a PIL Image.
+
+ Fetches the image from URL if necessary (including HTTP/HTTPS URLs).
+
+ Returns:
+ PIL Image object
+
+ Raises:
+ ValueError: If no image data is available
+ requests.RequestException: If fetching from URL fails
+ """
+ # Try to get image bytes from any available source
+ image_bytes = self.get_bytes()
+
+ if image_bytes:
+ return Image.open(io.BytesIO(image_bytes))
+ elif self.image_url:
+ if self.image_url.startswith(('http://', 'https://')):
+ # Fetch image from URL
+ try:
+ import requests
+ response = requests.get(self.image_url, timeout=30)
+ response.raise_for_status()
+ return Image.open(io.BytesIO(response.content))
+ except ImportError:
+ # Fallback to urllib if requests is not available
+ from urllib.request import urlopen
+ with urlopen(self.image_url, timeout=30) as response:
+ return Image.open(io.BytesIO(response.read()))
+ else:
+ # If it's a local file path
+ return Image.open(self.image_url)
+ else:
+ raise ValueError("No image data available to convert to PIL Image")
+
+ def show(self) -> Image.Image:
+ """A convenience alias for as_image()"""
+ return self.as_image()
+
+ def get_bytes(self) -> Optional[bytes]:
+ """Get raw image bytes.
+
+ Returns image_bytes if available, otherwise decodes image_data from base64.
+
+ Returns:
+ Raw image bytes or None if no image data available
+ """
+ if self.image_bytes:
+ return self.image_bytes
+ elif self.image_data:
+ return base64.b64decode(self.image_data)
+ return None
+
+ def get_base64(self) -> Optional[str]:
+ """Get base64-encoded image data.
+
+ Returns image_data if available, otherwise encodes image_bytes to base64.
+
+ Returns:
+ Base64-encoded string or None if no image data available
+ """
+ if self.image_data:
+ return self.image_data
+ elif self.image_bytes:
+ return base64.b64encode(self.image_bytes).decode('utf-8')
+ return None
+
+ def ensure_bytes(self) -> None:
+ """Ensure image_bytes is populated (converts from image_data if needed)."""
+ if not self.image_bytes and self.image_data:
+ self.image_bytes = base64.b64decode(self.image_data)
+
+ def ensure_base64(self) -> None:
+ """Ensure image_data is populated (converts from image_bytes if needed)."""
+ if not self.image_data and self.image_bytes:
+ self.image_data = base64.b64encode(self.image_bytes).decode('utf-8')
+
+
+# Union type alias for the supported content types (for type hints).
+ContentBlock = Union[TextContent, ImageContent]
diff --git a/opto/utils/backbone/template.py b/opto/utils/backbone/template.py
new file mode 100644
index 00000000..ae0b659d
--- /dev/null
+++ b/opto/utils/backbone/template.py
@@ -0,0 +1,194 @@
+"""PromptTemplate: ``str.format``-like templating that also supports
+multimodal :class:`ContentBlockList` values.
+"""
+from typing import Union
+
+from .content import ContentBlockList
+
+
+class PromptTemplate:
+ """Template for building ContentBlockLists with {placeholder} support.
+
+ Similar to str.format(), but supports multimodal content (ContentBlockList).
+
+ Return type depends on values:
+ - All strings ā returns str (backward compatible)
+ - Any multimodal content ā returns ContentBlockList
+
+ Features:
+ - Multiple placeholders: {a}, {b}, {c}
+ - Escaping: {{ and }} for literal braces
+ - Missing placeholders: left as-is in text
+ - Extra kwargs: silently ignored (no error)
+ - Nested templates: if value is PromptTemplate, formats it first
+ - Mixed values: str, ContentBlockList, or objects with to_content_blocks()
+
+ Examples:
+ # Define template (can be class attribute)
+ user_prompt_template = PromptTemplate('''
+ Now you see problem instance:
+
+ ================================
+ {problem_instance}
+ ================================
+ ''')
+
+ # Format with ContentBlockList (may contain images)
+ content = user_prompt_template.format(
+ problem_instance=problem.to_content_blocks()
+ )
+ # Returns ContentBlockList: [TextContent("Now you see..."), *problem_blocks, TextContent("===...")]
+
+ # Multiple placeholders
+ template = PromptTemplate("User: {user}\\nAssistant: {assistant}")
+ result = template.format(user=user_blocks, assistant=assistant_blocks)
+
+ # Nested templates
+ outer = PromptTemplate("Header\\n{body}\\nFooter")
+ inner = PromptTemplate("Content: {data}")
+ result = outer.format(body=inner, data="some data") # inner gets same kwargs
+
+ # Escaping braces
+ template = PromptTemplate('JSON example: {{"key": "{value}"}}')
+ result = template.format(value="hello") # {"key": "hello"}
+
+ # Extra kwargs are ignored (no error)
+ result = template.format(value="hello", unused_key="ignored")
+
+ # Missing placeholders left as-is
+ template = PromptTemplate("Hello {name}, score: {score}")
+ result = template.format(name="Alice") # "Hello Alice, score: {score}"
+ """
+
+ # Regex to find {placeholder} but not {{ or }}
+ _PLACEHOLDER_PATTERN = None # Lazy compiled
+
+ def __init__(self, template: str):
+ """Initialize with a template string.
+
+ Args:
+ template: Template string with {placeholder} syntax.
+ """
+ self.template = template
+
+ @classmethod
+ def _get_pattern(cls):
+ """Lazily compile the placeholder regex pattern."""
+ if cls._PLACEHOLDER_PATTERN is None:
+ import re
+ # Match {name} but not {{ or }}
+ # Captures the placeholder name
+ cls._PLACEHOLDER_PATTERN = re.compile(r'\{(\w+)\}')
+ return cls._PLACEHOLDER_PATTERN
+
+ def format(self, **kwargs) -> Union[str, 'ContentBlockList']:
+ """Format the template with the given values.
+
+ Similar to str.format(), but supports multimodal content.
+ Extra kwargs are silently ignored.
+
+ If all values are strings, returns a str (backward compatible).
+ If any value is a ContentBlockList or multimodal, returns ContentBlockList.
+
+ Args:
+ **kwargs: Placeholder values. Each value can be:
+ - str: inserted as text
+ - ContentBlockList: blocks spliced in at that position
+ - PromptTemplate: formatted first, then spliced in
+ - Object with to_content_blocks(): method called, result spliced
+ - Other: converted to str
+
+ Returns:
+ str: If all values are strings (backward compatible behavior).
+ ContentBlockList: If any value is multimodal content.
+ """
+ # Check if all values are simple strings - if so, use simple string formatting
+ pattern = self._get_pattern()
+ placeholder_names = set(pattern.findall(self.template))
+
+ # Only check values for placeholders that exist in the template
+ relevant_values = {k: v for k, v in kwargs.items() if k in placeholder_names}
+
+ if all(isinstance(v, str) for v in relevant_values.values()):
+ # All strings: use simple string replacement, return str
+ # Handle escaping and missing placeholders
+ result = self.template.replace("{{", "\x00LBRACE\x00").replace("}}", "\x00RBRACE\x00")
+
+ for name in placeholder_names:
+ placeholder = "{" + name + "}"
+ if name in kwargs:
+ result = result.replace(placeholder, kwargs[name])
+ # Missing placeholders left as-is
+
+ result = result.replace("\x00LBRACE\x00", "{").replace("\x00RBRACE\x00", "}")
+ return result
+
+ # Multimodal content: build ContentBlockList
+ result = ContentBlockList()
+
+ # Handle escaping: replace {{ with a sentinel, }} with another
+ LBRACE_SENTINEL = "\x00LBRACE\x00"
+ RBRACE_SENTINEL = "\x00RBRACE\x00"
+
+ text = self.template.replace("{{", LBRACE_SENTINEL).replace("}}", RBRACE_SENTINEL)
+
+ last_end = 0
+
+ for match in pattern.finditer(text):
+ # Add text before this placeholder
+ prefix = text[last_end:match.start()]
+ if prefix:
+ # Restore escaped braces in prefix
+ prefix = prefix.replace(LBRACE_SENTINEL, "{").replace(RBRACE_SENTINEL, "}")
+ result.append(prefix)
+
+ # Get placeholder name and value
+ placeholder_name = match.group(1)
+
+ if placeholder_name in kwargs:
+ value = kwargs[placeholder_name]
+ # Convert value to ContentBlockList and splice in
+ content = self._value_to_content(value, **kwargs)
+ result.extend(content)
+ else:
+ # Missing placeholder: leave as-is (restore original {name})
+ result.append("{" + placeholder_name + "}")
+
+ last_end = match.end()
+
+ # Add remaining text after last placeholder
+ suffix = text[last_end:]
+ if suffix:
+ suffix = suffix.replace(LBRACE_SENTINEL, "{").replace(RBRACE_SENTINEL, "}")
+ result.append(suffix)
+
+ return result
+
+ def _value_to_content(self, value, **kwargs) -> 'ContentBlockList':
+ """Convert a value to ContentBlockList.
+
+ Args:
+ value: The value to convert
+ **kwargs: Passed to nested PromptTemplate.render()
+
+ Returns:
+ ContentBlockList: The value as content blocks.
+ """
+ if isinstance(value, ContentBlockList):
+ return value
+ elif isinstance(value, PromptTemplate):
+ # Nested template: format it with the same kwargs
+ return value.format(**kwargs)
+ elif hasattr(value, 'to_content_blocks'):
+ # Object with to_content_blocks method (e.g., ProblemInstance)
+ return value.to_content_blocks()
+ elif isinstance(value, str):
+ return ContentBlockList(value)
+ else:
+ # Fallback: convert to string
+ return ContentBlockList(str(value))
+
+ def __repr__(self) -> str:
+ """Return a preview of the template."""
+ preview = self.template[:50] + "..." if len(self.template) > 50 else self.template
+ return f"PromptTemplate({preview!r})"
diff --git a/opto/utils/backbone/turns.py b/opto/utils/backbone/turns.py
new file mode 100644
index 00000000..73953b71
--- /dev/null
+++ b/opto/utils/backbone/turns.py
@@ -0,0 +1,748 @@
+"""Conversation turns: :class:`UserTurn` and :class:`AssistantTurn`.
+
+``AssistantTurn.autocast`` parses raw responses from LiteLLM/OpenAI (Responses
+and Completion APIs), Bedrock Converse, and Google GenAI into a uniform shape.
+"""
+from typing import List, Dict, Any, Optional
+from dataclasses import dataclass, field
+
+from .content import ContentBlockList, TextContent, ImageContent
+
+
+@dataclass
+class UserTurn:
+ """Represents a user message turn in the conversation"""
+ role: str = "user"
+
+ content: ContentBlockList = field(default_factory=ContentBlockList)
+
+ # Provider-specific settings
+ temperature: Optional[float] = None
+ max_tokens: Optional[int] = None
+ top_p: Optional[float] = None
+
+ # Metadata
+ timestamp: Optional[str] = None
+ metadata: Dict[str, Any] = field(default_factory=dict)
+
+ def __init__(self, content=None, tools=None, **kwargs):
+ """
+ Initialize UserTurn with content and tools.
+
+ Four ways to initialize:
+ 1. Empty: UserTurn() - creates empty turn with defaults
+ 2. Copy: UserTurn(existing_turn) - creates a copy of an existing UserTurn
+ 3. Positional args: UserTurn(content, tools) - pass content and/or tools
+ 4. Keyword args: UserTurn(content=..., tools=..., temperature=...) - explicit fields
+
+ Args:
+ content: ContentBlockList, list of content blocks, UserTurn (for copying), or None
+ tools: List of ToolDefinition or None
+ **kwargs: Additional fields (temperature, max_tokens, top_p, timestamp, metadata)
+ """
+ self.output_contains_image = False
+
+ # Handle copy constructor: UserTurn(existing_turn)
+ if isinstance(content, UserTurn):
+ source = content
+ self.role = source.role
+ self.content = ContentBlockList(source.content) # Deep copy the content list
+ self.temperature = source.temperature
+ self.max_tokens = source.max_tokens
+ self.top_p = source.top_p
+ self.timestamp = source.timestamp
+ self.metadata = dict(source.metadata) # Copy the metadata dict
+ return
+
+ # Handle content
+ if content is None:
+ content = ContentBlockList()
+ elif not isinstance(content, ContentBlockList):
+ # If it's a list, wrap it in ContentBlockList
+ content = ContentBlockList(content) if isinstance(content, list) else ContentBlockList([content])
+
+
+ # Set all fields
+ self.role = kwargs.get('role', "user")
+ self.content = content
+ self.temperature = kwargs.get('temperature', None)
+ self.max_tokens = kwargs.get('max_tokens', None)
+ self.top_p = kwargs.get('top_p', None)
+ self.timestamp = kwargs.get('timestamp', None)
+ self.metadata = kwargs.get('metadata', {})
+
+ def add_text(self, text: str) -> 'UserTurn':
+ """Add text content"""
+ self.content.append(TextContent(text=text))
+ return self
+
+ def add_image(self, url: Optional[str] = None, data: Optional[str] = None,
+ media_type: str = "image/jpeg") -> 'UserTurn':
+ """Add image content"""
+ self.content.append(ImageContent(
+ image_url=url,
+ image_data=data,
+ media_type=media_type
+ ))
+ return self
+
+ def add_image_file(self, filepath: str) -> 'UserTurn':
+ """Add image from file"""
+ self.content.append(ImageContent.from_file(filepath))
+ return self
+
+ def to_dict(self) -> Dict[str, Any]:
+ """Convert to dictionary format"""
+ return {
+ "role": "user",
+ "content": [c.to_dict() for c in self.content],
+ "temperature": self.temperature,
+ "max_tokens": self.max_tokens,
+ "top_p": self.top_p,
+ "metadata": self.metadata
+ }
+
+ def enable_image_generation(self):
+ self.output_contains_image = True
+
+ def __repr__(self) -> str:
+ """Safe string representation that handles missing attributes."""
+ content_preview = str(self.content)[:50] + "..." if len(str(self.content)) > 50 else str(self.content)
+ parts = [f"UserTurn(content={content_preview!r}"]
+
+ # Safely add optional fields if they exist
+ temperature = getattr(self, 'temperature', None)
+ if temperature is not None:
+ parts.append(f", temperature={temperature}")
+
+ parts.append(")")
+ return "".join(parts)
+
+ def to_litellm_format(self) -> Dict[str, Any]:
+ """Convert to LiteLLM Response API format (OpenAI Response API compatible)"""
+ return {
+ "role": "user",
+ "content": self.content.to_litellm_format(role="user")
+ }
+
+ def _repr_html_(self) -> str:
+ """Rich HTML representation for Jupyter notebooks with glassmorphism design."""
+ try:
+ from opto.utils.display.jupyter import render_user_turn
+ return render_user_turn(self)
+ except ImportError:
+ # Fallback to text representation if display module unavailable
+ return None
+
+
+@dataclass
+class Turn:
+ def __init__(self, **kwargs):
+ for key, value in kwargs.items():
+ setattr(self, key, value)
+
+
+@dataclass
+class AssistantTurn(Turn):
+ """Represents an assistant message turn in the conversation"""
+ role: str = "assistant"
+ content: ContentBlockList = field(default_factory=ContentBlockList)
+
+ # Provider-specific features
+ reasoning: Optional[str] = None # OpenAI reasoning/thinking
+ finish_reason: Optional[str] = None # "stop", "length", "tool_calls", etc.
+
+ # Token usage
+ prompt_tokens: Optional[int] = None
+ completion_tokens: Optional[int] = None
+
+ # Metadata
+ model: Optional[str] = None
+ timestamp: Optional[str] = None
+ metadata: Dict[str, Any] = field(default_factory=dict)
+
+ def __init__(self, *args, **kwargs):
+ """
+ Initialize AssistantTurn from a raw response or with explicit fields.
+
+ Three ways to initialize:
+ 1. Empty: AssistantTurn() - creates empty turn with defaults
+ 2. From raw response: AssistantTurn(response) - autocasts the response
+ 3. With fields: AssistantTurn(role="assistant", content=[...]) - explicit fields
+ """
+ if len(args) == 1 and isinstance(args[0], AssistantTurn):
+ # Case: Copy constructor - create a copy of another AssistantTurn
+ other = args[0]
+ super().__init__(
+ role=other.role,
+ content=ContentBlockList(other.content),
+ reasoning=other.reasoning,
+ finish_reason=other.finish_reason,
+ prompt_tokens=other.prompt_tokens,
+ completion_tokens=other.completion_tokens,
+ model=other.model,
+ timestamp=other.timestamp,
+ metadata=dict(other.metadata)
+ )
+ return
+
+ if len(args) > 0 and len(kwargs) == 0:
+ # Case 2: Single positional arg - autocast from raw response
+ value_dict = self.autocast(args[0])
+ super().__init__(**value_dict)
+ elif len(kwargs) > 0:
+ # Case 3: Keyword arguments - use them directly
+ super().__init__(**kwargs)
+ else:
+ # Case 1: No arguments - initialize with defaults
+ super().__init__(
+ role="assistant",
+ content=ContentBlockList(),
+ reasoning=None,
+ finish_reason=None,
+ prompt_tokens=None,
+ completion_tokens=None,
+ model=None,
+ timestamp=None,
+ metadata={}
+ )
+
+ @staticmethod
+ def from_google_genai(value: Any) -> Dict[str, Any]:
+ """Parse a Google GenAI response into a dictionary of AssistantTurn fields.
+
+ Supports both the legacy generate_content API and the new Interactions API.
+
+ Args:
+ value: Raw response from Google GenAI API
+
+ Returns:
+ Dict[str, Any]: Dictionary with keys corresponding to AssistantTurn fields
+ """
+ # Initialize the result dictionary with default values
+ result = {
+ "role": "assistant",
+ "content": ContentBlockList(),
+ "reasoning": None,
+ "finish_reason": None,
+ "prompt_tokens": None,
+ "completion_tokens": None,
+ "model": None,
+ "timestamp": None,
+ "metadata": {}
+ }
+
+ # Check if this is a normalized response (from our GoogleGenAILLM)
+ if hasattr(value, 'raw_response'):
+ raw_response = value.raw_response
+ else:
+ raw_response = value
+
+ # Handle Interactions API format (new)
+ if hasattr(raw_response, 'outputs'):
+ # This is an Interaction object
+ interaction = raw_response
+
+ # Extract text from outputs
+ if interaction.outputs and len(interaction.outputs) > 0:
+ for output in interaction.outputs:
+ if hasattr(output, 'text') and output.text:
+ result["content"].append(TextContent(text=output.text))
+ # Handle other output types if they exist
+ elif hasattr(output, 'content'):
+ # Content could be a list of parts
+ if isinstance(output.content, list):
+ for part in output.content:
+ if hasattr(part, 'text') and part.text:
+ result["content"].append(TextContent(text=part.text))
+ else:
+ result["content"].append(TextContent(text=str(output.content)))
+
+ # Extract model info
+ if hasattr(interaction, 'model'):
+ result["model"] = interaction.model
+
+ # Extract status as finish_reason
+ if hasattr(interaction, 'status'):
+ result["finish_reason"] = interaction.status
+
+ # Extract token usage from Interactions API
+ if hasattr(interaction, 'usage'):
+ usage = interaction.usage
+ if hasattr(usage, 'input_tokens'):
+ result["prompt_tokens"] = usage.input_tokens
+ elif hasattr(usage, 'prompt_token_count'):
+ result["prompt_tokens"] = usage.prompt_token_count
+
+ if hasattr(usage, 'output_tokens'):
+ result["completion_tokens"] = usage.output_tokens
+ elif hasattr(usage, 'candidates_token_count'):
+ result["completion_tokens"] = usage.candidates_token_count
+
+ # Extract interaction ID as metadata
+ if hasattr(interaction, 'id'):
+ result["metadata"]['interaction_id'] = interaction.id
+
+ # Handle legacy generate_content API format
+ else:
+ # Extract thinking/reasoning (for Gemini 2.5+ models)
+ if hasattr(raw_response, 'thoughts') and raw_response.thoughts:
+ # Gemini's thinking budget feature
+ result["reasoning"] = str(raw_response.thoughts)
+
+ # Extract model info
+ if hasattr(raw_response, 'model_version'):
+ result["model"] = raw_response.model_version
+
+ # Extract token usage (if available)
+ if hasattr(raw_response, 'usage_metadata'):
+ usage = raw_response.usage_metadata
+ if hasattr(usage, 'prompt_token_count'):
+ result["prompt_tokens"] = usage.prompt_token_count
+ if hasattr(usage, 'candidates_token_count'):
+ result["completion_tokens"] = usage.candidates_token_count
+
+ # Handle multimodal content from Gemini (candidates with parts)
+ content_extracted = False
+ if hasattr(raw_response, 'candidates') and raw_response.candidates:
+ candidate = raw_response.candidates[0]
+
+ # Extract from parts (supports multimodal responses with text and images)
+ if hasattr(candidate, 'content') and hasattr(candidate.content, 'parts'):
+ for part in candidate.content.parts:
+ # Handle text parts
+ if hasattr(part, 'text') and part.text:
+ result["content"].append(TextContent(text=part.text))
+ content_extracted = True
+ # Handle inline data (images, generated images, etc.)
+ elif hasattr(part, 'inline_data'):
+ # Try to extract image data, preferring direct inline_data access
+ inline = part.inline_data
+ image_bytes = None
+ image_data = None
+ media_type = 'image/jpeg'
+
+
+ # Extract from inline_data Blob (most reliable method)
+ # Google's Blob.data should be raw bytes
+ if hasattr(inline, 'data'):
+ data = inline.data
+ # Check if it's bytes or string
+ if isinstance(data, bytes):
+ # Store raw bytes for Gemini compatibility
+ # (Gemini prefers raw bytes when sending images)
+ image_bytes = data
+ elif isinstance(data, str):
+ # Already base64-encoded string
+ image_data = data
+ # Don't decode to bytes - keep as base64 for portability
+
+ if hasattr(inline, 'mime_type'):
+ media_type = inline.mime_type
+
+ # If we got the data, create ImageContent
+ # Store image_bytes only if we got raw bytes from Google
+ if image_data or image_bytes:
+ result["content"].append(ImageContent(
+ image_data=image_data,
+ image_bytes=image_bytes if isinstance(data, bytes) else None,
+ media_type=media_type
+ ))
+ content_extracted = True
+
+ # Extract finish reason
+ if hasattr(candidate, 'finish_reason'):
+ result["finish_reason"] = str(candidate.finish_reason)
+
+ # Fallback: Extract simple text content if no candidates/parts were found
+ if not content_extracted:
+ if hasattr(raw_response, 'text'):
+ result["content"].append(TextContent(text=raw_response.text))
+ elif hasattr(value, 'choices'):
+ # Fallback to normalized format
+ result["content"].append(TextContent(text=value.choices[0].message.content))
+
+ return result
+
+ @staticmethod
+ def from_litellm_openai_response_api(value: Any) -> Dict[str, Any]:
+ """Parse a LiteLLM/OpenAI-style response into a dictionary of AssistantTurn fields.
+
+ Handles both formats:
+ - New Responses API: Has 'output' field with ResponseOutputMessage objects
+ - Legacy Completion API: Has 'choices' field with message objects
+
+ Args:
+ value: Response from LiteLLM/OpenAI API (Responses API or Completion API)
+
+ Returns:
+ Dict[str, Any]: Dictionary with keys corresponding to AssistantTurn fields
+ """
+ # Initialize the result dictionary with default values
+ result = {
+ "role": "assistant",
+ "content": ContentBlockList(),
+ "reasoning": None,
+ "finish_reason": None,
+ "prompt_tokens": None,
+ "completion_tokens": None,
+ "model": None,
+ "timestamp": None,
+ "metadata": {}
+ }
+
+ # Handle Bedrock Converse API format (has 'output' field with 'message')
+ # Check both attribute-based and dict-based access for robustness
+ is_bedrock = False
+ bedrock_output = None
+ bedrock_value = value # Keep reference to the original value for later access
+
+ # Try attribute-based access first
+ if hasattr(value, 'output'):
+ output_val = value.output
+
+ if hasattr(output_val, 'message'):
+ is_bedrock = True
+ bedrock_output = output_val
+ # Also check dict-based access on the output attribute
+ elif isinstance(output_val, dict) and 'message' in output_val:
+ is_bedrock = True
+ bedrock_output = output_val
+
+ # If not found, try dict-based access on value itself
+ if not is_bedrock and isinstance(value, dict) and 'output' in value:
+ output_val = value['output']
+ if isinstance(output_val, dict) and 'message' in output_val:
+ is_bedrock = True
+ bedrock_output = output_val
+ bedrock_value = value # Use the dict directly
+
+ if is_bedrock and bedrock_output is not None:
+ # Bedrock Converse API format detected
+ # Get message with dict or attr access
+ message = bedrock_output.get('message') if isinstance(bedrock_output, dict) else (bedrock_output.message if hasattr(bedrock_output, 'message') else None)
+
+ if message:
+ # Extract role
+ if isinstance(message, dict):
+ result["role"] = message.get('role', 'assistant')
+ elif hasattr(message, 'role'):
+ result["role"] = message.role
+
+ # Extract content
+ content_list = message.get('content') if isinstance(message, dict) else (message.content if hasattr(message, 'content') else None)
+
+ if content_list:
+ for content_item in content_list:
+ # Handle text content (dict or attr)
+ text_val = None
+ if isinstance(content_item, dict):
+ text_val = content_item.get('text')
+ elif hasattr(content_item, 'text'):
+ text_val = content_item.text
+
+ if text_val:
+ result["content"].append(TextContent(text=text_val))
+
+ # Extract finish reason from stopReason (check both value and bedrock_value)
+ stop_reason = None
+ if isinstance(bedrock_value, dict):
+ stop_reason = bedrock_value.get('stopReason')
+ elif hasattr(bedrock_value, 'stopReason'):
+ stop_reason = bedrock_value.stopReason
+ if stop_reason:
+ result["finish_reason"] = stop_reason
+
+ # Extract token usage (check both value and bedrock_value)
+ usage = None
+ if isinstance(bedrock_value, dict):
+ usage = bedrock_value.get('usage')
+ elif hasattr(bedrock_value, 'usage'):
+ usage = bedrock_value.usage
+
+ if usage:
+ if isinstance(usage, dict):
+ result["prompt_tokens"] = usage.get('inputTokens')
+ result["completion_tokens"] = usage.get('outputTokens')
+ else:
+ if hasattr(usage, 'inputTokens'):
+ result["prompt_tokens"] = usage.inputTokens
+ if hasattr(usage, 'outputTokens'):
+ result["completion_tokens"] = usage.outputTokens
+
+ # Handle Responses API format (new format with 'output' field)
+ # The output field is a list of output items (messages, image generation calls, etc.)
+ # NOTE: LiteLLM may set value.object to 'chat.completion' or 'response' depending on the provider
+ elif hasattr(value, 'output') and hasattr(value, 'object'):
+ # Extract metadata
+ if hasattr(value, 'id'):
+ result["metadata"]['response_id'] = value.id
+ if hasattr(value, 'created_at'):
+ result["timestamp"] = str(value.created_at)
+
+ # Extract model info
+ if hasattr(value, 'model'):
+ result["model"] = value.model
+
+ # Extract status as finish_reason
+ if hasattr(value, 'status'):
+ result["finish_reason"] = value.status
+
+ # Extract content from output (list of output items)
+ if value.output and len(value.output) > 0:
+ for output_item in value.output:
+ # Handle ImageGenerationCall
+ if hasattr(output_item, 'type') and output_item.type == 'image_generation_call':
+ # Extract generated image
+ if hasattr(output_item, 'result') and output_item.result:
+ # Determine media type from output_format
+ media_type = 'image/jpeg' # default
+ if hasattr(output_item, 'output_format'):
+ format_map = {
+ 'png': 'image/png',
+ 'jpeg': 'image/jpeg',
+ 'jpg': 'image/jpeg',
+ 'webp': 'image/webp',
+ 'gif': 'image/gif'
+ }
+ media_type = format_map.get(output_item.output_format.lower(), 'image/jpeg')
+
+ # Add image to content
+ result["content"].append(ImageContent(
+ image_data=output_item.result,
+ media_type=media_type
+ ))
+
+ # Store additional metadata about the image generation
+ if hasattr(output_item, 'revised_prompt') and output_item.revised_prompt:
+ if 'image_generation' not in result["metadata"]:
+ result["metadata"]['image_generation'] = []
+ result["metadata"]['image_generation'].append({
+ 'id': output_item.id if hasattr(output_item, 'id') else None,
+ 'revised_prompt': output_item.revised_prompt,
+ 'size': output_item.size if hasattr(output_item, 'size') else None,
+ 'quality': output_item.quality if hasattr(output_item, 'quality') else None,
+ 'status': output_item.status if hasattr(output_item, 'status') else None
+ })
+
+ # Handle ResponseOutputMessage
+ elif hasattr(output_item, 'type') and output_item.type == 'message':
+ # Extract role
+ if hasattr(output_item, 'role'):
+ result["role"] = output_item.role
+
+ # Extract status for this message
+ if hasattr(output_item, 'status') and not result["finish_reason"]:
+ result["finish_reason"] = output_item.status
+
+ # Extract content items
+ if hasattr(output_item, 'content') and output_item.content:
+ for content_item in output_item.content:
+ # Handle text content
+ if hasattr(content_item, 'type') and content_item.type == 'output_text':
+ if hasattr(content_item, 'text') and content_item.text:
+ result["content"].append(TextContent(text=content_item.text))
+ # Handle other content types as they become available
+ elif hasattr(content_item, 'text') and content_item.text:
+ result["content"].append(TextContent(text=str(content_item.text)))
+
+ # Extract reasoning (for models with reasoning capabilities)
+ if hasattr(value, 'reasoning'):
+ reasoning_parts = []
+ if isinstance(value.reasoning, dict):
+ if value.reasoning.get('summary'):
+ reasoning_parts.append(f"Summary: {value.reasoning['summary']}")
+ if value.reasoning.get('effort'):
+ reasoning_parts.append(f"Effort: {value.reasoning['effort']}")
+ if reasoning_parts:
+ result["reasoning"] = "\n".join(reasoning_parts)
+ elif value.reasoning:
+ result["reasoning"] = str(value.reasoning)
+
+ # Extract token usage (Responses API format)
+ if hasattr(value, 'usage'):
+ if hasattr(value.usage, 'input_tokens'):
+ result["prompt_tokens"] = value.usage.input_tokens
+ if hasattr(value.usage, 'output_tokens'):
+ result["completion_tokens"] = value.usage.output_tokens
+
+ # Handle legacy Completion API format (has 'choices' field)
+ elif hasattr(value, 'choices') and len(value.choices) > 0:
+ choice = value.choices[0]
+ message = choice.message if hasattr(choice, 'message') else choice
+
+ # Extract text content
+ if hasattr(message, 'content') and message.content:
+ result["content"].append(TextContent(text=str(message.content)))
+
+
+ # Extract finish reason
+ if hasattr(choice, 'finish_reason'):
+ result["finish_reason"] = choice.finish_reason
+
+ # Extract reasoning/thinking (for OpenAI o1/o3 models)
+ if hasattr(message, 'reasoning') and message.reasoning:
+ result["reasoning"] = message.reasoning
+
+ # Extract token usage (Completion API format)
+ if hasattr(value, 'usage'):
+ if hasattr(value.usage, 'prompt_tokens'):
+ result["prompt_tokens"] = value.usage.prompt_tokens
+ if hasattr(value.usage, 'completion_tokens'):
+ result["completion_tokens"] = value.usage.completion_tokens
+
+ # Extract model info
+ if hasattr(value, 'model'):
+ result["model"] = value.model
+
+ return result
+
+ @staticmethod
+ def autocast(value: Any) -> Dict[str, Any]:
+ """Automatically parse a response from any API into a dictionary of AssistantTurn fields.
+
+ Automatically detects the response format and uses the appropriate parser:
+ - Google GenAI (generate_content or Interactions API)
+ - LiteLLM/OpenAI Responses API (new format with 'output' field)
+ - LiteLLM/OpenAI Completion API (legacy format with 'choices' field)
+
+ Args:
+ value: Raw response from any supported API
+
+ Returns:
+ Dict[str, Any]: Dictionary with keys corresponding to AssistantTurn fields
+ """
+
+ # Check if this is a normalized response (from our GoogleGenAILLM)
+ raw_response = value.raw_response if hasattr(value, 'raw_response') else value
+
+ # Detect Google GenAI format (Interactions API or generate_content)
+ # Google GenAI has 'outputs' (Google Interactions API) or 'candidates' (generate_content)
+ # Note: 'outputs' is for Google's Interactions API, 'output' is for LiteLLM Responses API
+ if hasattr(raw_response, 'outputs') or \
+ (hasattr(raw_response, 'candidates') and not hasattr(value, 'choices')) or \
+ hasattr(raw_response, 'usage_metadata'):
+ return AssistantTurn.from_google_genai(value)
+
+ # Detect LiteLLM/OpenAI/Bedrock format (Responses API, Completion API, or Bedrock Converse)
+ # Responses API has 'output' field and object='response'
+ # Completion API has 'choices' field
+ # Bedrock Converse API has 'output' field with nested 'message'
+ # Check both attribute and dict-based access
+ has_output = hasattr(value, 'output') or (isinstance(value, dict) and 'output' in value)
+ has_choices = hasattr(value, 'choices') or (isinstance(value, dict) and 'choices' in value)
+
+ if has_output or has_choices:
+ return AssistantTurn.from_litellm_openai_response_api(value)
+
+ # Fallback: if has 'text' attribute, might be a simple Google response
+ elif hasattr(raw_response, 'text'):
+ return AssistantTurn.from_google_genai(value)
+
+ # Default to empty result if format is not recognized
+ else:
+ return {
+ "role": "assistant",
+ "content": ContentBlockList(),
+ "tool_calls": [],
+ "unparsed_tool_calls": [],
+ "tool_results": [],
+ "reasoning": None,
+ "finish_reason": None,
+ "prompt_tokens": None,
+ "completion_tokens": None,
+ "model": None,
+ "timestamp": None,
+ "metadata": {}
+ }
+
+ def add_text(self, text: str) -> 'AssistantTurn':
+ """Add text content"""
+ self.content.append(text)
+ return self
+
+ def add_image(self, url: Optional[str] = None, data: Optional[str] = None,
+ media_type: str = "image/jpeg") -> 'AssistantTurn':
+ """Add image content (some models can generate images)"""
+ self.content.append(ImageContent(
+ image_url=url,
+ image_data=data,
+ media_type=media_type
+ ))
+ return self
+
+ def to_text(self) -> str:
+ """Get all text content concatenated. Images will be presented as placeholder text."""
+ return self.content.to_text()
+
+ def to_dict(self) -> Dict[str, Any]:
+ """Convert to dictionary format"""
+ return {
+ "role": self.role,
+ "content": [c.to_dict() for c in self.content],
+ "reasoning": self.reasoning,
+ "finish_reason": self.finish_reason,
+ "prompt_tokens": self.prompt_tokens,
+ "completion_tokens": self.completion_tokens,
+ "model": self.model,
+ "metadata": self.metadata
+ }
+
+ def get_text(self) -> ContentBlockList:
+ """Get all text content blocks.
+
+ Returns:
+ ContentBlockList: List containing only TextContent blocks
+ """
+ text_blocks = ContentBlockList()
+ for block in self.content:
+ if isinstance(block, TextContent):
+ text_blocks.append(block)
+ return text_blocks
+
+ def get_images(self) -> ContentBlockList:
+ """Get all image content blocks.
+
+ Returns:
+ ContentBlockList: List containing only ImageContent blocks
+ """
+ image_blocks = ContentBlockList()
+ for block in self.content:
+ if isinstance(block, ImageContent):
+ image_blocks.append(block)
+ return image_blocks
+
+ def __repr__(self) -> str:
+ """Safe string representation that handles missing attributes."""
+ content_preview = str(self.content)[:50] + "..." if len(str(self.content)) > 50 else str(self.content)
+ parts = [f"AssistantTurn(content={content_preview!r}"]
+
+ # Safely add optional fields if they exist
+ if hasattr(self, 'model') and self.model:
+ parts.append(f", model={self.model!r}")
+ if hasattr(self, 'prompt_tokens') and self.prompt_tokens:
+ parts.append(f", prompt_tokens={self.prompt_tokens}")
+ if hasattr(self, 'completion_tokens') and self.completion_tokens:
+ parts.append(f", completion_tokens={self.completion_tokens}")
+
+ parts.append(")")
+ return "".join(parts)
+
+ def to_litellm_format(self) -> Dict[str, Any]:
+ """Convert to LiteLLM Response API format (OpenAI Response API compatible)"""
+ result = {"role": self.role}
+
+ # Handle content blocks (text, images, etc.) - delegate to ContentBlockList
+ result["content"] = self.content.to_litellm_format(role=self.role)
+
+ return result
+
+
+ def _repr_html_(self) -> str:
+ """Rich HTML representation for Jupyter notebooks with glassmorphism design."""
+ try:
+ from opto.utils.display.jupyter import render_assistant_turn
+ return render_assistant_turn(self)
+ except ImportError:
+ # Fallback to text representation if display module unavailable
+ return None
diff --git a/opto/utils/display/README.md b/opto/utils/display/README.md
new file mode 100644
index 00000000..5269568a
--- /dev/null
+++ b/opto/utils/display/README.md
@@ -0,0 +1,129 @@
+# Opto Display Module
+
+Rendering utilities for visualizing Opto objects in various formats.
+
+## Architecture
+
+The display module separates **visualization logic** from **data handling logic**:
+
+```
+opto/utils/
+āāā backbone.py # Core data classes (UserTurn, AssistantTurn, Chat)
+āāā display/
+ āāā __init__.py # Public API
+ āāā jupyter.py # Jupyter notebook HTML rendering
+ āāā themes.py # Color schemes and styling
+ āāā README.md # This file
+```
+
+## Usage
+
+### Automatic Display in Jupyter
+
+Objects automatically render with glassmorphism styling:
+
+```python
+from opto.utils.backbone import UserTurn, AssistantTurn, Chat
+
+user_turn = UserTurn("Hello!")
+user_turn # Beautiful display! āØ
+```
+
+### Direct Rendering
+
+Use the display module directly for more control:
+
+```python
+from opto.utils.display import render_user_turn, render_chat
+from IPython.display import HTML
+
+user_turn = UserTurn("Hello!")
+html_output = render_user_turn(user_turn)
+HTML(html_output)
+```
+
+## Features
+
+- Glassmorphism Design
+- Multi-modal content (images, files interleaved with text), all rendered as HTML.
+
+### Custom Themes
+
+You can customize the color scheme:
+
+```python
+from opto.utils.display.themes import set_theme
+
+custom_theme = {
+ 'user': {
+ 'background': 'rgba(255, 240, 245, 0.85)',
+ 'border': 'rgba(233, 30, 99, 0.3)',
+ 'text_color': '#C2185B',
+ 'icon': 'š¤',
+ },
+ 'assistant': {
+ 'background': 'rgba(232, 245, 233, 0.85)',
+ 'border': 'rgba(76, 175, 80, 0.3)',
+ 'text_color': '#388E3C',
+ 'icon': 'š¤',
+ },
+ # ... other theme properties
+}
+
+set_theme(custom_theme)
+```
+
+### Custom Renderers
+
+You can create your own rendering functions:
+
+```python
+def my_custom_renderer(user_turn):
+ """Custom HTML renderer for UserTurn"""
+ return f"{user_turn.content.to_text()}
"
+
+# Use it directly
+from IPython.display import HTML
+HTML(my_custom_renderer(user_turn))
+```
+
+### Fallback Behavior
+
+If the display module is unavailable (e.g., import error), classes fall back to their `__repr__()` text representation:
+
+```python
+class UserTurn:
+ def _repr_html_(self):
+ try:
+ from opto.utils.display.jupyter import render_user_turn
+ return render_user_turn(self)
+ except ImportError:
+ return None # Falls back to __repr__
+```
+
+## Future Extension
+
+Potential additions to the display module:
+
+- **Terminal renderer**: ANSI color codes for CLI display
+- **Markdown renderer**: Export conversations as markdown
+- **LaTeX renderer**: For academic papers
+- **HTML export**: Static HTML page generation
+
+To add a new rendering format:
+
+1. Create `opto/utils/display/my_format.py`
+2. Implement `render_*` functions for each class
+3. Export in `__init__.py`
+4. Update documentation
+
+Example:
+
+```python
+# opto/utils/display/terminal.py
+def render_user_turn(turn):
+ """Render UserTurn with ANSI colors for terminal"""
+ from colorama import Fore, Style
+ return f"{Fore.CYAN}š¤ User:{Style.RESET_ALL} {turn.content.to_text()}"
+```
+
diff --git a/opto/utils/display/__init__.py b/opto/utils/display/__init__.py
new file mode 100644
index 00000000..ce4ef555
--- /dev/null
+++ b/opto/utils/display/__init__.py
@@ -0,0 +1,27 @@
+"""
+Display utilities for rendering Opto objects in various formats.
+
+This module provides rendering functions for different output formats:
+- Jupyter notebooks (HTML)
+- Terminal/CLI
+- Markdown export
+
+Usage:
+ from opto.utils.display import render_for_jupyter
+ html = render_for_jupyter(user_turn)
+"""
+
+from .jupyter import (
+ render_user_turn,
+ render_assistant_turn,
+ render_chat,
+ render_content_block_list,
+)
+
+__all__ = [
+ 'render_user_turn',
+ 'render_assistant_turn',
+ 'render_chat',
+ 'render_content_block_list',
+]
+
diff --git a/opto/utils/display/jupyter.py b/opto/utils/display/jupyter.py
new file mode 100644
index 00000000..393e7fa1
--- /dev/null
+++ b/opto/utils/display/jupyter.py
@@ -0,0 +1,543 @@
+"""
+Jupyter notebook HTML rendering for Opto objects.
+
+This module handles all HTML generation for displaying Opto objects
+in Jupyter notebooks with glassmorphism styling.
+"""
+
+import html as html_module
+import uuid
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from opto.utils.backbone import UserTurn, AssistantTurn, Chat, ContentBlockList
+
+from .themes import get_theme
+
+
+def _escape(text: str) -> str:
+ """HTML escape helper."""
+ return html_module.escape(str(text))
+
+
+def _escape_with_linebreaks(text: str) -> str:
+ """HTML escape with proper newline and tab handling."""
+ # First escape HTML
+ escaped = html_module.escape(str(text))
+ # Convert newlines to
tags
+ escaped = escaped.replace('\n', '
')
+ # Convert tabs to 4 spaces (visible)
+ escaped = escaped.replace('\t', ' ')
+ return escaped
+
+
+def _render_image_block(block) -> str:
+ """Render an image content block."""
+ parts = []
+ parts.append('')
+
+ # Get image source - prioritize URL, then base64 data
+ img_src = None
+ if hasattr(block, 'image_url') and block.image_url:
+ img_src = block.image_url
+ elif hasattr(block, 'image_data') and block.image_data:
+ # Base64 encoded image
+ media_type = getattr(block, 'media_type', 'image/jpeg')
+ img_src = f"data:{media_type};base64,{block.image_data}"
+
+ theme = get_theme()
+ max_height = theme['common']['image_max_height']
+
+ if img_src:
+ parts.append(f'
})
')
+ else:
+ # Fallback to placeholder if no image source
+ media_type = getattr(block, 'media_type', 'image/jpeg')
+ parts.append('
š¼ļø')
+ parts.append(f'Image ({media_type})
')
+
+ parts.append('
+
+ {block_type}:
+
+
+ {_escape(block_text)}
+
+
+ ''')
+
+ return ''.join(inline_parts), ''.join(detail_parts), len(blocks), num_images
+
+
+def render_user_turn(turn: 'UserTurn') -> str:
+ """Render a UserTurn as Jupyter HTML."""
+ theme = get_theme()
+ user_theme = theme['user']
+ common = theme['common']
+
+ # Generate unique ID
+ turn_id = str(uuid.uuid4())[:8]
+
+ parts = []
+
+ # Main turn container
+ parts.append(f'''
+
+ ''')
+
+ # Role header
+ parts.append(f'''
+
+ {user_theme['icon']}
+ User
+
+ ''')
+
+ # Content
+ parts.append('
')
+
+ # Render content blocks
+ inline_html, detail_html, num_blocks, num_images = _render_content_blocks(turn.content, turn_id)
+ parts.append(inline_html)
+ parts.append('
')
+
+ # Metadata badges
+ parts.append('
')
+
+ # Content blocks badge
+ block_label = f"{num_blocks} content block{'s' if num_blocks != 1 else ''}"
+ if num_images > 0:
+ block_label += f" ({num_images} image{'s' if num_images != 1 else ''})"
+
+ parts.append(f'''
+
+ {block_label} ā¼
+
+ ''')
+
+ # Tools badge
+ tools = getattr(turn, 'tools', [])
+ if tools:
+ parts.append(f'''
+
+ š§ {len(tools)} tool{'s' if len(tools) != 1 else ''} available
+
+ ''')
+
+ # Settings badges
+ temperature = getattr(turn, 'temperature', None)
+ if temperature is not None:
+ parts.append(f'''
+
+ temperature: {temperature}
+
+ ''')
+
+ parts.append('
')
+
+ # Content blocks detail (expandable)
+ parts.append(f'''
+
+ ''')
+ parts.append(f'''
+
+ ''')
+ parts.append(detail_html)
+ parts.append('
')
+
+ # Close main container
+ parts.append('
+ ''')
+
+ # Role header with token badge
+ parts.append(f'''
+
+
+ {assistant_theme['icon']}
+ Assistant
+
+ ''')
+
+ # Token count badge
+ prompt_tokens = getattr(turn, 'prompt_tokens', None)
+ completion_tokens = getattr(turn, 'completion_tokens', None)
+ if prompt_tokens or completion_tokens:
+ total_tokens = (prompt_tokens or 0) + (completion_tokens or 0)
+ parts.append(f'''
+
+ š° {total_tokens} tokens
+
+ ''')
+
+ parts.append('
')
+
+ # Reasoning section (if present)
+ reasoning = getattr(turn, 'reasoning', None)
+ if reasoning:
+ reasoning_theme = theme['reasoning']
+ parts.append(f'''
+
+
+ {reasoning_theme['icon']} Reasoning:
+
+ {_escape(reasoning)}
+
+ ''')
+
+ # Content
+ parts.append('
')
+
+ # Render content blocks
+ inline_html, _, _, _ = _render_content_blocks(turn.content, turn_id)
+ parts.append(inline_html)
+ parts.append('
')
+
+ # Tool calls section (if present)
+ if hasattr(turn, 'tool_calls') and turn.tool_calls:
+ import json
+ tool_theme = theme['tool_calls']
+
+ parts.append(f'''
+
+
+ {tool_theme['icon']} Tool Calls:
+
+ ''')
+
+ for tc in turn.tool_calls:
+ func_name = tc.function.name if tc.function else "unknown"
+ args_str = tc.function.arguments if tc.function else "{}"
+ parts.append(f'''
+
+ {_escape(func_name)}
+ ({_escape(args_str)})
+
+ ''')
+
+ # Show results if available
+ tool_results = getattr(turn, 'tool_results', [])
+ matching_results = [tr for tr in tool_results if tr.tool_call_id == tc.id]
+ if matching_results:
+ for tr in matching_results:
+ result_content = str(tr.content)[:200]
+ if len(str(tr.content)) > 200:
+ result_content += '...'
+ parts.append(f'''
+
+ Result: {_escape(result_content)}
+
+ ''')
+
+ parts.append('
')
+
+ # Metadata badges
+ parts.append('
')
+
+ model = getattr(turn, 'model', None)
+ if model:
+ parts.append(f'''
+
+ model: {_escape(model)}
+
+ ''')
+
+ finish_reason = getattr(turn, 'finish_reason', None)
+ if finish_reason:
+ parts.append(f'''
+
+ finish: {_escape(finish_reason)}
+
+ ''')
+
+ parts.append('
')
+
+ # Close main container
+ parts.append('
+ ''')
+
+ # Header
+ parts.append(f'''
+
+ {content_theme['icon']}
+ Content Blocks
+
+ ''')
+
+ # Content
+ parts.append('
')
+
+ # Render content blocks
+ inline_html, detail_html, num_blocks, num_images = _render_content_blocks(blocks, list_id)
+ parts.append(inline_html)
+ parts.append('
')
+
+ # Metadata badges
+ parts.append('
')
+
+ block_label = f"{num_blocks} block{'s' if num_blocks != 1 else ''}"
+ if num_images > 0:
+ block_label += f" ({num_images} image{'s' if num_images != 1 else ''})"
+
+ parts.append(f'''
+
+ {block_label} ā¼
+
+ ''')
+
+ parts.append('
')
+
+ # Content blocks detail (expandable)
+ parts.append(f'''
+
+ ''')
+ parts.append(f'''
+
+ ''')
+ parts.append(detail_html)
+ parts.append('
')
+
+ # Close main container
+ parts.append('
+ ''')
+
+ # Chat header
+ parts.append(f'''
+
+
š¬ Chat History
+
+ ''')
+
+ # Stats badges
+ num_turns = len(chat.turns)
+ parts.append(f'''
+
+ {num_turns} turn{'s' if num_turns != 1 else ''}
+
+ ''')
+
+ total_tokens = chat.get_token_count_estimate()
+ if total_tokens > 0:
+ parts.append(f'''
+
+ Total: ~{total_tokens} tokens
+
+ ''')
+
+ parts.append('
')
+
+ # System prompt (if present)
+ if chat.system_prompt:
+ parts.append(f'''
+
+
+ {system_theme['icon']} System Prompt:
+
+ {_escape(chat.system_prompt)}
+
+ ''')
+
+ # Render turns with middle collapsing for long conversations
+ COLLAPSE_THRESHOLD = 20
+ SHOW_FIRST = 2
+ SHOW_LAST = 2
+
+ if len(chat.turns) > COLLAPSE_THRESHOLD:
+ # Show first few turns
+ for turn in chat.turns[:SHOW_FIRST]:
+ if hasattr(turn, '_repr_html_'):
+ parts.append(turn._repr_html_())
+
+ # Collapsed middle section
+ num_hidden = len(chat.turns) - SHOW_FIRST - SHOW_LAST
+ parts.append(f'''
+
+ ... {num_hidden} more turn{'s' if num_hidden != 1 else ''} ... (click to expand)
+
+ ''')
+
+ # Hidden middle turns
+ parts.append(f'
')
+ for turn in chat.turns[SHOW_FIRST:-SHOW_LAST]:
+ if hasattr(turn, '_repr_html_'):
+ parts.append(turn._repr_html_())
+ parts.append('
')
+
+ # Show last few turns
+ for turn in chat.turns[-SHOW_LAST:]:
+ if hasattr(turn, '_repr_html_'):
+ parts.append(turn._repr_html_())
+ else:
+ # Show all turns
+ for turn in chat.turns:
+ if hasattr(turn, '_repr_html_'):
+ parts.append(turn._repr_html_())
+
+ # Close main container
+ parts.append('