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、port、database、user、password），否則此項為必填。

host?:

string

資料庫伺服器的主機名稱或 IP 位址。與其他主機參數一併使用，作為 connectionString 的替代方案。

port?:

number

資料庫伺服器的連接埠編號。如未指定，預設為 5432。

database?:

string

要連線的資料庫名稱。

user?:

string

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

password?:

string

資料庫使用者的密碼。

pool?:

pg.Pool

預先設定的 pg.Pool 實例。使用此項可重用現有連線池。提供後，Mastra 不會建立自己的連線池，呼叫 store.close() 時亦不會將其關閉。

schemaName?:

string

你希望儲存使用的綱要名稱。預設為 'public'。

ssl?:

boolean | ConnectionOptions

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

max?:

number

連線池的連線數目上限。預設為 20。

idleTimeoutMillis?:

number

閒置連線在關閉前可維持多久。預設為 30000（30 秒）。

disableInit?:

boolean

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

skipDefaultIndexes?:

boolean

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

indexes?:

CreateIndexOptions[]

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

Constructor 範例

你可以用以下方式建立 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
})

補充說明

綱要管理

此儲存實作會自動處理綱要的建立及更新，並建立以下資料表：

• 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。吞吐量取決於硬件、綱要設計、索引及資料保留政策等部署因素，應針對你的特定環境進行驗證。對於高流量正式環境，可考慮：

• 使用 insert-only tracing 策略，減少資料庫寫入操作
• 設定資料表分割，以有效保留資料
• 如需進一步擴充，透過 composite storage 將可觀測性遷移至 ClickHouse

初始化

將 storage 傳入 Mastra class 時，系統會在任何儲存操作之前自動呼叫 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 執行 query：

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

使用這些欄位時：

• 你須負責妥善處理連線及 transaction。
• 只有在連線池由 Mastra 建立時，關閉 store（store.close()）才會銷毀連線池。
• 直接存取會繞過 PostgresStore method 提供的任何額外邏輯或驗證。

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

配合 Next.js 使用

在 Next.js 應用程式中使用 PostgresStore 時，開發期間的 Hot Module Replacement (HMR) 可能會建立多個 storage 實例，導致出現以下警告：

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

要避免此情況，請將 PostgresStore 實例儲存在 global object，令其在 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 等其他 storage Provider。

提示

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

用法範例

為 Agent 加入 memory

要為 Agent 加入 PostgreSQL memory，請使用 Memory class，並以 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 設定此要求的 recall 範圍。設定 lastMessages: 5 以限制按時間順序 recall 的內容，並使用 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 storage 提供索引管理，以改善 query 效能。

預設索引

PostgreSQL storage 會在初始化期間，為常用 query 模式建立複合索引：

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

這些索引可改善經過篩選及排序的 query 效能，包括訊息 query 的 dateRange 篩選條件。

設定索引

你可以透過 constructor 選項控制索引建立方式：

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 用於 unique constraint
• where: 'condition' 用於 partial index
• method: 'brin' 用於 time-series 資料
• storage: { fillfactor: 90 } 用於頻繁更新的資料表
• concurrent: true 用於 non-blocking 建立（預設）

索引選項

name:

string

索引的唯一名稱

table:

string

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

columns:

string[]

資料欄名稱陣列，可選擇指定排序方式（例如 ['id', 'createdAt DESC']）

unique?:

boolean

建立 unique constraint 索引

concurrent?:

boolean

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

where?:

string

Partial index 條件（PostgreSQL 專用）

method?:

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

索引方法（預設：'btree'）

opclass?:

string

GIN/GIST 索引的 operator class

storage?:

Record<string, any>

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

tablespace?:

string

用於放置索引的 tablespace 名稱

特定綱要的索引

使用自訂綱要時，索引名稱會以綱要名稱作為前綴：

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 query：

// 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（預設）	範圍 query、排序、一般用途	中等	快
hash	僅限相等比較	小	= 比較非常快
gin	JSONB、陣列、全文搜尋	大	contains 操作快
gist	幾何資料、全文搜尋	中等	nearest-neighbor 操作快
spgist	非平衡資料、文字模式	小	特定模式搜尋快
brin	具自然排序的大型資料表	非常小	範圍 query 快