본문으로 건너뛰기

Generation 어댑터 빌드

이미지, 오디오, 동영상 실행에는 자체 영속성이 필요합니다. 다시 로드했을 때 슬롯의 마지막 생성을 찾을 수 있는 실행 레코드와 미디어 자체를 복원할 수 있는 바이트 저장소가 필요합니다. 이 페이지에서는 SQLite(Node에 내장된 node:sqlite)를 사용해 이 세 가지 스토어를 빌드합니다.

먼저 직접 어댑터 빌드를 읽고 어댑터의 형태와 앱에 필요한 스토어를 확인합니다. 모든 메서드 시그니처와 불변 조건은 스토어 레퍼런스에 있습니다.

미디어 생성은 채팅과 다르게 영속화됩니다. 채팅의 runs 스토어를 전혀 사용하지 않습니다.

  • 필수: runId(생성이 발급하는 실행/요청 ID)를 키로 사용하는 generationRuns 스토어인 GenerationRunStore입니다. runs에 대응하는 스토어입니다.
  • 선택 사항(생성된 바이트를 유지하려는 경우): artifacts 스토어(메타데이터)와 blobs 스토어(바이트)입니다. 이 두 스토어는 반드시 함께 제공해야 합니다.

threadId는 실행이 속한 슬롯이며 각 실행 레코드에 기록됩니다.

다음 세 테이블은 채팅 어댑터가 사용하는 네 테이블과 독립적입니다.

CREATE TABLE IF NOT EXISTS generation_runs (
  run_id text PRIMARY KEY NOT NULL,
  thread_id text NOT NULL,
  activity text NOT NULL,
  provider text NOT NULL,
  model text NOT NULL,
  status text NOT NULL,
  started_at integer NOT NULL,
  finished_at integer,
  error_json text,
  result_json text,
  artifacts_json text,
  usage_json text
);
CREATE TABLE IF NOT EXISTS artifacts (
  artifact_id text PRIMARY KEY NOT NULL,
  run_id text NOT NULL,
  thread_id text NOT NULL,
  blob_key text,
  name text NOT NULL,
  mime_type text NOT NULL,
  size integer NOT NULL,
  source_url text,
  created_at integer NOT NULL
);
CREATE INDEX IF NOT EXISTS artifacts_run_order
  ON artifacts (run_id, created_at, artifact_id);
CREATE INDEX IF NOT EXISTS artifacts_thread_order
  ON artifacts (thread_id, created_at, artifact_id);
CREATE TABLE IF NOT EXISTS blobs (
  key text PRIMARY KEY NOT NULL,
  bytes blob NOT NULL,
  size integer NOT NULL,
  etag text NOT NULL,
  content_type text,
  custom_metadata_json text,
  created_at integer NOT NULL,
  updated_at integer NOT NULL
);

생성 실행: 멱등적 생성, 패치, 스레드별 최신 항목​

GenerationRunStore는 생성에서 RunStore에 해당합니다. 다음 세 가지 계약을 준수해야 합니다.

  • createOrResume은 멱등적입니다. runId에 대한 두 번째 호출은 저장된 레코드를 변경 없이 반환하므로 실행을 재개해도 startedAt, activity 또는 상태가 재설정되지 않습니다. INSERT ... ON CONFLICT DO NOTHING으로 이를 구현할 수 있습니다.
  • 알 수 없는 runId에 대한 update는 아무 작업도 하지 않습니다.
  • findLatestForThread는 스레드에 연결된 실행 중 startedAt이 가장 큰 실행을 반환합니다. 서버 주도 클라이언트가 마운트될 때 reconstructGeneration이 이를 호출해 스레드의 마지막 생성을 복원합니다.
import { DatabaseSync } from 'node:sqlite'
import { defineGenerationRunStore } from '@tanstack/ai-persistence'
import type {
  GenerationRunRecord,
  GenerationRunStatus,
} from '@tanstack/ai-persistence'

function toGenerationRunStatus(value: unknown): GenerationRunStatus {
  switch (value) {
    case 'running':
    case 'completed':
    case 'failed':
    case 'interrupted':
      return value
    default:
      throw new TypeError(`Unexpected generation run status: ${String(value)}`)
  }
}

