33488fd634
Backend: - FastAPI WebSocket /ws endpoint with streaming via LangGraph astream - LangGraph Supervisor connecting 3 mock agents (order_lookup, order_actions, fallback) - YAML Agent Registry with Pydantic validation and immutable configs - PostgresSaver checkpoint persistence via langgraph-checkpoint-postgres - Session TTL with 30-min sliding window and interrupt extension - LLM provider abstraction (Anthropic/OpenAI/Google) - Token usage + cost tracking callback handler - Input validation: message size cap, thread_id format, content length - Security: no hardcoded defaults, startup API key validation, no input reflection Frontend: - React 19 + TypeScript + Vite chat UI - WebSocket hook with reconnect + exponential backoff - Streaming token display with agent attribution - Interrupt approval/reject UI for write operations - Collapsible tool call viewer Testing: - 87 unit tests, 87% coverage (exceeds 80% requirement) - Ruff lint + format clean Infrastructure: - Docker Compose (PostgreSQL 16 + backend) - pyproject.toml with full dependency management
62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
"""Database connection pool and PostgresSaver checkpoint setup."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
|
|
from psycopg.rows import dict_row
|
|
from psycopg_pool import AsyncConnectionPool
|
|
|
|
if TYPE_CHECKING:
|
|
from app.config import Settings
|
|
|
|
_CONVERSATIONS_DDL = """
|
|
CREATE TABLE IF NOT EXISTS conversations (
|
|
thread_id TEXT PRIMARY KEY,
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
last_activity TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
total_tokens INTEGER NOT NULL DEFAULT 0,
|
|
total_cost_usd DOUBLE PRECISION NOT NULL DEFAULT 0.0,
|
|
status TEXT NOT NULL DEFAULT 'active'
|
|
);
|
|
"""
|
|
|
|
_INTERRUPTS_DDL = """
|
|
CREATE TABLE IF NOT EXISTS active_interrupts (
|
|
interrupt_id TEXT PRIMARY KEY,
|
|
thread_id TEXT NOT NULL REFERENCES conversations(thread_id),
|
|
action TEXT NOT NULL,
|
|
params JSONB NOT NULL DEFAULT '{}',
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
|
resolved_at TIMESTAMPTZ,
|
|
resolution TEXT
|
|
);
|
|
"""
|
|
|
|
|
|
async def create_pool(settings: Settings) -> AsyncConnectionPool:
|
|
"""Create an async connection pool with the required psycopg settings."""
|
|
pool = AsyncConnectionPool(
|
|
conninfo=settings.database_url,
|
|
kwargs={"autocommit": True, "row_factory": dict_row},
|
|
min_size=2,
|
|
max_size=10,
|
|
)
|
|
await pool.open()
|
|
return pool
|
|
|
|
|
|
async def create_checkpointer(pool: AsyncConnectionPool) -> AsyncPostgresSaver:
|
|
"""Create and initialize the LangGraph checkpointer."""
|
|
checkpointer = AsyncPostgresSaver(conn=pool)
|
|
await checkpointer.setup()
|
|
return checkpointer
|
|
|
|
|
|
async def setup_app_tables(pool: AsyncConnectionPool) -> None:
|
|
"""Create application-specific tables (conversations, active_interrupts)."""
|
|
async with pool.connection() as conn:
|
|
await conn.execute(_CONVERSATIONS_DDL)
|
|
await conn.execute(_INTERRUPTS_DDL)
|