Skip to main content

​
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