// `node:sqlite` types columns as a SQL-value union, so coerce/narrow each field
// (String / Number / typeof) and JSON-parse the text columns, with no cast.
function mapGenerationRun(row: Record<string, unknown>): GenerationRunRecord {
  return {
    runId: String(row.run_id),
    threadId: String(row.thread_id),
    activity: String(row.activity),
    provider: String(row.provider),
    model: String(row.model),
    status: toGenerationRunStatus(row.status),
    startedAt: Number(row.started_at),
    ...(row.finished_at != null ? { finishedAt: Number(row.finished_at) } : {}),
    ...(typeof row.error_json === 'string'
      ? { error: JSON.parse(row.error_json) }
      : {}),
    ...(typeof row.result_json === 'string'
      ? { result: JSON.parse(row.result_json) }
      : {}),
    ...(typeof row.artifacts_json === 'string'
      ? { artifacts: JSON.parse(row.artifacts_json) }
      : {}),
    ...(typeof row.usage_json === 'string'
      ? { usage: JSON.parse(row.usage_json) }
      : {}),
  }
}

function createGenerationRunStore(db: DatabaseSync) {
  const select = db.prepare('SELECT * FROM generation_runs WHERE run_id = ?')
  const insert = db.prepare(
    `INSERT INTO generation_runs
       (run_id, thread_id, activity, provider, model, status, started_at)
     VALUES (?, ?, ?, ?, ?, ?, ?)
     ON CONFLICT(run_id) DO NOTHING`,
  )
  const latest = db.prepare(
    `SELECT * FROM generation_runs WHERE thread_id = ?
     ORDER BY started_at DESC LIMIT 1`,
  )
  return defineGenerationRunStore({
    async createOrResume(input) {
      const existing = select.get(input.runId)
      if (existing) return mapGenerationRun(existing)
      const status: GenerationRunStatus = input.status ?? 'running'
      insert.run(
        input.runId,
        input.threadId,
        input.activity,
        input.provider,
        input.model,
        status,
        input.startedAt,
      )
      return {
        runId: input.runId,
        threadId: input.threadId,
        activity: input.activity,
        provider: input.provider,
        model: input.model,
        status,
        startedAt: input.startedAt,
      }
    },
    async update(runId, patch) {
      const sets: Array<string> = []
      const params: Array<string | number> = []
      if (patch.status !== undefined) {
        sets.push('status = ?')
        params.push(patch.status)
      }
      if (patch.finishedAt !== undefined) {
        sets.push('finished_at = ?')
        params.push(patch.finishedAt)
      }
      if (patch.error !== undefined) {
        sets.push('error_json = ?')
        params.push(JSON.stringify(patch.error))
      }
      if (patch.result !== undefined) {
        sets.push('result_json = ?')
        params.push(JSON.stringify(patch.result))
      }
      if (patch.artifacts !== undefined) {
        sets.push('artifacts_json = ?')
        params.push(JSON.stringify(patch.artifacts))
      }
      if (patch.usage !== undefined) {
        sets.push('usage_json = ?')
        params.push(JSON.stringify(patch.usage))
      }
      // Empty patch, or an unknown run id, touches nothing (UPDATE no-ops).
      if (sets.length === 0) return
      params.push(runId)
      db.prepare(
        `UPDATE generation_runs SET ${sets.join(', ')} WHERE run_id = ?`,
      ).run(...params)
    },
    async get(runId) {
      const row = select.get(runId)
      return row ? mapGenerationRun(row) : null
    },
    // The most recent run linked to a thread. `reconstructGeneration` calls this
    // so a server-driven client (`persistence: true`) hydrates the last
    // generation for its thread by the stable thread id, without a run id.
    async findLatestForThread(threadId) {
      const row = latest.get(threadId)
      return row ? mapGenerationRun(row) : null
    },
  })
}

아티팩트: 미디어 메타데이터​

ArtifactStore는 생성된 파일마다 메타데이터 행 하나를 보유합니다. 여기에는 runId, mimeType, size, createdAt이 포함됩니다. 바이트는 아래의 blob 스토어에 저장됩니다.

  • save는 upsert입니다.
  • list(runId)는 실행의 모든 아티팩트를 반환하며, 없으면 []을 반환합니다.
  • listForThread(threadId)는 스레드의 모든 아티팩트를 정확한 (createdAt, artifactId) 오름차순으로 반환합니다. 페이지나 최신 실행만이 아니라 스레드의 전체 기록을 반환해야 합니다. 스냅샷 캡처는 이 기준을 사용합니다.
  • delete / deleteForRun은 필수입니다. 미디어를 영속적으로 저장하는 목적은 보존과 삭제를 지원하는 것이며, 이 메서드들은 BlobStore.delete에 대응합니다.

