Module livekit.agents.llm.chat_context

Classes

class ChatAudio (frame: rtc.AudioFrame | list[rtc.AudioFrame])

Expand source code

@dataclass
class ChatAudio:
    frame: rtc.AudioFrame | list[rtc.AudioFrame]

ChatAudio(frame: 'rtc.AudioFrame | list[rtc.AudioFrame]')

Class variables

var frame : AudioFrame | list[AudioFrame]

class ChatContext (messages: list[ChatMessage] = <factory>)

Expand source code

@dataclass
class ChatContext:
    messages: list[ChatMessage] = field(default_factory=list)
    _metadata: dict[str, Any] = field(default_factory=dict, repr=False, init=False)

    def append(
        self, *, text: str = "", images: list[ChatImage] = [], role: ChatRole = "system"
    ) -> ChatContext:
        self.messages.append(ChatMessage.create(text=text, images=images, role=role))
        return self

    def copy(self) -> ChatContext:
        copied_chat_ctx = ChatContext(messages=[m.copy() for m in self.messages])
        copied_chat_ctx._metadata = self._metadata
        return copied_chat_ctx

ChatContext(messages: 'list[ChatMessage]' = )

Class variables

var messages : list[ChatMessage]

Methods

def append(self,
*,
text: str = '',
images: list[ChatImage] = [],
role: ChatRole = 'system') ‑> ChatContext

Expand source code

def append(
    self, *, text: str = "", images: list[ChatImage] = [], role: ChatRole = "system"
) -> ChatContext:
    self.messages.append(ChatMessage.create(text=text, images=images, role=role))
    return self

def copy(self) ‑> ChatContext

Expand source code

def copy(self) -> ChatContext:
    copied_chat_ctx = ChatContext(messages=[m.copy() for m in self.messages])
    copied_chat_ctx._metadata = self._metadata
    return copied_chat_ctx

class ChatImage (image: str | rtc.VideoFrame,
inference_width: int | None = None,
inference_height: int | None = None,
inference_detail: "Literal['auto', 'high', 'low']" = 'auto')

Expand source code

@dataclass
class ChatImage:
    """
    ChatImage is used to input images into the ChatContext on supported LLM providers / plugins.

    You may need to consult your LLM provider's documentation on supported URL types.

    ```python
    # Pass a VideoFrame directly, which will be automatically converted to a JPEG data URL internally
    async for event in rtc.VideoStream(video_track):
        chat_image = ChatImage(image=event.frame)
        # this instance is now available for your ChatContext

    # Encode your VideoFrame yourself for more control, and pass the result as a data URL (see EncodeOptions for more details)
    from livekit.agents.utils.images import encode, EncodeOptions, ResizeOptions

    image_bytes = encode(
        event.frame,
        EncodeOptions(
            format="PNG",
            resize_options=ResizeOptions(
                width=512, height=512, strategy="scale_aspect_fit"
            ),
        ),
    )
    chat_image = ChatImage(
        image=f"data:image/png;base64,{base64.b64encode(image_bytes).decode('utf-8')}"
    )

    # With an external URL
    chat_image = ChatImage(image="https://example.com/image.jpg")
    ```
    """

    image: str | rtc.VideoFrame
    """
    Either a string URL or a VideoFrame object
    """
    inference_width: int | None = None
    """
    Resizing parameter for rtc.VideoFrame inputs (ignored for URL images)
    """
    inference_height: int | None = None
    """
    Resizing parameter for rtc.VideoFrame inputs (ignored for URL images)
    """
    inference_detail: Literal["auto", "high", "low"] = "auto"
    """
    Detail parameter for LLM provider, if supported.
    
    Currently only supported by OpenAI (see https://platform.openai.com/docs/guides/vision?lang=node#low-or-high-fidelity-image-understanding)
    """
    _cache: dict[Any, Any] = field(default_factory=dict, repr=False, init=False)
    """
    _cache is used internally by LLM implementations to store a processed version of the image
    for later use.
    """

ChatImage is used to input images into the ChatContext on supported LLM providers / plugins.

You may need to consult your LLM provider's documentation on supported URL types.

# Pass a VideoFrame directly, which will be automatically converted to a JPEG data URL internally
async for event in rtc.VideoStream(video_track):
    chat_image = ChatImage(image=event.frame)
    # this instance is now available for your ChatContext

# Encode your VideoFrame yourself for more control, and pass the result as a data URL (see EncodeOptions for more details)
from livekit.agents.utils.images import encode, EncodeOptions, ResizeOptions

image_bytes = encode(
    event.frame,
    EncodeOptions(
        format="PNG",
        resize_options=ResizeOptions(
            width=512, height=512, strategy="scale_aspect_fit"
        ),
    ),
)
chat_image = ChatImage(
    image=f"data:image/png;base64,{base64.b64encode(image_bytes).decode('utf-8')}"
)

# With an external URL
chat_image = ChatImage(image="https://example.com/image.jpg")

Class variables

var image : str | VideoFrame

Either a string URL or a VideoFrame object

var inference_detail : Literal['auto', 'high', 'low']

Detail parameter for LLM provider, if supported.

