> Discover all available pages from the documentation index: https://mastra.zisheng.pro/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`): 希望存储使用的 schema 名称。默认为 'public'。

**ssl** (`boolean | ConnectionOptions`): 连接的 SSL 配置；设为 true 可使用默认 SSL，或提供 ConnectionOptions 对象以使用自定义 SSL 设置。

**max** (`number`): 连接池中的最大连接数。默认为 20。

**idleTimeoutMillis** (`number`): 连接在关闭前可保持空闲的时长。默认为 30000（30 秒）。

**disableInit** (`boolean`): 设为 true 时，将禁用自动创建表和迁移。适用于单独运行迁移的 CI/CD 流水线。

**skipDefaultIndexes** (`boolean`): 设为 true 时，初始化期间不会创建默认索引。

**indexes** (`CreateIndexOptions[]`): 初始化期间要创建的自定义索引。

## 构造函数示例

可以通过以下方式实例化 `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
})
```

## 其他说明

### Schema 管理

存储实现会自动处理 schema 的创建和更新。它会创建以下表：

- `mastra_workflow_snapshot`：存储 Workflow 状态和执行数据
- `mastra_evals`：存储评估结果和元数据
- `mastra_threads`：存储会话线程
- `mastra_messages`：存储单条消息
- `mastra_traces`：存储遥测和追踪数据
- `mastra_scorers`：存储评分和评估数据
- `mastra_resources`：存储资源工作记忆数据
- `mastra_notifications`：存储通知收件箱记录和投递元数据

`PostgresStore` 通过 `getStore('notifications')` 提供通知存储。

### 可观测性

PostgreSQL 支持可观测性，并且可以处理较低的 Trace 量。吞吐能力取决于硬件、schema 设计、索引和保留策略等部署因素，应针对具体环境进行验证。对于高容量生产环境，请考虑：

- 使用 `insert-only` [追踪策略](https://mastra.zisheng.pro/docs/observability/integrations/exporters/mastra-storage) 以减少数据库写入操作
- 设置表分区以高效保留数据
- 如果需要进一步扩展，将可观测性迁移到[通过复合存储使用的 ClickHouse](https://mastra.zisheng.pro/reference/storage/composite)

### 初始化

将存储传入 Mastra 类时，会在任何存储操作之前自动调用 `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 而直接使用存储，必须显式调用 `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` 会公开底层数据库客户端和连接池，供高级用例使用：

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

**使用 `store.db` 执行查询：**

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

使用这些字段时：

- 你负责正确处理连接和事务。
- 只有当 Mastra 创建连接池时，关闭存储（`store.close()`）才会销毁连接池。
- 直接访问会绕过 PostgresStore 方法提供的任何额外逻辑或验证。

这种方式适用于需要低层访问的高级场景。

### 与 Next.js 一起使用

在 Next.js 应用程序中使用 `PostgresStore` 时，开发期间的 [Hot Module Replacement (HMR)](https://nextjs.org/docs/architecture/fast-refresh) 可能会创建多个存储实例，从而导致以下警告：

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

为防止这种情况，请将 `PostgresStore` 实例存储在全局对象上，使其在 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` 等其他存储 Provider。

> **提示:** 此单例模式仅在使用 HMR 的本地开发期间需要。生产构建中，模块只会加载一次。

## 使用示例

### 为 Agent 添加记忆

要为 Agent 添加 PostgreSQL 记忆，请使用 `Memory` 类，并使用 `PostgresStore` 创建新的 `storage` 键。`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` 限定此请求的回忆范围。设置 `lastMessages: 5` 可限制基于近期消息的回忆；使用 `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 存储提供索引管理功能，以优化查询性能。

### 默认索引

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` 筛选器。

### 配置索引

可以通过构造函数选项控制索引创建：

```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` 用于唯一约束
- `where: 'condition'` 用于部分索引
- `method: 'brin'` 用于时序数据
- `storage: { fillfactor: 90 }` 用于更新频繁的表
- `concurrent: true` 用于非阻塞创建（默认值）

### 索引选项

**name** (`string`): 索引的唯一名称

**table** (`string`): 表名（例如 'mastra\_threads'）

**columns** (`string[]`): 列名称数组，可包含可选的排序顺序（例如 \['id', 'createdAt DESC']）

**unique** (`boolean`): 创建唯一约束索引

**concurrent** (`boolean`): 在不锁定表的情况下创建索引（默认值：true）

**where** (`string`): 部分索引条件（PostgreSQL 特有）

**method** (`'btree' | 'hash' | 'gin' | 'gist' | 'spgist' | 'brin'`): 索引方法（默认值：'btree'）

**opclass** (`string`): 用于 GIN/GIST 索引的运算符类

**storage** (`Record<string, any>`): 存储参数（例如 { fillfactor: 90 }）

**tablespace** (`string`): 用于放置索引的表空间名称

### Schema 专属索引

使用自定义 schema 时，索引名称会以 schema 名称作为前缀：

```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` 访问器使用直接 SQL 查询：

```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**（默认） | 范围查询、排序、通用用途  | 中等   | 快         |
| **hash**      | 仅限等值比较        | 小    | 对 `=` 非常快 |
| **gin**       | JSONB、数组、全文搜索 | 大    | 对包含查询很快   |
| **gist**      | 几何数据、全文搜索     | 中等   | 对最近邻查询很快  |
| **spgist**    | 非平衡数据、文本模式    | 小    | 对特定模式很快   |
| **brin**      | 具有自然排序的大型表    | 很小   | 对范围查询很快   |