Stockage PostgreSQL

L’implémentation du stockage PostgreSQL fournit une solution de stockage prête pour la production qui utilise des bases de données PostgreSQL.

Installation

npm

npm install @mastra/pg@latest

pnpm

pnpm add @mastra/pg@latest

Yarn

yarn add @mastra/pg@latest

Bun

bun add @mastra/pg@latest

Utilisation

import { PostgresStore } from '@mastra/pg'

const storage = new PostgresStore({
  id: 'pg-storage',
  connectionString: process.env.DATABASE_URL,
})

Paramètres

id:

string

Identifiant unique de cette instance de stockage.

connectionString?:

string

Chaîne de connexion PostgreSQL (par exemple, postgresql://user:pass@host:5432/dbname). Requise sauf si vous utilisez pool ou les paramètres individuels de l’hôte (host, port, database, user, password).

host?:

string

Nom d’hôte ou adresse IP du serveur de base de données. Utilisé avec les autres paramètres de l’hôte comme alternative à connectionString.

port?:

number

Numéro de port du serveur de base de données. Sa valeur par défaut est 5432 s’il n’est pas spécifié.

database?:

string

Nom de la base de données à laquelle se connecter.

user?:

string

Utilisateur de la base de données pour l’authentification.

password?:

string

Mot de passe de l’utilisateur de la base de données.

pool?:

pg.Pool

Instance pg.Pool préconfigurée. Utilisez-la pour réutiliser un pool de connexions existant. Lorsqu’elle est fournie, Mastra ne crée pas son propre pool et ne le ferme pas lors de l’appel de store.close().

schemaName?:

string

Nom du schéma que le stockage doit utiliser. Sa valeur par défaut est 'public'.

ssl?:

boolean | ConnectionOptions

Configuration SSL de la connexion ; définissez-la sur true pour utiliser le SSL par défaut, ou fournissez un objet ConnectionOptions pour des paramètres SSL personnalisés.

max?:

number

Nombre maximal de connexions dans le pool. Sa valeur par défaut est 20.

idleTimeoutMillis?:

number

Durée pendant laquelle une connexion peut rester inactive avant d’être fermée. Sa valeur par défaut est 30000 (30 secondes).

disableInit?:

boolean

Lorsque cette option vaut true, la création automatique des tables et les migrations sont désactivées. Utile pour les pipelines CI/CD dans lesquels les migrations sont exécutées séparément.

skipDefaultIndexes?:

boolean

Lorsque cette option vaut true, les index par défaut ne sont pas créés pendant l’initialisation.

indexes?:

CreateIndexOptions[]

Index personnalisés à créer pendant l’initialisation.

Exemples de constructeur

Vous pouvez instancier PostgresStore des manières suivantes :

import { PostgresStore } from '@mastra/pg'
import { Pool } from 'pg'

// Using a connection string
const store1 = new PostgresStore({
  id: 'pg-storage-1',
  connectionString: 'postgresql://user:password@localhost:5432/mydb',
})

// Using a connection string with pool options
const store2 = new PostgresStore({
  id: 'pg-storage-2',
  connectionString: 'postgresql://user:password@localhost:5432/mydb',
  schemaName: 'custom_schema',
  max: 30, // Max pool connections
  idleTimeoutMillis: 60000, // Idle timeout
  ssl: { rejectUnauthorized: false },
})

// Using individual connection parameters
const store3 = new PostgresStore({
  id: 'pg-storage-3',
  host: 'localhost',
  port: 5432,
  database: 'mydb',
  user: 'user',
  password: 'password',
})

// Using a pre-configured pg.Pool (recommended for pool reuse)
const existingPool = new Pool({
  connectionString: 'postgresql://user:password@localhost:5432/mydb',
  max: 20,
  // ... your custom pool configuration
})

const store4 = new PostgresStore({
  id: 'pg-storage-4',
  pool: existingPool,
  schemaName: 'custom_schema', // optional
})

Remarques supplémentaires

Gestion du schéma

L’implémentation du stockage gère automatiquement la création et les mises à jour du schéma. Elle crée les tables suivantes :

• mastra_workflow_snapshot : stocke l’état des Workflows et les données d’exécution
• mastra_evals : stocke les résultats d’évaluation et les métadonnées
• mastra_threads : stocke les threads de conversation
• mastra_messages : stocke les messages individuels
• mastra_traces : stocke les données de télémétrie et de traçage
• mastra_scorers : stocke les données de notation et d’évaluation
• mastra_resources : stocke les données de mémoire de travail des ressources
• mastra_notifications : stocke les entrées de la boîte de réception des notifications et les métadonnées de livraison

PostgresStore expose le stockage des notifications au moyen de getStore('notifications').

Observabilité

PostgreSQL prend en charge l’observabilité et peut gérer de faibles volumes de traces. La capacité de débit dépend de facteurs de déploiement tels que le matériel, la conception du schéma, l’indexation et les politiques de conservation ; elle doit être validée pour votre environnement précis. Pour les environnements de production à fort volume, envisagez les options suivantes :

• Utiliser la stratégie de traçage insert-only pour réduire les opérations d’écriture dans la base de données
• Configurer le partitionnement des tables pour une conservation efficace des données
• Migrer l’observabilité vers ClickHouse au moyen du stockage composite si vous devez poursuivre la montée en charge

Initialisation

Lorsque vous transmettez le stockage à la classe Mastra, init() est appelée automatiquement avant toute opération de stockage :

import { Mastra } from '@mastra/core'
import { PostgresStore } from '@mastra/pg'

const storage = new PostgresStore({
  id: 'pg-storage',
  connectionString: process.env.DATABASE_URL,
})

const mastra = new Mastra({
  storage, // init() is called automatically
})

Si vous utilisez directement le stockage sans Mastra, vous devez appeler explicitement init() pour créer les tables :

import { PostgresStore } from '@mastra/pg'

const storage = new PostgresStore({
  id: 'pg-storage',
  connectionString: process.env.DATABASE_URL,
})

// Required when using storage directly
await storage.init()

// Access domain-specific stores via getStore()
const memoryStore = await storage.getStore('memory')
const thread = await memoryStore?.getThreadById({ threadId: '...' })

attention

Si init() n’est pas appelée, les tables ne seront pas créées et les opérations de stockage échoueront silencieusement ou généreront des erreurs.

Utilisation d’un pool existant

Si votre application possède déjà un pg.Pool (par exemple, partagé avec un ORM ou utilisé pour la Row Level Security), vous pouvez le transmettre directement à PostgresStore :

import { Pool } from 'pg'
import { PostgresStore } from '@mastra/pg'

// Your existing pool (shared across your application)
const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: 20,
})