blobKey는 있는 그대로 영속화합니다. 이 값은 바이트가 실제로 저장된 위치를 기록하며, storageKey 매퍼는 바이트를 어디에나 저장할 수 있으므로 리더가 경로를 다시 계산할 수 없습니다. resolveArtifactBlobKey(record)는 해당 열이 존재하기 전에 작성된 행에만 기본 규칙을 사용합니다. 이 값을 삭제하면 사용자 지정 키로 저장된 모든 아티팩트를 읽을 수 없게 됩니다.

import { DatabaseSync } from 'node:sqlite'
import { defineArtifactStore } from '@tanstack/ai-persistence'
import type { ArtifactRecord } from '@tanstack/ai-persistence'

function mapArtifact(row: Record<string, unknown>): ArtifactRecord {
  return {
    artifactId: String(row.artifact_id),
    runId: String(row.run_id),
    threadId: String(row.thread_id),
    ...(typeof row.blob_key === 'string' ? { blobKey: row.blob_key } : {}),
    name: String(row.name),
    mimeType: String(row.mime_type),
    size: Number(row.size),
    ...(typeof row.source_url === 'string'
      ? { sourceUrl: row.source_url }
      : {}),
    createdAt: Number(row.created_at),
  }
}

function createArtifactStore(db: DatabaseSync) {
  const upsert = db.prepare(
    `INSERT INTO artifacts
       (artifact_id, run_id, thread_id, blob_key, name, mime_type, size, source_url, created_at)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
     ON CONFLICT(artifact_id) DO UPDATE SET
       run_id = excluded.run_id, thread_id = excluded.thread_id,
       blob_key = excluded.blob_key, name = excluded.name,
       mime_type = excluded.mime_type, size = excluded.size,
       source_url = excluded.source_url, created_at = excluded.created_at`,
  )
  const selectOne = db.prepare('SELECT * FROM artifacts WHERE artifact_id = ?')
  const byRun = db.prepare(
    'SELECT * FROM artifacts WHERE run_id = ? ORDER BY created_at ASC, artifact_id ASC',
  )
  const byThread = db.prepare(
    'SELECT * FROM artifacts WHERE thread_id = ? ORDER BY created_at ASC, artifact_id ASC',
  )
  return defineArtifactStore({
    async save(record) {
      upsert.run(
        record.artifactId,
        record.runId,
        record.threadId,
        record.blobKey ?? null,
        record.name,
        record.mimeType,
        record.size,
        record.sourceUrl ?? null,
        record.createdAt,
      )
    },
    async get(artifactId) {
      const row = selectOne.get(artifactId)
      return row ? mapArtifact(row) : null
    },
    async list(runId) {
      return byRun.all(runId).map(mapArtifact)
    },
    async listForThread(threadId) {
      return byThread.all(threadId).map(mapArtifact)
    },
    async delete(artifactId) {
      db.prepare('DELETE FROM artifacts WHERE artifact_id = ?').run(artifactId)
    },
    async deleteForRun(runId) {
      db.prepare('DELETE FROM artifacts WHERE run_id = ?').run(runId)
    },
  })
}

blob: 바이트​

BlobStore는 작은 객체 저장소입니다. withGenerationPersistence는 각 생성 파일을 artifacts/<runId>/<artifactId> 키로 저장하므로, 접두사로 필터링한 list({ prefix: 'artifacts/<runId>/' })로 실행의 미디어를 열거할 수 있습니다.

  • put은 스트림, 버퍼, 문자열 또는 Blob인 모든 BlobBody를 허용합니다. 아래 헬퍼는 이를 바이트로 정규화합니다.
  • get은 해당 슬라이스만 반환해 options.range를 준수하며, 이는 미디어 플레이어의 탐색에 필요합니다. BlobStore 계약을 참조합니다.
  • list는 prefix를 리터럴로 일치시키고 키셋 커서로 페이지를 나눕니다.
