> Discover all available pages from the documentation index: https://mastra.zisheng.pro/zh-HK/llms.txt

# PostgreSQL 儲存

PostgreSQL 儲存實作使用 PostgreSQL 資料庫，提供可用於正式環境的儲存方案。

## 安裝

**npm**:

```bash
npm install @mastra/pg@latest
```

**pnpm**:

```bash
pnpm add @mastra/pg@latest
```

**Yarn**:

```bash
yarn add @mastra/pg@latest
```

**Bun**:

```bash
bun add @mastra/pg@latest
```

## 用法

```typescript
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` 實例：

```ts
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 策略](https://mastra.zisheng.pro/zh-HK/docs/observability/integrations/exporters/mastra-storage)，減少資料庫寫入操作
- 設定資料表分割，以有效保留資料
- 如需進一步擴充，透過 [composite storage 將可觀測性遷移至 ClickHouse](https://mastra.zisheng.pro/zh-HK/reference/storage/composite)

### 初始化

將 storage 傳入 Mastra class 時，系統會在任何儲存操作之前自動呼叫 `init()`：

```typescript
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()` 來建立資料表：

```typescript
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`：

```typescript
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 及連線池，以支援進階使用情境：

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

**使用 `store.db` 執行 query：**

```typescript
// 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`：**

```typescript
// 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)](https://nextjs.org/docs/architecture/fast-refresh) 可能會建立多個 storage 實例，導致出現以下警告：

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

要避免此情況，請將 `PostgresStore` 實例儲存在 global object，令其在 HMR 重新載入後仍然存在：

```typescript
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 設定中使用匯出的實例：

```typescript
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` 可以是遠端位置或本機資料庫連線。

```typescript
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` 則訊息作為上下文。

```typescript
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 選項控制索引建立方式：

```typescript
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 名稱

### 特定綱要的索引

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

```typescript
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：

```typescript
// 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 快           |