const storage = new PostgresStore({
  id: 'shared-storage',
  pool: pool,
})

Comportement du cycle de vie du pool :

• Lorsque vous fournissez un pool : Mastra utilise votre pool, mais ne le ferme pas lors de l’appel de store.close(). Vous gérez le cycle de vie du pool.
• Lorsque Mastra crée un pool : Mastra possède le pool et le ferme lors de l’appel de store.close().

Accès direct à la base de données et au pool

PostgresStore expose le client de base de données et le pool sous-jacents pour les cas d’utilisation avancés :

store.db // DbClient - query interface with helpers (any, one, tx, etc.)
store.pool // pg.Pool - the underlying connection pool

Utilisation de store.db pour les requêtes :

// Execute queries with helper methods
const users = await store.db.any('SELECT * FROM users WHERE active = $1', [true])
const user = await store.db.one('SELECT * FROM users WHERE id = $1', [userId])
const maybeUser = await store.db.oneOrNone('SELECT * FROM users WHERE email = $1', [email])

// Use transactions
const result = await store.db.tx(async t => {
  await t.none('INSERT INTO logs (message) VALUES ($1)', ['Started'])
  const data = await t.any('SELECT * FROM items')
  return data
})

Utilisation directe de store.pool :

// Get a client for manual connection management
const client = await store.pool.connect()
try {
  await client.query('SET LOCAL app.user_id = $1', [userId])
  const result = await client.query('SELECT * FROM protected_table')
  return result.rows
} finally {
  client.release()
}

Lorsque vous utilisez ces champs :

• Vous êtes responsable de la gestion correcte des connexions et des transactions.
• La fermeture du stockage (store.close()) ne détruit le pool que si Mastra l’a créé.
• L’accès direct contourne toute logique ou validation supplémentaire fournie par les méthodes de PostgresStore.

Cette approche est destinée aux scénarios avancés nécessitant un accès de bas niveau.

Utilisation avec Next.js

Lorsque vous utilisez PostgresStore dans des applications Next.js, le Hot Module Replacement (HMR) pendant le développement peut entraîner la création de plusieurs instances de stockage et provoquer cet avertissement :

