PostgreSQL 儲存空間

PostgreSQL 儲存空間實作採用 PostgreSQL 資料庫，提供可用於正式環境的儲存解決方案。

安裝

npm

npm install @mastra/pg@latest

pnpm

pnpm add @mastra/pg@latest

Yarn

yarn add @mastra/pg@latest

Bun

bun add @mastra/pg@latest

使用方式

import { PostgresStore } from '@mastra/pg'

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

參數

id:

string

此儲存空間執行個體的唯一識別碼。

connectionString?:

string

PostgreSQL 連線字串（例如 postgresql://user:pass@host:5432/dbname）。除非使用 pool 或個別 host 參數（host、port、database、user、password），否則為必填。

host?:

string

資料庫伺服器 hostname 或 IP 位址。與其他 host 參數搭配使用，可替代 connectionString。

port?:

number

資料庫伺服器 port 號碼。若未指定，預設為 5432。

database?:

string

要連線的資料庫名稱。

user?:

string

用於驗證的資料庫使用者。

password?:

string

資料庫使用者的密碼。

pool?:

pg.Pool

預先設定的 pg.Pool 執行個體。使用此選項可重複使用現有連線池。提供此選項時，Mastra 不會建立自己的連線池，呼叫 store.close() 時也不會將它關閉。

schemaName?:

string

希望儲存空間使用的 schema 名稱。預設為 'public'。

ssl?:

boolean | ConnectionOptions

連線的 SSL 設定；設為 true 可使用預設 SSL，或提供 ConnectionOptions 物件以使用自訂 SSL 設定。

max?:

number

連線池的最大連線數。預設為 20。

idleTimeoutMillis?:

number

連線在關閉前可閒置的時間。預設為 30000（30 秒）。

disableInit?:

boolean

設為 true 時，會停用自動建立資料表與 migration。適合另行執行 migration 的 CI/CD pipeline。

skipDefaultIndexes?:

boolean

設為 true 時，初始化期間不會建立預設索引。

indexes?:

CreateIndexOptions[]

初始化期間要建立的自訂索引。

建構函式範例

你可以使用下列方式建立 PostgresStore 執行個體：

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

其他注意事項

Schema 管理

儲存空間實作會自動處理 schema 的建立與更新，並建立下列資料表：

• mastra_workflow_snapshot：儲存 Workflow 狀態與執行資料
• mastra_evals：儲存評估結果與 metadata
• mastra_threads：儲存對話 thread
• mastra_messages：儲存個別訊息
• mastra_traces：儲存遙測與 tracing 資料
• mastra_scorers：儲存評分與評估資料
• mastra_resources：儲存資源的 working memory 資料
• mastra_notifications：儲存通知收件匣記錄與傳遞 metadata

PostgresStore 透過 getStore('notifications') 提供通知儲存空間。

可觀測性

PostgreSQL 支援可觀測性，並可處理少量 Trace。吞吐量取決於硬體、schema 設計、索引與保留政策等部署因素，應針對實際環境進行驗證。對於大量資料的正式環境，請考慮：

• 使用 insert-only tracing strategy，減少資料庫寫入操作
• 設定資料表 partition，以有效率地保留資料
• 如需進一步擴充，請將可觀測性遷移至透過複合儲存空間使用的 ClickHouse

初始化

將 storage 傳入 Mastra 類別時，系統會在進行任何儲存操作前自動呼叫 init()：

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

若不透過 Mastra 而直接使用 storage，必須明確呼叫 init() 來建立資料表：

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: '...' })

警告

若未呼叫 init()，系統不會建立資料表，儲存操作將會無聲失敗或擲回錯誤。

使用現有連線池

若應用程式已有 pg.Pool（例如與 ORM 共用，或用於 Row Level Security），可直接將它傳入 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,
})

連線池生命週期行為：

• 由你提供連線池時：Mastra 會使用該連線池，但呼叫 store.close() 時不會關閉它。你必須自行管理連線池生命週期。
• 由 Mastra 建立連線池時：Mastra 擁有該連線池，並會在呼叫 store.close() 時將它關閉。

