《AI Agent 工程实战》5.1 Schema 设计：会话/消息/运行/记忆表 + 迁移

CREATE TABLE sessions (
    id           uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    workspace_id text NOT NULL,
    title        text NOT NULL DEFAULT '',
    created_at   timestamptz NOT NULL DEFAULT now(),
    updated_at   timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE messages (
    id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    workspace_id text NOT NULL,
    session_id   uuid NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
    role         text NOT NULL CHECK (role IN ('system','user','assistant','tool')),
    payload      jsonb NOT NULL,
    created_at   timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE runs (
    id           uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    workspace_id text NOT NULL,
    session_id   uuid NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
    status       text NOT NULL DEFAULT 'queued'
                 CHECK (status IN ('queued','running','succeeded','failed','cancelled')),
    attempts     int  NOT NULL DEFAULT 0,
    error        text,
    created_at   timestamptz NOT NULL DEFAULT now(),
    started_at   timestamptz,
    finished_at  timestamptz
);

CREATE TABLE memories (
    id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    workspace_id text NOT NULL,
    user_id      text NOT NULL,
    kind         text NOT NULL DEFAULT 'fact',
    content      text NOT NULL,
    embedding    real[] NOT NULL DEFAULT '{}',
    created_at   timestamptz NOT NULL DEFAULT now(),
    expires_at   timestamptz
);

CREATE INDEX idx_sessions_ws_updated ON sessions (workspace_id, updated_at DESC);
CREATE INDEX idx_runs_ws_status     ON runs (workspace_id, status, created_at);

SELECT payload->'tool_calls'->0->>'name'                    AS tool,
       payload->'tool_calls'->0->'arguments'->>'q'          AS q
FROM messages
WHERE session_id = $1 AND payload ? 'tool_calls';

docker run -d --name ab-pg \
  -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=research \
  -p 15435:5432 postgres:17-alpine

import json, psycopg

DSN = "postgresql://postgres:postgres@localhost:15435/research"

with psycopg.connect(DSN) as conn:
    sid = conn.execute(
        "INSERT INTO sessions (workspace_id, title) VALUES (%s, %s) RETURNING id",
        ("ws-acme", "向量检索调研")).fetchone()[0]
    msgs = [
        ("user", {"role": "user", "content": "帮我查一下 pgvector 的召回率"}),
        ("assistant", {"role": "assistant", "content": "",
                       "tool_calls": [{"id": "c1", "name": "search",
                                       "arguments": {"q": "pgvector recall"}}]}),
        ("tool", {"role": "tool", "tool_call_id": "c1", "content": "命中 3 篇"}),
        ("assistant", {"role": "assistant", "content": "召回率见报告第 2 节。"}),
    ]
    with conn.cursor() as cur:
        cur.executemany(
            "INSERT INTO messages (workspace_id, session_id, role, payload) VALUES (%s,%s,%s,%s)",
            [("ws-acme", sid, r, json.dumps(p, ensure_ascii=False)) for r, p in msgs])
    conn.execute("INSERT INTO runs (workspace_id, session_id, status) VALUES (%s,%s,'queued')",
                 ("ws-acme", sid))
    conn.execute("INSERT INTO memories (workspace_id, user_id, kind, content) VALUES (%s,%s,%s,%s)",
                 ("ws-acme", "u-42", "preference", "偏好带引用的简短结论"))
    conn.commit()

session_id = 1ac365cd-430f-481d-8ccf-293e301a679e
  #1 user      '帮我查一下 pgvector 的召回率'
  #2 assistant ''   tool_calls=[{"id": "c1", "name": "search", "arguments": {"q": "pgvector recall"}}]
  #3 tool      '命中 3 篇'
  #4 assistant '召回率见报告第 2 节。'

-- 只取工具调用，不用改表结构 --
  tool=search  q='pgvector recall'

# migrate.py
import pathlib, psycopg

DSN = "postgresql://postgres:postgres@localhost:15435/research"
MIGRATIONS = pathlib.Path("migrations")

def migrate(dsn: str = DSN) -> list[str]:
    applied: list[str] = []
    with psycopg.connect(dsn, autocommit=True) as conn:
        conn.execute("CREATE TABLE IF NOT EXISTS schema_migrations ("
                     "version text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())")
        conn.execute("SELECT pg_advisory_lock(42001)")   # 同一时刻只允许一个进程改结构
        try:
            done = {r[0] for r in conn.execute("SELECT version FROM schema_migrations")}
            for path in sorted(MIGRATIONS.glob("*.sql")):
                if path.name in done:
                    continue
                with conn.transaction():                 # 每个迁移是一个事务
                    conn.execute(path.read_text())
                    conn.execute("INSERT INTO schema_migrations(version) VALUES (%s)", (path.name,))
                applied.append(path.name)
        finally:
            conn.execute("SELECT pg_advisory_unlock(42001)")
    return applied

if __name__ == "__main__":
    print("applied:", migrate())

$ python migrate.py
applied: ['001_init.sql']
$ python migrate.py
applied: []