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

希望存储使用的 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：

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 追踪策略 以减少数据库写入操作
• 设置表分区以高效保留数据
• 如果需要进一步扩展，将可观测性迁移到通过复合存储使用的 ClickHouse

初始化

将存储传入 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 而直接使用存储，必须显式调用 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 会公开底层数据库客户端和连接池，供高级用例使用：

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.close()）才会销毁连接池。
• 直接访问会绕过 PostgresStore 方法提供的任何额外逻辑或验证。

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

与 Next.js 一起使用

在 Next.js 应用程序中使用 PostgresStore 时，开发期间的 Hot Module Replacement (HMR) 可能会创建多个存储实例，从而导致以下警告：

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

为防止这种情况，请将 PostgresStore 实例存储在全局对象上，使其在 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 的本地开发期间需要。生产构建中，模块只会加载一次。

使用示例

为 Agent 添加记忆

要为 Agent 添加 PostgreSQL 记忆，请使用 Memory 类，并使用 PostgresStore 创建新的 storage 键。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 筛选器。

配置索引

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

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 名称作为前缀：

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 查询：

// 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	具有自然排序的大型表	很小	对范围查询很快