Storage

class afterimage.storage.BaseStorage(*args, **kwargs)

Bases: Protocol

Protocol defining the interface for storage implementations.

abstractmethod async asave_conversations(conversations: List[EvaluatedConversationWithContext | ConversationWithContext | BaseModel]) → None

abstractmethod async asave_documents(documents: List[Document]) → None

abstractmethod load_conversations(limit: int | None = None, offset: int | None = None) → List[ConversationWithContext]

abstractmethod load_documents(limit: int | None = None, offset: int | None = None) → List[Document]

abstractmethod save_conversations(conversations: List[EvaluatedConversationWithContext | ConversationWithContext | BaseModel]) → None

abstractmethod save_documents(documents: List[Document]) → None

class afterimage.storage.JSONLStorage(conversations_path: str | Path | None = None, documents_path: str | Path | None = None, encoding: str = 'utf-8', lock_timeout: int = 30)

Bases: BaseStorage

Stores conversations and documents in JSONL format.

async asave_conversations(conversations: List[EvaluatedConversationWithContext | ConversationWithContext | BaseModel]) → None

async asave_documents(documents: List[Document]) → None

load_conversations(limit: int | None = None, offset: int | None = None) → List[EvaluatedConversationWithContext]

Load conversations from JSONL file.

Parameters:

• limit – Maximum number of conversations to load

• offset – Number of conversations to skip

Returns:

List of conversations

load_documents(limit: int | None = None, offset: int | None = None) → List[Document]

Load documents from JSONL file.

Parameters:

• limit – Maximum number of documents to load

• offset – Number of documents to skip

Returns:

List of documents

save_conversations(conversations: List[EvaluatedConversationWithContext | ConversationWithContext | BaseModel]) → None

save_documents(documents: List[Document]) → None

class afterimage.storage.SQLStorage(url: str, conversations_table_name: str = 'conversations', documents_table_name: str = 'documents', metadata_fields: List[str] | None = None, batch_size: int = 100)

Bases: BaseStorage

Stores conversations and documents using SQLAlchemy.

async asave_conversations(conversations: List[EvaluatedConversationWithContext | ConversationWithContext | BaseModel]) → None

Save conversations to database asynchronously.

async asave_documents(documents: List[Document]) → None

load_conversations(limit: int | None = None, offset: int | None = None, filters: Dict[str, Any] | None = None, order_by: List[tuple] | None = None) → List[EvaluatedConversationWithContext]

Load conversations from database with filtering and sorting.

Parameters:

• limit – Maximum number of conversations to load

• offset – Number of conversations to skip

• filters – Dict of field-value pairs for filtering

• order_by – List of (field, direction) tuples for sorting

Returns:

List of conversations

load_documents(limit: int | None = None, offset: int | None = None, filters: Dict[str, Any] | None = None, order_by: List[tuple] | None = None) → List[Document]

Load documents from database with filtering and sorting.

Parameters:

• limit – Maximum number of documents to load

• offset – Number of documents to skip

• filters – Dict of field-value pairs for filtering

• order_by – List of (field, direction) tuples for sorting

Returns:

List of documents

save_conversations(conversations: List[ConversationWithContext | EvaluatedConversationWithContext]) → None

Save conversations to database.

Parameters:

conversations – List of conversations to save

save_documents(documents: List[Document]) → None