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 구성입니다. 기본 SSL을 사용하려면 true로 설정하고, 사용자 정의 SSL 설정에는 ConnectionOptions 객체를 제공합니다.

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

추가 참고사항

스키마 관리

저장소 구현은 스키마 생성 및 업데이트를 자동으로 처리합니다. 다음 테이블이 생성됩니다.

• mastra_workflow_snapshot: Workflow 상태 및 실행 데이터를 저장합니다.
• mastra_evals: 평가 결과 및 메타데이터를 저장합니다.
• mastra_threads: 대화 스레드를 저장합니다.
• mastra_messages: 개별 메시지를 저장합니다.
• mastra_traces: 원격 측정 및 추적 데이터를 저장합니다.
• mastra_scorers: 점수 및 평가 데이터를 저장합니다.
• mastra_resources: 자원 작업 Memory 데이터를 저장합니다.
• mastra_notifications: 알림 받은 편지함 기록 및 전달 메타데이터를 저장합니다.

PostgresStore다음을 통해 알림 저장소를 노출합니다.getStore('notifications').

Observability

PostgreSQL은 Observability을 지원하고 낮은 추적 볼륨을 처리할 수 있습니다. 처리량 용량은 하드웨어, 스키마 디자인, 인덱싱 및 보존 정책과 같은 배포 요소에 따라 달라지며 특정 환경에 대해 검증되어야 합니다. 대량 생산 환경의 경우 다음을 고려하십시오.

• 데이터베이스 쓰기 작업을 줄이려면 insert-only 추적 전략을 사용합니다.
• 효율적인 데이터 보존을 위해 테이블 파티셔닝을 설정합니다.
• 더 확장해야 한다면 Observability를 복합 저장소를 통해 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과 공유하거나 행 수준 보안에 사용) 이를 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 directly:

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

다음 필드를 사용하는 경우:

• 적절한 연결 및 트랜잭션 처리는 사용자가 책임져야 합니다.
• 저장소를 닫으면(store.close()) Mastra가 생성한 경우에만 풀이 제거됩니다.
• 직접 액세스하면 PostgresStore 메서드가 제공하는 추가 로직이나 유효성 검사를 우회합니다. 이 접근 방식은 낮은 수준의 액세스가 필요한 고급 시나리오를 위한 것입니다.

Next.js와 함께 사용

Next.js 애플리케이션에서 PostgresStore를 사용할 때 개발 중 Hot Module Replacement(HMR)로 인해 여러 저장소 인스턴스가 생성되어 다음 경고가 표시될 수 있습니다.

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

이를 방지하려면 HMR 새로고침 후에도 유지되도록 PostgresStore 인스턴스를 전역 객체에 저장하세요.

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에 Memory 추가

Agent에 PostgreSQL Memory를 추가하려면 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, DESC에서 생성됨)
• mastra_messages_thread_id_createdat_idx: (thread_id, DESC에서 생성됨)
• mastra_ai_spans_traceid_startedat_idx: (traceId, DESC에서 시작됨)
• mastra_ai_spans_parentspanid_startedat_idx: (parentSpanId, DESC에서 시작됨)
• mastra_ai_spans_name_startedat_idx: (이름, DESC에서 시작함)
• mastra_ai_spans_scope_startedat_idx: (범위, DESC에서 시작됨)
• mastra_scores_trace_id_span_id_created_at_idx: (traceId, spanId, 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

인덱스를 배치할 테이블스페이스 이름입니다.

스키마별 인덱스

사용자 정의 스키마를 사용할 때 인덱스 이름 앞에는 스키마 이름이 붙습니다.

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을 통한 인덱스 관리

고급 인덱스 관리(나열, 삭제, 분석)를 위해서는 다음을 통해 직접 SQL 쿼리를 사용하세요.db accessor:

// 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	자연스러운 순서가 있는 대규모 테이블	매우 작음	범위 검색에 빠름