Module livekit.agents.llm.chat_context

Classes

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

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

Expand source code

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

Class variables

var frame : AudioFrame | list[AudioFrame]

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

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

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

Class variables

var messages : list[ChatMessage]

Methods

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

def copy(self) ‑> ChatContext

class ChatImage (image: str | rtc.VideoFrame, inference_width: int | None = None, inference_height: int | None = None)

ChatImage(image: 'str | rtc.VideoFrame', inference_width: 'int | None' = None, inference_height: 'int | None' = None)

Expand source code

@dataclass
class ChatImage:
    image: str | rtc.VideoFrame
    inference_width: int | None = None
    inference_height: int | None = None
    _cache: dict[Any, Any] = field(default_factory=dict, repr=False, init=False)
    """_cache is used  by LLM implementations to store a processed version of the image
    for later use.
    """

Class variables

var image : str | VideoFrame

var inference_height : int | None

var inference_width : int | None

class ChatMessage (role: ChatRole, id: str | None = None, 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)

ChatMessage(role: 'ChatRole', id: 'str | None' = None, 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 | None = None  # 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],
    ) -> "ChatMessage":
        return ChatMessage(role="assistant", tool_calls=called_functions)

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

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

            return ChatMessage(role=role, content=content)

    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,
            name=self.name,
            content=content,
            tool_calls=tool_calls,
            tool_call_id=self.tool_call_id,
        )
        copied_msg._metadata = self._metadata
        return copied_msg

Class variables

var content : Union[str, ChatImage, ChatAudio, list[Union[str, ChatImage, ChatAudio]], ForwardRef(None)]

var id : str | None

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') ‑> ChatMessage

def create_tool_calls(called_functions: list[function_context.FunctionCallInfo]) ‑> ChatMessage

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

Methods

def copy(self)