cmdc_memory_pg

CMDC PostgreSQL backend — Checkpoint + EpisodicMemory 持久化。

让 cmdc Agent 在 BEAM 节点重启 / 跨设备 / 跨进程的场景下，完整保留对话上下文 + 情景记忆。

v0.1 范围（严控二件套）

v0.1 明示不含

• ❌ pgvector 真语义检索 — similarity_search/3 降级为 ILIKE 文本匹配（与 ETS backend 同行为）
• ❌ 3-tier Memory（Working / Semantic / Procedural）— 留 v0.2
• ❌ Composite 路由 backend — 见 cmdc 主库 CMDC.Backend.Composite
• ❌ KV jsonb backend — 留 v0.2
• ❌ Cloak encryption 强制集成 — 给集成方留 CMDC.Checkpoint.Snapshot.redact/2 hook, 按需在 wrapper 层接 Cloak / KMS（详见 CheckpointBackend moduledoc）

安装

defp deps do
  [
    {:cmdc, "~> 0.5"},
    {:cmdc_memory_pg, "~> 0.1"}
  ]
end

配置

# config/runtime.exs
config :cmdc_memory_pg, CMDCMemoryPg.Repo,
  database: "cmdc_prod",
  username: System.fetch_env!("PGUSER"),
  password: System.fetch_env!("PGPASSWORD"),
  hostname: System.get_env("PGHOST", "localhost"),
  port: String.to_integer(System.get_env("PGPORT", "5432")),
  pool_size: 10

# 设为 CMDC.Checkpoint 默认 backend
config :cmdc, :checkpoint_backend, CMDCMemoryPg.CheckpointBackend

启动

defmodule MyApp.Application do
  def start(_type, _args) do
    children = [
      CMDCMemoryPg.Repo,
      # ... 其他子进程
    ]
    Supervisor.start_link(children, strategy: :one_for_one)
  end
end

Migration

$ mix ecto.create
$ mix ecto.migrate

migration 创建 2 张表：

使用

1. Checkpoint 持久化

# 抓快照
{:ok, snap} = CMDC.checkpoint!(session)

# 跨 BEAM 恢复
{:ok, snap} = CMDC.Checkpoint.load("sess-prod-001")
{:ok, new_session} = CMDC.resume_session!(snap)

无需指定 backend — 配置 :cmdc, :checkpoint_backend 后默认走 PG。

2. 情景记忆 few-shot

# 配置 EpisodicMemory Plugin 用 PG backend
{:ok, session} =
  CMDC.create_agent(
    model: "anthropic:claude-sonnet-4-5",
    user_data: %{user_id: "alice"},
    plugins: [
      {CMDC.Plugin.Builtin.EpisodicMemory,
       memory_store: :ignored,
       memory_module: CMDCMemoryPg.EpisodicMemoryBackend}
    ]
  )

# 成功对话自动写入；下次同用户类似 query 自动 few-shot 加载

与 Cloak 集成（可选 encryption at rest）

cmdc 主库提供 CMDC.Checkpoint.Snapshot.redact/2 helper — 集成方在 wrapper backend 层接 Cloak：

defmodule MyApp.EncryptedCheckpointBackend do
  @behaviour CMDC.Checkpoint.Backend

  @impl true
  def save(sid, snap, opts) do
    sanitized = CMDC.Checkpoint.Snapshot.redact(snap, &MyApp.Vault.encrypt/1)
    CMDCMemoryPg.CheckpointBackend.save(sid, sanitized, opts)
  end

  @impl true
  def load(sid, opts) do
    case CMDCMemoryPg.CheckpointBackend.load(sid, opts) do
      {:ok, snap} ->
        decrypted = CMDC.Checkpoint.Snapshot.redact(snap, &MyApp.Vault.decrypt/1)
        {:ok, decrypted}

      other -> other
    end
  end

  # list / delete 透传
  defdelegate list(sid, opts), to: CMDCMemoryPg.CheckpointBackend
  defdelegate delete(sid, opts), to: CMDCMemoryPg.CheckpointBackend
end

测试

测试套件分两层：

• 单元测试 — 验证 backend 逻辑 / 序列化 / 路径解析等（不依赖真实 PG）
• 集成测试 — 真实 PG，需 docker：

$ docker compose up -d
$ mix ecto.setup
$ mix test --include pg

不带 --include pg 时跳过 PG 集成测，便于纯单元 CI。

v0.2 路线图

License

Apache 2.0