面向 AI Agent 的数据库：共享实时数据，但不与生产系统共担风险

CREATE ROLE agent_reader LOGIN PASSWORD 'replace-with-a-secret';
GRANT CONNECT ON DATABASE appdb TO agent_reader;
GRANT USAGE ON SCHEMA public TO agent_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO agent_reader;
ALTER DEFAULT PRIVILEGES IN SCHEMA public
  GRANT SELECT ON TABLES TO agent_reader;
ALTER ROLE agent_reader SET default_transaction_read_only = on;
ALTER ROLE agent_reader SET statement_timeout = '2s';
ALTER ROLE agent_reader SET lock_timeout = '250ms';

python3 -m venv .venv
source .venv/bin/activate
pip install 'psycopg[binary]==3.2.6'

# burst_read_test.py
import os
import time
import statistics
from concurrent.futures import ThreadPoolExecutor, as_completed

import psycopg

DSN = os.environ["DATABASE_URL"]
QUERY = os.getenv("QUERY", "SELECT now(), current_database()")
WORKERS = int(os.getenv("WORKERS", "20"))
REQUESTS_PER_WORKER = int(os.getenv("REQUESTS_PER_WORKER", "50"))


def run_worker(worker_id: int):
    latencies_ms = []
    errors = []

    try:
        with psycopg.connect(DSN, autocommit=False) as conn:
            for _ in range(REQUESTS_PER_WORKER):
                started = time.perf_counter()
                try:
                    with conn.transaction():
                        conn.execute("SET TRANSACTION READ ONLY")
                        conn.execute("SET LOCAL statement_timeout = '2s'")
                        with conn.cursor() as cur:
                            cur.execute(QUERY)
                            cur.fetchall()
                    latencies_ms.append((time.perf_counter() - started) * 1000)
                except Exception as exc:
                    errors.append(f"{type(exc).__name__}: {exc}")
    except Exception as exc:
        errors.append(f"connect: {type(exc).__name__}: {exc}")

    return latencies_ms, errors


def percentile(values, p):
    if not values:
        return float("nan")
    ordered = sorted(values)
    index = min(len(ordered) - 1, int((len(ordered) - 1) * p))
    return ordered[index]


started = time.perf_counter()
all_latencies = []
all_errors = []

with ThreadPoolExecutor(max_workers=WORKERS) as pool:
    futures = [pool.submit(run_worker, i) for i in range(WORKERS)]
    for future in as_completed(futures):
        latencies, errors = future.result()
        all_latencies.extend(latencies)
        all_errors.extend(errors)

elapsed = time.perf_counter() - started
completed = len(all_latencies)

print(f"workers={WORKERS}")
print(f"completed={completed} errors={len(all_errors)} elapsed_s={elapsed:.2f}")
print(f"throughput_qps={completed / elapsed:.2f}")
if all_latencies:
    print(f"mean_ms={statistics.mean(all_latencies):.2f}")
    print(f"p50_ms={percentile(all_latencies, 0.50):.2f}")
    print(f"p95_ms={percentile(all_latencies, 0.95):.2f}")
    print(f"p99_ms={percentile(all_latencies, 0.99):.2f}")
if all_errors:
    print("sample_errors:")
    for error in all_errors[:5]:
        print(f"- {error}")

export DATABASE_URL='postgresql://agent_reader:replace-with-a-secret@db-host:5432/appdb?sslmode=require'
export QUERY='SELECT id, status FROM orders WHERE id = 10001'
export WORKERS=50
export REQUESTS_PER_WORKER=100
python burst_read_test.py