asyncpg и aiopg:
PostgreSQL для AI-агентов

pip install asyncpg
# Или через uv (рекомендуется)
uv add asyncpg

import asyncio
import asyncpg

async def main():
    # Подключаемся к PostgreSQL
    conn = await asyncpg.connect(
        host="localhost",
        port=5432,
        database="agent_db",
        user="postgres",
        password="secret",
    )
    # Или через DSN:
    # conn = await asyncpg.connect("postgresql://postgres:secret@localhost/agent_db")

    # Создаём таблицу
    await conn.execute("""
        CREATE TABLE IF NOT EXISTS agent_memory (
            id SERIAL PRIMARY KEY,
            session_id TEXT NOT NULL,
            role TEXT NOT NULL,        -- 'user' или 'assistant'
            content TEXT NOT NULL,
            created_at TIMESTAMPTZ DEFAULT NOW()
        )
    """)

    # Вставляем запись
    await conn.execute(
        "INSERT INTO agent_memory (session_id, role, content) VALUES ($1, $2, $3)",
        "session-123", "user", "Как работает asyncio?"
    )

    # Получаем несколько строк
    rows = await conn.fetch(
        "SELECT * FROM agent_memory WHERE session_id = $1 ORDER BY created_at",
        "session-123"
    )
    for row in rows:
        print(f"[{row['role']}] {row['content']}")

    # Получаем одну строку
    row = await conn.fetchrow(
        "SELECT * FROM agent_memory WHERE id = $1", 1
    )
    print(row['content'])

    # Получаем скалярное значение
    count = await conn.fetchval(
        "SELECT COUNT(*) FROM agent_memory WHERE session_id = $1",
        "session-123"
    )
    print(f"Сообщений в сессии: {count}")

    await conn.close()

asyncio.run(main())

import asyncio
import asyncpg
from contextlib import asynccontextmanager

# Глобальный пул — создаётся один раз при старте приложения
pool: asyncpg.Pool | None = None

async def init_pool():
    global pool
    pool = await asyncpg.create_pool(
        dsn="postgresql://postgres:secret@localhost/agent_db",
        min_size=2,    # Минимум 2 постоянных подключения
        max_size=10,   # Максимум 10 (под нагрузкой)
        command_timeout=30,  # Таймаут запроса в секундах
    )

async def close_pool():
    global pool
    if pool:
        await pool.close()

# Использование пула: acquire + release через context manager
async def get_session_history(session_id: str) -> list[dict]:
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            """SELECT role, content, created_at
               FROM agent_memory
               WHERE session_id = $1
               ORDER BY created_at ASC""",
            session_id
        )
        return [dict(row) for row in rows]

async def save_message(session_id: str, role: str, content: str):
    async with pool.acquire() as conn:
        await conn.execute(
            "INSERT INTO agent_memory (session_id, role, content) VALUES ($1, $2, $3)",
            session_id, role, content
        )

# Транзакция: несколько операций атомарно
async def save_agent_turn(session_id: str, user_msg: str, agent_response: str):
    async with pool.acquire() as conn:
        async with conn.transaction():
            await conn.execute(
                "INSERT INTO agent_memory (session_id, role, content) VALUES ($1, $2, $3)",
                session_id, "user", user_msg
            )
            await conn.execute(
                "INSERT INTO agent_memory (session_id, role, content) VALUES ($1, $2, $3)",
                session_id, "assistant", agent_response
            )
            # Если здесь исключение — оба INSERT откатятся

# Пакетная вставка (executemany) — намного быстрее цикла
async def bulk_save_messages(messages: list[tuple[str, str, str]]):
    """messages: [(session_id, role, content), ...]"""
    async with pool.acquire() as conn:
        await conn.executemany(
            "INSERT INTO agent_memory (session_id, role, content) VALUES ($1, $2, $3)",
            messages
        )

async def main():
    await init_pool()
    try:
        await save_agent_turn("sess-1", "Привет!", "Привет! Чем могу помочь?")
        history = await get_session_history("sess-1")
        for msg in history:
            print(f"[{msg['role']}] {msg['content']}")
    finally:
        await close_pool()

asyncio.run(main())

import asyncpg
from pgvector.asyncpg import register_vector  # pip install pgvector