import { DatabaseSync } from 'node:sqlite'
import { defineBlobStore, resolveBlobRange } from '@tanstack/ai-persistence'
import type {
  BlobBody,
  BlobObject,
  BlobRecord,
} from '@tanstack/ai-persistence'

async function toBytes(body: BlobBody): Promise<Uint8Array> {
  if (typeof body === 'string') return new TextEncoder().encode(body)
  if (body instanceof ArrayBuffer) return new Uint8Array(body.slice(0))
  if (ArrayBuffer.isView(body)) {
    return new Uint8Array(body.buffer, body.byteOffset, body.byteLength).slice()
  }
  if (body instanceof Blob) {
    return new Uint8Array(await body.arrayBuffer())
  }
  // ReadableStream<Uint8Array>: drain it into one buffer.
  const reader = body.getReader()
  const chunks: Array<Uint8Array> = []
  let total = 0
  for (;;) {
    const { done, value } = await reader.read()
    if (done) break
    chunks.push(value)
    total += value.byteLength
  }
  const bytes = new Uint8Array(total)
  let offset = 0
  for (const chunk of chunks) {
    bytes.set(chunk, offset)
    offset += chunk.byteLength
  }
  return bytes
}

function mapBlobRecord(row: Record<string, unknown>): BlobRecord {
  return {
    key: String(row.key),
    ...(row.size != null ? { size: Number(row.size) } : {}),
    ...(typeof row.etag === 'string' ? { etag: row.etag } : {}),
    ...(typeof row.content_type === 'string'
      ? { contentType: row.content_type }
      : {}),
    ...(typeof row.custom_metadata_json === 'string'
      ? { customMetadata: JSON.parse(row.custom_metadata_json) }
      : {}),
    ...(row.created_at != null ? { createdAt: Number(row.created_at) } : {}),
    ...(row.updated_at != null ? { updatedAt: Number(row.updated_at) } : {}),
  }
}

function blobObject(
  record: BlobRecord,
  bytes: Uint8Array,
  range?: { offset: number; length: number },
): BlobObject {
  return {
    ...record,
    // `size` keeps describing the whole object; `range` describes these bytes.
    ...(range ? { range } : {}),
    body: new ReadableStream<Uint8Array>({
      start(controller) {
        controller.enqueue(bytes.slice())
        controller.close()
      },
    }),
    arrayBuffer() {
      const copy = new ArrayBuffer(bytes.byteLength)
      new Uint8Array(copy).set(bytes)
      return Promise.resolve(copy)
    },
    text: () => Promise.resolve(new TextDecoder().decode(bytes)),
  }
}