WARNING: Creating a duplicate database object for the same connection.

Pour éviter cela, stockez l’instance PostgresStore dans l’objet global afin qu’elle persiste entre les rechargements HMR :

src/mastra/storage.ts

import { PostgresStore } from '@mastra/pg'
import { Memory } from '@mastra/memory'

// Extend the global type to include our instances
declare global {
  var pgStore: PostgresStore | undefined
  var memory: Memory | undefined
}

// Get or create the PostgresStore instance
function getPgStore(): PostgresStore {
  if (!global.pgStore) {
    if (!process.env.DATABASE_URL) {
      throw new Error('DATABASE_URL is not defined in environment variables')
    }
    global.pgStore = new PostgresStore({
      id: 'pg-storage',
      connectionString: process.env.DATABASE_URL,
      ssl: process.env.DATABASE_SSL === 'true' ? { rejectUnauthorized: false } : false,
    })
  }
  return global.pgStore
}

// Get or create the Memory instance
function getMemory(): Memory {
  if (!global.memory) {
    global.memory = new Memory({
      storage: getPgStore(),
    })
  }
  return global.memory
}

export const storage = getPgStore()
export const memory = getMemory()

Utilisez ensuite les instances exportées dans votre configuration Mastra :

src/mastra/index.ts

import { Mastra } from '@mastra/core/mastra'
import { storage } from './storage'

export const mastra = new Mastra({
  storage,
  // ...other config
})

Ce pattern garantit qu’une seule instance de PostgresStore est créée, quel que soit le nombre de rechargements du module pendant le développement. Le même pattern peut s’appliquer à d’autres providers de stockage comme LibSQLStore.

astuce

Ce pattern singleton n’est nécessaire que pendant le développement local avec HMR. Dans les builds de production, les modules ne sont chargés qu’une seule fois.

Exemple d’utilisation

Ajout de la mémoire à un Agent

Pour ajouter une mémoire PostgreSQL à un Agent, utilisez la classe Memory et créez une nouvelle clé storage à l’aide de PostgresStore. La connectionString peut désigner un emplacement distant ou une connexion à une base de données locale.

src/mastra/agents/example-pg-agent.ts

import { Memory } from '@mastra/memory'
import { Agent } from '@mastra/core/agent'
import { PostgresStore } from '@mastra/pg'

export const pgAgent = new Agent({
  id: 'pg-agent',
  name: 'PG Agent',
  instructions:
    'You are an AI agent with the ability to automatically recall memories from previous interactions.',
  model: 'openai/gpt-5.6-sol',
  memory: new Memory({
    storage: new PostgresStore({
      id: 'pg-agent-storage',
      connectionString: process.env.DATABASE_URL!,
    }),
    options: {
      generateTitle: true, // Explicitly enable automatic title generation
    },
  }),
})

Utilisation de l’Agent

Utilisez memoryOptions pour délimiter le rappel pour cette requête. Définissez lastMessages: 5 pour limiter le rappel fondé sur la récence, et utilisez semanticRecall afin de récupérer les topK: 3 messages les plus pertinents, avec messageRange: 2 messages voisins pour fournir le contexte autour de chaque correspondance.

src/test-pg-agent.ts

import 'dotenv/config'

import { mastra } from './mastra'

const threadId = '123'
const resourceId = 'user-456'

const agent = mastra.getAgent('pg-agent')

const message = await agent.stream('My name is Mastra', {
  memory: {
    thread: threadId,
    resource: resourceId,
  },
})

await message.textStream.pipeTo(new WritableStream())

const stream = await agent.stream("What's my name?", {
  memory: {
    thread: threadId,
    resource: resourceId,
  },
  memoryOptions: {
    lastMessages: 5,
    semanticRecall: {
      topK: 3,
      messageRange: 2,
    },
  },
})

for await (const chunk of stream.textStream) {
  process.stdout.write(chunk)
}

Gestion des index

Le stockage PostgreSQL permet de gérer les index afin d’optimiser les performances des requêtes.

Index par défaut

Le stockage PostgreSQL crée pendant l’initialisation des index composites adaptés aux patterns de requêtes courants :