async def setup_vector_memory(conn: asyncpg.Connection):
    # Включаем расширение pgvector
    await conn.execute("CREATE EXTENSION IF NOT EXISTS vector")

    # Регистрируем тип vector в asyncpg (делается один раз на соединение)
    await register_vector(conn)

    # Создаём таблицу для долгосрочной памяти агента
    await conn.execute("""
        CREATE TABLE IF NOT EXISTS agent_facts (
            id SERIAL PRIMARY KEY,
            user_id TEXT NOT NULL,
            content TEXT NOT NULL,           -- Факт в текстовом виде
            embedding vector(1536),          -- OpenAI ada-002: 1536 измерений
            source TEXT,                     -- Откуда факт: 'conversation', 'document'
            created_at TIMESTAMPTZ DEFAULT NOW()
        )
    """)

    # Индекс для ускорения поиска (HNSW — лучший для production)
    await conn.execute("""
        CREATE INDEX IF NOT EXISTS agent_facts_embedding_idx
        ON agent_facts
        USING hnsw (embedding vector_cosine_ops)
    """)

import asyncio
import asyncpg
import numpy as np
from openai import AsyncOpenAI
from pgvector.asyncpg import register_vector

client = AsyncOpenAI()

async def get_embedding(text: str) -> list[float]:
    """Получить embedding через OpenAI API"""
    response = await client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

async def save_fact(pool: asyncpg.Pool, user_id: str, content: str, source: str = "conversation"):
    """Сохранить факт с embedding в долгосрочную память агента"""
    embedding = await get_embedding(content)

    async with pool.acquire() as conn:
        await register_vector(conn)
        await conn.execute(
            """INSERT INTO agent_facts (user_id, content, embedding, source)
               VALUES ($1, $2, $3, $4)""",
            user_id, content, embedding, source
        )
    print(f"✅ Факт сохранён: {content[:60]}...")

async def search_memory(pool: asyncpg.Pool, user_id: str, query: str, top_k: int = 5) -> list[dict]:
    """Найти релевантные факты по семантическому сходству"""
    query_embedding = await get_embedding(query)

    async with pool.acquire() as conn:
        await register_vector(conn)
        rows = await conn.fetch(
            """SELECT content, source, created_at,
                      1 - (embedding <=> $1) AS similarity  -- cosine similarity
               FROM agent_facts
               WHERE user_id = $2
               ORDER BY embedding <=> $1   -- ORDER BY расстоянию (меньше = ближе)
               LIMIT $3""",
            query_embedding, user_id, top_k
        )
    return [dict(row) for row in rows]

async def main():
    pool = await asyncpg.create_pool("postgresql://postgres:secret@localhost/agent_db")

    async with pool.acquire() as conn:
        await setup_vector_memory(conn)

    # Сохраняем факты о пользователе
    await save_fact(pool, "user-42", "Пользователь предпочитает краткие ответы без технических деталей")
    await save_fact(pool, "user-42", "Работает в сфере маркетинга, не разработчик")
    await save_fact(pool, "user-42", "Интересуется применением AI для анализа конкурентов")

    # Ищем релевантную память перед ответом
    query = "Какой формат ответов предпочтительнее для этого пользователя?"
    results = await search_memory(pool, "user-42", query, top_k=3)

    print(f"\nРелевантные факты для запроса: '{query}'")
    for r in results:
        print(f"  [{r['similarity']:.3f}] {r['content']}")

    await pool.close()

asyncio.run(main())

import asyncio
import aiopg  # pip install aiopg

DSN = "dbname=agent_db user=postgres password=secret host=localhost"

async def main():
    # Создаём пул
    async with aiopg.create_pool(DSN, minsize=2, maxsize=10) as pool:
        async with pool.acquire() as conn:
            async with conn.cursor() as cur:
                # API cursor-based (как psycopg2)
                await cur.execute(
                    "SELECT role, content FROM agent_memory WHERE session_id = %s",  # %s, не $1
                    ("session-123",)
                )
                rows = await cur.fetchall()
                for row in rows:
                    print(f"[{row[0]}] {row[1]}")

                # Вставка
                await cur.execute(
                    "INSERT INTO agent_memory (session_id, role, content) VALUES (%s, %s, %s)",
                    ("session-123", "user", "Привет!")
                )

