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)

Attributes

logger

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

async create_conversation(user_id: str, metadata: ConversationMetadata) → dict[str, Any] | None

Create a new conversation.

Parameters:

• user_id – User ID

• metadata – Conversation metadata

Returns:

Conversation details if created successfully, None otherwise

async get_conversation(thread_id: str, user_id: str) → dict[str, Any] | None

Get conversation details.

Parameters:

• thread_id – Thread ID

• user_id – User ID

Returns:

Conversation details if found, None otherwise

async list_conversations(user_id: str, limit: int = 20, offset: int = 0) → list[dict[str, Any]]

List conversations for a user.

Parameters:

• user_id – User ID

• limit – Maximum number of conversations to return

• offset – Offset for pagination

Returns:

List of conversations

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.

persistence

supabase_config

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

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.

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"}
... )

agent_id: str

custom_data: dict[str, Any] | None = None

description: str | None = None

tags: list[str] | None = None

title: str | None = None

dataflow.persistence.conversations.logger