Docs

Examples

Raw SQL Agent Memory

Persist Anvia agent session memory with a raw SQL adapter.

Use a raw SQL MemoryStore when your application has a direct database client, a small storage layer, or a migration system that does not use an ORM.

Scenario

A Node service talks to Postgres through pg. The agent receives a stable session id from the product route, then the memory store loads and appends full Anvia Message objects as JSON.

Table

CREATE EXTENSION IF NOT EXISTS pgcrypto;

CREATE TABLE agent_memory_sessions (
  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  scope_key text NOT NULL UNIQUE,
  session_id text NOT NULL,
  user_id text,
  tenant_id text,
  metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX agent_memory_sessions_scope_idx
  ON agent_memory_sessions (session_id, user_id, tenant_id);

CREATE TABLE agent_memory_messages (
  id bigserial PRIMARY KEY,
  memory_session_id uuid NOT NULL REFERENCES agent_memory_sessions(id) ON DELETE CASCADE,
  run_id text NOT NULL,
  turn integer NOT NULL,
  position integer NOT NULL,
  role text NOT NULL,
  message jsonb NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE UNIQUE INDEX agent_memory_messages_position_idx
  ON agent_memory_messages (memory_session_id, position);

Expected Message JSON

MemoryStore.load(...) returns Promise<Message[]>. This example stores one Anvia Message object per row, then returns the ordered row set as an array.

The message union is:

type Message = SystemMessage | UserMessage | AssistantMessage | ToolMessage;
type UserContent = Text | ImageContent | DocumentContent;
type AssistantContent = Text | ImageContent | Reasoning | ToolCall;
type ToolContent = ToolResult;

The full shape returned from load(...) is an array. This illustrative array includes every supported message and content variant; real transcripts usually contain only the variants produced by that run.

[
  {
    "role": "system",
    "content": "Stable runtime instructions."
  },
  {
    "role": "user",
    "content": [
      {
        "type": "text",
        "text": "Where is order A-100?",
        "signature": "sig_user_text"
      },
      {
        "type": "image",
        "source": { "type": "url", "url": "https://files.example.com/photo.png" },
        "detail": "high"
      },
      {
        "type": "image",
        "source": {
          "type": "base64",
          "data": "iVBORw0KGgo=",
          "mediaType": "image/png"
        },
        "detail": "low"
      },
      {
        "type": "document",
        "source": {
          "type": "url",
          "url": "https://files.example.com/invoice.pdf",
          "mediaType": "application/pdf",
          "filename": "invoice.pdf"
        }
      },
      {
        "type": "document",
        "source": {
          "type": "base64",
          "data": "JVBERi0xLjQ=",
          "mediaType": "application/pdf",
          "filename": "invoice-copy.pdf"
        }
      },
      {
        "type": "document",
        "source": {
          "type": "text",
          "text": "Invoice text extracted upstream.",
          "mediaType": "text/plain",
          "filename": "invoice.txt"
        }
      }
    ]
  },
  {
    "role": "assistant",
    "id": "msg_assistant_1",
    "content": [
      {
        "type": "text",
        "text": "I will look that up.",
        "signature": "sig_assistant_text"
      },
      {
        "type": "image",
        "source": { "type": "url", "url": "https://files.example.com/generated.png" },
        "detail": "auto"
      },
      {
        "type": "image",
        "source": {
          "type": "base64",
          "data": "iVBORw0KGgo=",
          "mediaType": "image/png"
        }
      },
      {
        "type": "reasoning",
        "id": "rs_1",
        "text": "Checked order status and summarized the result.",
        "content": [
          {
            "type": "text",
            "text": "Need live order state.",
            "signature": "sig_reasoning_text"
          },
          {
            "type": "summary",
            "text": "Order lookup is required."
          },
          {
            "type": "encrypted",
            "data": "encrypted_reasoning_blob"
          },
          {
            "type": "redacted",
            "data": "redacted_reasoning_blob"
          }
        ]
      },
      {
        "type": "tool_call",
        "id": "call_1",
        "callId": "fc_1",
        "function": {
          "name": "lookup_order",
          "arguments": { "orderId": "A-100" }
        },
        "signature": "sig_tool_call",
        "additionalParams": { "providerToolCallId": "provider_call_1" }
      }
    ]
  },
  {
    "role": "tool",
    "content": [
      {
        "type": "tool_result",
        "id": "call_1",
        "callId": "fc_1",
        "content": [
          {
            "type": "text",
            "text": "{\"status\":\"shipped\"}"
          },
          {
            "type": "image",
            "data": "iVBORw0KGgo=",
            "mediaType": "image/png"
          }
        ]
      }
    ]
  },
  {
    "role": "assistant",
    "content": [{ "type": "text", "text": "Order A-100 has shipped." }]
  }
]

Each row’s message column contains one item from that array.

Memory Store

import type { Pool } from "pg";
import { type MemoryStore, type Message } from "@anvia/core";
import type { MemoryAppendInput, MemoryContext } from "@anvia/core/memory";

function tenantId(context: MemoryContext) {
  const value = context.metadata?.tenantId;
  return typeof value === "string" ? value : null;
}

function scopeValues(context: MemoryContext) {
  return [context.sessionId, context.userId ?? null, tenantId(context)] as const;
}

function scopeKey(context: MemoryContext) {
  return JSON.stringify(scopeValues(context));
}

function metadataJson(context: MemoryContext) {
  return JSON.stringify(context.metadata ?? {});
}

function fromJson(value: unknown): Message {
  return typeof value === "string" ? (JSON.parse(value) as Message) : (value as Message);
}

export class SqlMemoryStore implements MemoryStore {
  constructor(private readonly pool: Pool) {}

  async load(context: MemoryContext): Promise<Message[]> {
    const { rows } = await this.pool.query<{ message: unknown }>(
      `SELECT messages.message
       FROM agent_memory_sessions sessions
       JOIN agent_memory_messages messages
         ON messages.memory_session_id = sessions.id
       WHERE sessions.scope_key = $1
       ORDER BY messages.position ASC`,
      [scopeKey(context)],
    );

    return rows.map((row) => fromJson(row.message));
  }

  async append(input: MemoryAppendInput): Promise<void> {
    if (input.messages.length === 0) {
      return;
    }

    const client = await this.pool.connect();

    try {
      await client.query("BEGIN");
      const key = scopeKey(input.context);
      await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [key]);

      const { rows: sessionRows } = await client.query<{ id: string }>(
        `INSERT INTO agent_memory_sessions
           (scope_key, session_id, user_id, tenant_id, metadata)
         VALUES ($1, $2, $3, $4, $5::jsonb)
         ON CONFLICT (scope_key) DO UPDATE
           SET updated_at = now(),
               metadata = EXCLUDED.metadata
         RETURNING id`,
        [
          key,
          input.context.sessionId,
          input.context.userId ?? null,
          tenantId(input.context),
          metadataJson(input.context),
        ],
      );
      const memorySessionId = sessionRows[0]?.id;
      if (memorySessionId === undefined) {
        throw new Error("Failed to resolve memory session.");
      }

      const { rows } = await client.query<{ position: number }>(
        `SELECT COALESCE(MAX(position), -1) AS position
         FROM agent_memory_messages
         WHERE memory_session_id = $1`,
        [memorySessionId],
      );
      const start = rows[0]?.position ?? -1;

      for (const [index, message] of input.messages.entries()) {
        await client.query(
          `INSERT INTO agent_memory_messages
             (memory_session_id, run_id, turn, position, role, message)
           VALUES ($1, $2, $3, $4, $5, $6::jsonb)`,
          [
            memorySessionId,
            input.runId,
            input.turn,
            start + index + 1,
            message.role,
            JSON.stringify(message),
          ],
        );
      }

      await client.query("COMMIT");
    } catch (error) {
      await client.query("ROLLBACK");
      throw error;
    } finally {
      client.release();
    }
  }

  async clear(context: MemoryContext): Promise<void> {
    await this.pool.query(
      `DELETE FROM agent_memory_sessions
       WHERE scope_key = $1`,
      [scopeKey(context)],
    );
  }
}

Use It

import { Pool } from "pg";
import { AgentBuilder } from "@anvia/core";

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const memoryStore = new SqlMemoryStore(pool);

const agent = new AgentBuilder("support", model)
  .instructions("Use prior memory before answering.")
  .memory(memoryStore, { savePolicy: "turn" })
  .build();

const response = await agent
  .session("thread_123", {
    userId: "user_456",
    metadata: { tenantId: "tenant_789" },
  })
  .prompt("Continue from the previous support answer.")
  .send();

console.log(response.output);

Production Checks

  • Use scope_key for one canonical nullable user and tenant scope lookup.
  • Keep session ownership and lifecycle metadata on agent_memory_sessions.
  • Keep full prompt history rows in agent_memory_messages.
  • Use transactions and advisory locks before allowing concurrent writes to the same conversation.

Next Patterns