asyncio.run(main())

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
import asyncpg

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: создаём пул
    app.state.db = await asyncpg.create_pool(
        dsn="postgresql://postgres:secret@localhost/agent_db",
        min_size=2,
        max_size=10,
    )
    yield
    # Shutdown: закрываем пул
    await app.state.db.close()

app = FastAPI(lifespan=lifespan)

@app.post("/chat/{session_id}")
async def chat(session_id: str, message: str, request: Request):
    pool: asyncpg.Pool = request.app.state.db

    # Сохраняем сообщение пользователя
    async with pool.acquire() as conn:
        await conn.execute(
            "INSERT INTO agent_memory (session_id, role, content) VALUES ($1, $2, $3)",
            session_id, "user", message
        )
        # Получаем историю для контекста
        history = await conn.fetch(
            "SELECT role, content FROM agent_memory WHERE session_id = $1 ORDER BY created_at DESC LIMIT 20",
            session_id
        )

    # ... вызов LLM с историей ...
    return {"status": "ok"}

import asyncio
import asyncpg

async def load_agent_context(pool: asyncpg.Pool, user_id: str, session_id: str, query: str):
    """
    Загружаем всё нужное параллельно:
    - история текущей сессии
    - релевантные факты из долгосрочной памяти
    - профиль пользователя
    """

    async def get_history():
        async with pool.acquire() as conn:
            return await conn.fetch(
                "SELECT role, content FROM agent_memory WHERE session_id = $1 ORDER BY created_at DESC LIMIT 10",
                session_id
            )

    async def get_relevant_facts():
        embedding = await get_embedding(query)  # API-запрос к OpenAI
        async with pool.acquire() as conn:
            from pgvector.asyncpg import register_vector
            await register_vector(conn)
            return await conn.fetch(
                "SELECT content FROM agent_facts WHERE user_id = $1 ORDER BY embedding <=> $2 LIMIT 5",
                user_id, embedding
            )

    async def get_user_profile():
        async with pool.acquire() as conn:
            return await conn.fetchrow(
                "SELECT name, preferences, language FROM users WHERE id = $1",
                user_id
            )

    # Все три запроса выполняются одновременно
    history, facts, profile = await asyncio.gather(
        get_history(),
        get_relevant_facts(),
        get_user_profile(),
    )

    return {
        "history": [dict(r) for r in history],
        "facts": [r["content"] for r in facts],
        "profile": dict(profile) if profile else {},
    }

import asyncpg

# --- Подключение ---
conn = await asyncpg.connect(dsn="postgresql://user:pass@host/db")
pool = await asyncpg.create_pool(dsn=..., min_size=2, max_size=10)

# --- Запросы ---
await conn.execute("INSERT INTO t VALUES ($1, $2)", val1, val2)
rows = await conn.fetch("SELECT * FROM t WHERE id = $1", id)       # → list[Record]
row  = await conn.fetchrow("SELECT * FROM t WHERE id = $1", id)    # → Record | None
val  = await conn.fetchval("SELECT COUNT(*) FROM t")               # → Any
await conn.executemany("INSERT INTO t VALUES ($1)", [(1,), (2,)])

# --- Пул ---
async with pool.acquire() as conn:
    rows = await conn.fetch(...)

# --- Транзакция ---
async with pool.acquire() as conn:
    async with conn.transaction():
        await conn.execute(...)
        await conn.execute(...)  # откат если исключение

# --- pgvector ---
from pgvector.asyncpg import register_vector
async with pool.acquire() as conn:
    await register_vector(conn)
    # Сохранить
    await conn.execute("INSERT INTO t (emb) VALUES ($1)", [0.1, 0.2, ...])
    # Поиск по косинусному сходству
    rows = await conn.fetch(
        "SELECT *, 1 - (emb <=> $1) AS sim FROM t ORDER BY emb <=> $1 LIMIT 5",
        query_embedding
    )

# --- aiopg (psycopg2 стиль) ---
import aiopg
async with aiopg.create_pool(dsn) as pool:
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT * FROM t WHERE id = %s", (1,))
            rows = await cur.fetchall()  # list[tuple]