• mastra_threads_resourceid_createdat_idx: (resourceId, createdAt DESC)
• mastra_messages_thread_id_createdat_idx: (thread_id, createdAt DESC)
• mastra_ai_spans_traceid_startedat_idx: (traceId, startedAt DESC)
• mastra_ai_spans_parentspanid_startedat_idx: (parentSpanId, startedAt DESC)
• mastra_ai_spans_name_startedat_idx: (name, startedAt DESC)
• mastra_ai_spans_scope_startedat_idx: (scope, startedAt DESC)
• mastra_scores_trace_id_span_id_created_at_idx: (traceId, spanId, createdAt DESC)

Ces index améliorent les performances des requêtes filtrées avec tri, y compris celles qui utilisent des filtres dateRange sur les messages.

Configuration des index

Vous pouvez contrôler la création des index au moyen des options du constructeur :

import { PostgresStore } from '@mastra/pg'

// Skip default indexes (manage indexes separately)
const store = new PostgresStore({
  id: 'pg-storage',
  connectionString: process.env.DATABASE_URL,
  skipDefaultIndexes: true,
})

// Add custom indexes during initialization
const storeWithCustomIndexes = new PostgresStore({
  id: 'pg-storage',
  connectionString: process.env.DATABASE_URL,
  indexes: [
    {
      name: 'idx_threads_metadata_type',
      table: 'mastra_threads',
      columns: ["metadata->>'type'"],
    },
    {
      name: 'idx_messages_status',
      table: 'mastra_messages',
      columns: ["metadata->>'status'"],
    },
  ],
})

Pour les types d’index avancés, vous pouvez spécifier des options supplémentaires :

• unique: true pour les contraintes d’unicité
• where: 'condition' pour les index partiels
• method: 'brin' pour les données de séries temporelles
• storage: { fillfactor: 90 } pour les tables fréquemment mises à jour
• concurrent: true pour une création non bloquante (valeur par défaut)

Options des index

name:

string

Nom unique de l’index

table:

string

Nom de la table (par exemple, 'mastra_threads')

columns:

string[]

Tableau de noms de colonnes avec un ordre de tri facultatif (par exemple, ['id', 'createdAt DESC'])

unique?:

boolean

Crée un index de contrainte d’unicité

concurrent?:

boolean

Crée l’index sans verrouiller la table (valeur par défaut : true)

where?:

string

Condition d’index partiel (propre à PostgreSQL)

method?:

'btree' | 'hash' | 'gin' | 'gist' | 'spgist' | 'brin'

Méthode d’indexation (valeur par défaut : 'btree')

opclass?:

string

Classe d’opérateur pour les index GIN/GIST

storage?:

Record<string, any>

Paramètres de stockage (par exemple, { fillfactor: 90 })

tablespace?:

string

Nom du tablespace dans lequel placer l’index

Index propres au schéma

Lorsque vous utilisez des schémas personnalisés, le nom du schéma est ajouté en préfixe aux noms des index :

const storage = new PostgresStore({
  id: 'pg-storage',
  connectionString: process.env.DATABASE_URL,
  schemaName: 'custom_schema',
  indexes: [
    {
      name: 'idx_threads_status',
      table: 'mastra_threads',
      columns: ['status'],
    },
  ],
})

// Creates index as: custom_schema_idx_threads_status

Gestion des index avec SQL

Pour une gestion avancée des index (affichage, suppression et analyse), utilisez des requêtes SQL directes au moyen de l’accesseur db :

// List indexes for a table
const indexes = await storage.db.any(`
  SELECT indexname, indexdef
  FROM pg_indexes
  WHERE tablename = 'mastra_messages'
`)

// Drop an index
await storage.db.none('DROP INDEX IF EXISTS idx_my_custom_index')

// Analyze index usage
const stats = await storage.db.one(`
  SELECT idx_scan, idx_tup_read
  FROM pg_stat_user_indexes
  WHERE indexrelname = 'mastra_messages_thread_id_createdat_idx'
`)

Types d’index et cas d’utilisation

PostgreSQL propose différents types d’index optimisés pour des scénarios précis :

Type d’index	Utilisation optimale	Stockage	Vitesse
btree (par défaut)	Requêtes par plage, tri, usage général	Modéré	Rapide
hash	Comparaisons d’égalité uniquement	Faible	Très rapide pour =
gin	JSONB, tableaux, recherche en texte intégral	Élevé	Rapide pour les tests de contenu
gist	Données géométriques, recherche en texte intégral	Modéré	Rapide pour le plus proche voisin
spgist	Données non équilibrées, patterns de texte	Faible	Rapide pour des patterns précis
brin	Grandes tables avec un ordre naturel	Très faible	Rapide pour les plages