直接存取資料庫與連線池

PostgresStore 會提供底層資料庫 client 與連線池，以供進階使用情境使用：

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

使用 store.db 進行查詢：

// 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
})

直接使用 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()
}

使用這些欄位時：

• 你必須負責正確處理連線與交易。
• 只有在連線池由 Mastra 建立時，關閉 store（store.close()）才會銷毀連線池。
• 直接存取會略過 PostgresStore 方法提供的任何額外邏輯或驗證。

此方式適用於需要低階存取的進階情境。

搭配 Next.js 使用

在 Next.js 應用程式中使用 PostgresStore 時，開發期間的 Hot Module Replacement（HMR）可能建立多個 storage 執行個體，因而出現下列警告：

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

若要避免此情況，請將 PostgresStore 執行個體儲存在 global 物件上，使其在 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()

接著在 Mastra 設定中使用匯出的執行個體：

src/mastra/index.ts

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

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

此模式可確保不論開發期間重新載入模組多少次，都只會建立一個 PostgresStore 執行個體。相同模式也可套用至 LibSQLStore 等其他儲存 Provider。

提示

只有搭配 HMR 進行本機開發時才需要此 singleton 模式。在正式環境 build 中，模組只會載入一次。

使用範例

為 Agent 新增記憶體

若要為 Agent 新增 PostgreSQL 記憶體，請使用 Memory 類別，並以 PostgresStore 建立新的 storage key。connectionString 可以是遠端位置或本機資料庫連線。

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
    },
  }),
})

使用 Agent

使用 memoryOptions 設定此請求的回憶範圍。設定 lastMessages: 5 以限制依時間順序回憶的訊息數量，並使用 semanticRecall 擷取 topK: 3 筆最相關訊息；其中包含 messageRange: 2 筆相鄰訊息，作為各配對結果的上下文。

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

索引管理

PostgreSQL 儲存空間提供索引管理功能，可最佳化查詢效能。

預設索引

PostgreSQL 儲存空間會在初始化期間，針對常見查詢模式建立複合索引：

• 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)

這些索引可提升包含排序之篩選查詢的效能，包括訊息查詢的 dateRange filter。

設定索引

你可以透過建構函式選項控制索引建立：

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'"],
    },
  ],
})

對於進階索引型別，可以指定其他選項：

• unique: true：唯一 constraint
• where: 'condition'：部分索引
• method: 'brin'：時間序列資料
• storage: { fillfactor: 90 }：更新密集的資料表
• concurrent: true：非阻塞建立（預設）

索引選項

name:

string

索引的唯一名稱

table:

string

資料表名稱（例如 'mastra_threads'）

columns:

string[]

欄位名稱陣列，可包含選用的排序順序（例如 ['id', 'createdAt DESC']）

unique?:

boolean

建立唯一 constraint 索引

concurrent?:

boolean

建立索引時不鎖定資料表（預設：true）

where?:

string

部分索引條件（PostgreSQL 專用）

method?:

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

索引 method（預設：'btree'）

opclass?:

string

GIN/GIST 索引的 operator class

storage?:

Record<string, any>

儲存參數（例如 { fillfactor: 90 }）

tablespace?:

string

放置索引的 tablespace 名稱

Schema 專用索引

使用自訂 schema 時，索引名稱會加上 schema 名稱 prefix：

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

透過 SQL 管理索引

對於進階索引管理（列出、刪除、分析），請透過 db accessor 使用直接 SQL 查詢：

// 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'
`)

索引型別與使用情境

PostgreSQL 提供針對特定情境最佳化的不同索引型別：

索引型別	最適合	儲存空間	速度
btree（預設）	範圍查詢、排序、一般用途	中等	快速
hash	僅相等比較	小	= 極快
gin	JSONB、陣列、全文搜尋	大	contains 快速
gist	幾何資料、全文搜尋	中等	最近鄰快速
spgist	非平衡資料、文字模式	小	特定模式快速
brin	具有自然順序的大型資料表	極小	範圍查詢快速