MessageStore (Base Interface)
1. Behavior and Context
In the jazzmine architecture, MessageStore acts as the primary data access layer. Every component that needs to persist or recall data—such as the Agent, MessageEnhancer, and ConversationSummarizer—interacts with an implementation of this class.
By using an abstract base, jazzmine allows developers to switch between a simple JSON file for local development and a high-performance database like PostgreSQL or MongoDB for production by simply changing a configuration dictionary.
2. Purpose
- Backend Agnosticism: Decouples the framework logic from specific database implementations.
- Data Integrity: Standardizes how conversations and messages are structured across different storage engines.
- Observability Storage: Provides specialized methods to handle TurnTrace objects, which are critical for auditing agent behavior and costs.
- Session Management: Enables multi-user support by organizing data around user_id and conversation_id.
3. High-Level API
The MessageStore is intended to be created using the initialize factory method. You do not instantiate the subclasses directly in standard application code.
Example: Initializing a Store
from jazzmine.core.message_store import MessageStore
# Configuration for a PostgreSQL backend
config = {
"storage": "postgres",
"dsn": "postgresql://user:pass@localhost/db",
"pool_min": 2,
"pool_max": 10
}
# Factory returns the concrete PostgresMessageStore instance
store = await MessageStore.initialize(config)
# Store a message
await store.store_message(my_message_object)
# Cleanup on shutdown
await store.close()5. Detailed Functionality
initialize(config: dict) [Class Method]
Functionality: A static factory that instantiates and initializes the correct concrete storage implementation based on the storage key in the config.
Parameters:
- config (dict): A dictionary containing:
- storage (str): "json", "postgres", or "mongodb".
- Backend specifics: path (for JSON), dsn (for Postgres), or uri/db (for MongoDB).
How it works: It maps the storage string to the internal class list, instantiates the class with the config, and calls the internal _init() method to establish connections or create tables.
get_immediate_context(conversation_id, n) [Abstract]
Functionality: Retrieves the most recent "sliding window" of messages for an LLM prompt.
Parameters:
- conversation_id (str): The ID of the chat session.
- n (int): The number of messages to retrieve.
How it works: The implementation must fetch the n most recent messages where is_flagged is False, and return them in chronological order (oldest to newest). This is critical for maintaining the logical flow of a transcript.
store_turn_trace(trace) [Abstract]
Functionality: Persists a comprehensive observability record for a single agent turn.
Parameters:
- trace (TurnTrace): The Pydantic model containing LLM calls, tool execution logs, and reasoning steps.
How it works: This is typically called after the agent has generated a response. It links to the assistant message via message_id.
Conversation Management Methods
| Method | Purpose |
|---|---|
| store_conversation(conversation) | Saves high-level metadata (title, user_id, agent_id). |
| get_conversations_by_user_id(user_id) | Returns all chat sessions for a specific user, sorted by recent activity. |
| delete_conversation(conversation_id) | Performs a cascading delete of the session, all messages, and all traces. |
Message Management Methods
| Method | Purpose |
|---|---|
| store_message(message) | Saves a single turn. Idempotent based on ID. |
| store_messages(list) | Bulk insertion for migration or high-frequency logging. |
| flag_message(message_id) | Marks a message as toxic or irrelevant so it's ignored by context retrieval. |
| get_messages_by_conversation_id(id) | Returns the full transcript of a conversation. |
Turn Trace Retrieval Methods
| Method | Purpose |
|---|---|
| get_turn_trace_by_message_id(id) | Fetches the "thoughts" and logs behind a specific AI reply. |
| get_turn_traces_by_agent_id(id, ...) | Powers administrative dashboards for cost and latency monitoring. |
6. Error Handling
- ValueError: Raised by initialize if an unsupported storage type is requested.
- Implementation Errors: Subclasses are responsible for raising backend-specific errors (e.g., ConnectionError for databases).
- KeyError: Raised internally by the _require helper if a requested conversation_id does not exist in the store.
7. Remarks
- Idempotency: All store_* methods follow "Upsert" semantics. If you call them with an ID that already exists, the record is updated in place.
- Async Interface: The entire MessageStore API is asynchronous. You must await every method call.
- Resource Management: You must call await store.close() during your application's shutdown sequence to ensure connection pools are released and file handles are closed properly.