​
marvin.thread

Thread management for conversations.

This module provides the Thread class for managing conversation context.

​
Classes

​
LLMCall

class LLMCall(id: uuid.UUID, thread_id: str, usage: Usage, timestamp: datetime)

Represents an LLM call.

Methods:

  • get_messages 
    def get_messages(self) -> LLMCallMessages
    Get the messages for this LLM call.
  • get_messages_async 
    def get_messages_async(self) -> LLMCallMessages
    Get the messages for this LLM call.

​
LLMCallMessages

class LLMCallMessages(prompt: list[Message], completion: list[Message])

​
Message

class Message(id: uuid.UUID = uuid.uuid4(), thread_id: str = None, message: PydanticAIMessage, created_at: datetime = utc_now())

​
Thread

class Thread(id: str = lambda: str(uuid.uuid4())(), parent_id: str | None = None)

Main runtime object for managing conversation context.

Methods:

  • add_agent_message

    def add_agent_message(self, message: str) -> Message

    Add an agent message to the thread.

  • add_agent_message_async

    def add_agent_message_async(self, message: str) -> Message

    Add an agent message to the thread.

  • add_info_message

    def add_info_message(self, message: str, prefix: str = None) -> Message

    Add an info message to the thread.

  • add_info_message_async

    def add_info_message_async(self, message: str, prefix: str = None) -> Message

    Add an info message to the thread.

  • add_messages

    def add_messages(self, messages: list[PydanticAIMessage]) -> list[Message]

    Add multiple messages to the thread.

    Args: messages: List of messages to add (UserMessage, AssistantMessage, etc.) llm_call_id: Optional ID of the LLM call that generated these messages

  • add_messages_async

    def add_messages_async(self, messages: list[PydanticAIMessage]) -> list[Message]

    Add multiple messages to the thread.

    Args: messages: List of messages to add (UserMessage, AssistantMessage, etc.) llm_call_id: Optional ID of the LLM call that generated these messages

  • add_system_message

    def add_system_message(self, message: str) -> Message

    Add a system message to the thread.

  • add_system_message_async

    def add_system_message_async(self, message: str) -> Message

    Add a system message to the thread.

  • add_user_message

    def add_user_message(self, message: str | Sequence[UserContent]) -> Message

    Add a user message to the thread.

  • add_user_message_async

    def add_user_message_async(self, message: str | Sequence[UserContent]) -> Message

    Add a user message to the thread.

  • get_current

    def get_current(cls) -> Optional[Thread]

    Get the current thread from context.

  • get_llm_calls

    def get_llm_calls(self, before: datetime | None = None, after: datetime | None = None, limit: int | None = None) -> list[LLMCall]

    Get LLM calls for this thread.

    Args: before: Only return calls before this timestamp after: Only return calls after this timestamp limit: Maximum number of calls to return

    Returns: List of LLM calls in chronological order

  • get_llm_calls_async

    def get_llm_calls_async(self, before: datetime | None = None, after: datetime | None = None, limit: int | None = None) -> list[LLMCall]

    Get LLM calls for this thread.

    Args: before: Only return calls before this timestamp after: Only return calls after this timestamp limit: Maximum number of calls to return

    Returns: List of LLM calls in chronological order

  • get_messages

    def get_messages(self, before: datetime | None = None, after: datetime | None = None, limit: int | None = None, include_system_messages: bool = False) -> list[Message]

    Get all messages in this thread.

    Args: before: Only return messages before this timestamp after: Only return messages after this timestamp limit: Maximum number of messages to return include_system_messages: Whether to include system messages Returns: List of messages in chronological order

  • get_messages_async

    def get_messages_async(self, before: datetime | None = None, after: datetime | None = None, limit: int | None = None, include_system_messages: bool = False) -> list[Message]

    Get all messages in this thread.

    Args: before: Only return messages before this timestamp after: Only return messages after this timestamp limit: Maximum number of messages to return include_system_messages: Whether to include system messages Returns: List of messages in chronological order

  • get_usage

    def get_usage(self, before: datetime | None = None, after: datetime | None = None) -> Usage

    Get the usage for this thread.

    Args: before: Only include usage before this timestamp after: Only include usage after this timestamp

    Returns: Total usage for the specified time range

  • get_usage_async

    def get_usage_async(self, before: datetime | None = None, after: datetime | None = None) -> Usage

    Get the usage for this thread.

    Args: before: Only include usage before this timestamp after: Only include usage after this timestamp

    Returns: Total usage for the specified time range

​
Functions

​
get_current_thread

def get_current_thread() -> Thread | None

Get the currently active thread from context.

Returns: The current Thread instance or None if no thread is active.

​
get_last_thread

def get_last_thread() -> Thread | None

Get the last thread that was set as current.

This function is intended for debugging purposes only, and will only work in certain contexts where the last thread is available in memory.

​
get_thread

def get_thread(thread: Thread | str | None) -> Thread

Get a thread from the given input.

Args: thread: Thread instance, thread ID, or None

Returns: A Thread instance

Parent Module: marvin