dataflow.persistence.conversations¶

Conversation persistence for the Haive framework.

This module provides functionality for persisting and retrieving conversations between users and agents. It includes classes for managing conversation metadata, messages, and the overall conversation lifecycle.

The conversation system uses Supabase as the storage backend, providing reliable persistence with proper authentication and access control. It supports operations like creating conversations, adding messages, retrieving history, and deleting conversations.

Typical usage example:

from haive.dataflow.persistence.conversations import ConversationManager, ConversationMetadata

# Create a conversation manager manager = ConversationManager()

# Create a new conversation conversation_id = await manager.create_conversation(

user_id=”user-123”, metadata=ConversationMetadata(

agent_id=”agent-456”, title=”Technical Support”, tags=[“support”, “technical”]

)

)

# Add messages to the conversation await manager.add_message(

conversation_id=conversation_id, content=”How do I reset my password?”, role=”user”, user_id=”user-123”

)

# Get conversation messages messages = await manager.get_messages(conversation_id)

Classes¶

ConversationManager

Manager for conversations with LangGraph integration.

ConversationMetadata

Metadata for a conversation.

Module Contents¶

class dataflow.persistence.conversations.ConversationManager¶

Manager for conversations with LangGraph integration.

This class provides methods for creating, retrieving, updating, and deleting conversations and their messages. It integrates with LangGraph for storing agent state and conversation history, and uses Supabase as the persistence backend.

The conversation manager handles: - Creating and retrieving conversations - Adding and retrieving messages - Managing conversation metadata - Integrating with LangGraph for state persistence - Enforcing access control

supabase_config¶

Configuration for the Supabase connection

persistence¶

Adapter for Supabase persistence

_client¶

Lazy-loaded Supabase client instance

Initialize the conversation manager.

Sets up the Supabase configuration and persistence adapter. The actual Supabase client is lazy-loaded when first needed.

async create_conversation(user_id, metadata)¶

Create a new conversation.

Parameters:
Returns:

Conversation details if created successfully, None otherwise

Return type:

dict[str, Any] | None

async get_conversation(thread_id, user_id)¶

Get conversation details.

Parameters:

  • thread_id (str) – Thread ID

  • user_id (str) – User ID

Returns:

Conversation details if found, None otherwise

Return type:

dict[str, Any] | None

async list_conversations(user_id, limit=20, offset=0)¶

List conversations for a user.

Parameters:

  • user_id (str) – User ID

  • limit (int) – Maximum number of conversations to return

  • offset (int) – Offset for pagination

Returns:

List of conversations

Return type:

list[dict[str, Any]]

property client¶

Lazy-loaded Supabase admin client.

This property provides access to the Supabase client, initializing it on first access if needed. It uses the service role key to ensure administrative access to the database.

Returns:

The initialized Supabase client instance

Note

This uses the service role key, which has elevated privileges. Use with caution and ensure proper access control.

class dataflow.persistence.conversations.ConversationMetadata(/, **data)¶

Bases: pydantic.BaseModel

Metadata for a conversation.

This model defines the metadata associated with a conversation, including the agent used, title, description, and custom data. It provides a structured way to store and retrieve conversation context.

Parameters:

data (Any)

agent_id¶

ID of the agent associated with the conversation

title¶

Optional title for the conversation

description¶

Optional detailed description of the conversation

tags¶

Optional list of tags for categorizing the conversation

custom_data¶

Optional dictionary of additional custom data

Examples

>>> metadata = ConversationMetadata(
...     agent_id="agent-123",
...     title="Customer Support",
...     tags=["support", "billing"],
...     custom_data={"priority": "high"}
... )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.