function createBlobStore(db: DatabaseSync) {
  const upsert = db.prepare(
    `INSERT INTO blobs
       (key, bytes, size, etag, content_type, custom_metadata_json, created_at, updated_at)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?)
     ON CONFLICT(key) DO UPDATE SET
       bytes = excluded.bytes, size = excluded.size, etag = excluded.etag,
       content_type = excluded.content_type,
       custom_metadata_json = excluded.custom_metadata_json,
       updated_at = excluded.updated_at`,
  )
  const selectCreated = db.prepare('SELECT created_at FROM blobs WHERE key = ?')
  const selectOne = db.prepare('SELECT * FROM blobs WHERE key = ?')
  // Metadata without the bytes, and the bounded slice: a ranged read must not
  // load the whole object to hand back a piece of it.
  const selectMeta = db.prepare(
    `SELECT key, size, etag, content_type, custom_metadata_json,
            created_at, updated_at
       FROM blobs WHERE key = ?`,
  )
  const selectSlice = db.prepare(
    'SELECT substr(bytes, ?, ?) AS bytes FROM blobs WHERE key = ?',
  )
  return defineBlobStore({
    async put(key, body, options) {
      const bytes = await toBytes(body)
      const now = Date.now()
      const prior = selectCreated.get(key)
      const createdAt =
        prior && prior.created_at != null ? Number(prior.created_at) : now
      const etag = String(now)
      upsert.run(
        key,
        bytes,
        bytes.byteLength,
        etag,
        options?.contentType ?? null,
        options?.customMetadata ? JSON.stringify(options.customMetadata) : null,
        createdAt,
        now,
      )
      return {
        key,
        size: bytes.byteLength,
        etag,
        createdAt,
        updatedAt: now,
        ...(options?.contentType !== undefined
          ? { contentType: options.contentType }
          : {}),
        ...(options?.customMetadata !== undefined
          ? { customMetadata: options.customMetadata }
          : {}),
      }
    },
    async get(key, options) {
      if (!options?.range) {
        const row = selectOne.get(key)
        if (!row) return null
        const bytes =
          row.bytes instanceof Uint8Array ? row.bytes : new Uint8Array()
        return blobObject(mapBlobRecord(row), bytes)
      }
      // Metadata first, WITHOUT the bytes, so the clamp costs no I/O...
      const meta = selectMeta.get(key)
      if (!meta) return null
      const served = resolveBlobRange(Number(meta.size), options.range)
      // ...then let SQLite cut the slice (`substr` is 1-based and byte-wise
      // over a BLOB). Reading the row whole and slicing in JS would load the
      // entire object on every video seek, the cost ranges exist to avoid.
      const slice = selectSlice.get(served.offset + 1, served.length, key)
      if (!slice) return null
      const bytes =
        slice.bytes instanceof Uint8Array ? slice.bytes : new Uint8Array()
      return blobObject(mapBlobRecord(meta), bytes, served)
    },
    async head(key) {
      // Metadata only: never pull the bytes to answer a question about them.
      const row = selectMeta.get(key)
      return row ? mapBlobRecord(row) : null
    },
    async delete(key) {
      db.prepare('DELETE FROM blobs WHERE key = ?').run(key)
    },
    async list(options) {
      if (options?.limit === 0) return { objects: [], truncated: false }
      // Match the prefix with `substr(...) = ?` rather than LIKE: SQLite's LIKE
      // is case-INsensitive for ASCII and treats `%`/`_` as wildcards, while the
      // contract says a prefix matches literally and case-sensitively. Then page
      // with a keyset cursor (keys strictly greater than the last one returned).
      const prefix = options?.prefix ?? ''
      const params: Array<string | number> = [prefix, prefix]
      let where = 'substr(key, 1, length(?)) = ?'
      if (options?.cursor !== undefined) {
        where += ' AND key > ?'
        params.push(options.cursor)
      }
      let sql = `SELECT * FROM blobs WHERE ${where} ORDER BY key ASC`
      const limit = options?.limit
      if (limit !== undefined) {
        sql += ' LIMIT ?' // fetch one extra row to detect truncation
        params.push(limit + 1)
      }
      const rows = db
        .prepare(sql)
        .all(...params)
        .map(mapBlobRecord)
      if (limit !== undefined && rows.length > limit) {
        const page = rows.slice(0, limit)
        const cursor = page.at(-1)?.key
        return {
          objects: page,
          truncated: true,
          ...(cursor !== undefined ? { cursor } : {}),
        }
      }
      return { objects: rows, truncated: false }
    },
  })
}

생성 어댑터 조합​

세 스토어를 같은 방식으로 defineAIPersistence에 전달합니다. generationRuns만 사용해도 유효한 생성 어댑터(실행 레코드만 있고 바이트 저장소는 없음)가 됩니다. 미디어를 유지하려면 artifacts + blobs를 함께 추가합니다.

import { DatabaseSync } from 'node:sqlite'
import { defineAIPersistence } from '@tanstack/ai-persistence'
// The three generation store factories and the schema string, from your modules.
import { createArtifactStore } from './artifact-store'
import { createBlobStore } from './blob-store'
import { createGenerationRunStore } from './generation-run-store'
import { GENERATION_SCHEMA_SQL } from './generation-schema'

export function generationPersistence(options: {
  url: string
  migrate?: boolean
}) {
  const db = new DatabaseSync(options.url)
  if (options.migrate) db.exec(GENERATION_SCHEMA_SQL)
  return defineAIPersistence({
    stores: {
      generationRuns: createGenerationRunStore(db),
      artifacts: createArtifactStore(db),
      blobs: createBlobStore(db),
    },
  })
}

그 결과를 generateImage / generateVideo / … 호출에서 withGenerationPersistence에 전달합니다. 생성 영속성을 참조합니다. 또한 composePersistence를 사용해 이 스토어를 기존 채팅 어댑터에 결합할 수 있으므로 하나의 백엔드가 withPersistence와 withGenerationPersistence를 모두 제공할 수 있습니다.

다음 단계​