Currently only supported by OpenAI (see https://platform.openai.com/docs/guides/vision?lang=node#low-or-high-fidelity-image-understanding)

var inference_height : int | None

Resizing parameter for rtc.VideoFrame inputs (ignored for URL images)

var inference_width : int | None

Resizing parameter for rtc.VideoFrame inputs (ignored for URL images)

class ChatMessage (role: ChatRole,
id: str = <factory>,
name: str | None = None,
content: ChatContent | list[ChatContent] | None = None,
tool_calls: list[function_context.FunctionCallInfo] | None = None,
tool_call_id: str | None = None,
tool_exception: Exception | None = None)

Expand source code

@dataclass
class ChatMessage:
    role: ChatRole
    id: str = field(
        default_factory=lambda: utils.shortuuid("item_")
    )  # used by the OAI realtime API
    name: str | None = None
    content: ChatContent | list[ChatContent] | None = None
    tool_calls: list[function_context.FunctionCallInfo] | None = None
    tool_call_id: str | None = None
    tool_exception: Exception | None = None
    _metadata: dict[str, Any] = field(default_factory=dict, repr=False, init=False)

    @staticmethod
    def create_tool_from_called_function(
        called_function: function_context.CalledFunction,
    ) -> "ChatMessage":
        if not called_function.task.done():
            raise ValueError("cannot create a tool result from a running ai function")

        tool_exception: Exception | None = None
        try:
            content = called_function.task.result()
        except BaseException as e:
            if isinstance(e, Exception):
                tool_exception = e
            content = f"Error: {e}"

        return ChatMessage(
            role="tool",
            name=called_function.call_info.function_info.name,
            content=content,
            tool_call_id=called_function.call_info.tool_call_id,
            tool_exception=tool_exception,
        )

    @staticmethod
    def create_tool_calls(
        called_functions: list[function_context.FunctionCallInfo],
        *,
        text: str = "",
    ) -> "ChatMessage":
        return ChatMessage(role="assistant", tool_calls=called_functions, content=text)

    @staticmethod
    def create(
        *,
        text: str = "",
        images: list[ChatImage] = [],
        role: ChatRole = "system",
        id: str | None = None,
    ) -> "ChatMessage":
        id = id or utils.shortuuid("item_")
        if len(images) == 0:
            return ChatMessage(role=role, content=text, id=id)
        else:
            content: list[ChatContent] = []
            if text:
                content.append(text)

            if len(images) > 0:
                content.extend(images)

            return ChatMessage(role=role, content=content, id=id)

    def copy(self):
        content = self.content
        if isinstance(content, list):
            content = content.copy()

        tool_calls = self.tool_calls
        if tool_calls is not None:
            tool_calls = tool_calls.copy()

        copied_msg = ChatMessage(
            role=self.role,
            id=self.id,
            name=self.name,
            content=content,
            tool_calls=tool_calls,
            tool_call_id=self.tool_call_id,
        )
        copied_msg._metadata = self._metadata
        return copied_msg

ChatMessage(role: 'ChatRole', id: 'str' = , name: 'str | None' = None, content: 'ChatContent | list[ChatContent] | None' = None, tool_calls: 'list[function_context.FunctionCallInfo] | None' = None, tool_call_id: 'str | None' = None, tool_exception: 'Exception | None' = None)

Class variables

var content : str | ChatImage | ChatAudio | list[str | ChatImage | ChatAudio] | None

var id : str

var name : str | None

var role : Literal['system', 'user', 'assistant', 'tool']

var tool_call_id : str | None

var tool_calls : list[FunctionCallInfo] | None

var tool_exception : Exception | None

Static methods

def create(*,
text: str = '',
images: list[ChatImage] = [],
role: ChatRole = 'system',
id: str | None = None) ‑> ChatMessage

Expand source code

@staticmethod
def create(
    *,
    text: str = "",
    images: list[ChatImage] = [],
    role: ChatRole = "system",
    id: str | None = None,
) -> "ChatMessage":
    id = id or utils.shortuuid("item_")
    if len(images) == 0:
        return ChatMessage(role=role, content=text, id=id)
    else:
        content: list[ChatContent] = []
        if text:
            content.append(text)

        if len(images) > 0:
            content.extend(images)

        return ChatMessage(role=role, content=content, id=id)

def create_tool_calls(called_functions: list[function_context.FunctionCallInfo], *, text: str = '') ‑> ChatMessage

Expand source code

@staticmethod
def create_tool_calls(
    called_functions: list[function_context.FunctionCallInfo],
    *,
    text: str = "",
) -> "ChatMessage":
    return ChatMessage(role="assistant", tool_calls=called_functions, content=text)

def create_tool_from_called_function(called_function: function_context.CalledFunction) ‑> ChatMessage

Expand source code

@staticmethod
def create_tool_from_called_function(
    called_function: function_context.CalledFunction,
) -> "ChatMessage":
    if not called_function.task.done():
        raise ValueError("cannot create a tool result from a running ai function")

    tool_exception: Exception | None = None
    try:
        content = called_function.task.result()
    except BaseException as e:
        if isinstance(e, Exception):
            tool_exception = e
        content = f"Error: {e}"

    return ChatMessage(
        role="tool",
        name=called_function.call_info.function_info.name,
        content=content,
        tool_call_id=called_function.call_info.tool_call_id,
        tool_exception=tool_exception,
    )

Methods

def copy(self)

Expand source code

def copy(self):
    content = self.content
    if isinstance(content, list):
        content = content.copy()

    tool_calls = self.tool_calls
    if tool_calls is not None:
        tool_calls = tool_calls.copy()

    copied_msg = ChatMessage(
        role=self.role,
        id=self.id,
        name=self.name,
        content=content,
        tool_calls=tool_calls,
        tool_call_id=self.tool_call_id,
    )
    copied_msg._metadata = self._metadata
    return copied_msg