데이터베이스 연결

DB 연결은 노드가 아니라 애플리케이션 수명주기에 맞춰 관리합니다. 호출마다 연결하면 TCP·TLS 비용이 반복되고 동시 요청이 늘 때 max_connections에 쉽게 닿습니다.

선택	맞는 경우
SDK `register_pg_pool()`	단순 조회·쓰기, SQL 직접 작성, 의존성 최소화
SQLAlchemy 2.x	모델·관계·트랜잭션·Alembic이 필요함

비즈니스 DB는 POSTGRES_URL이나 전용 DSN을 사용하세요. 세션 메모리의 POSTGRES_MEMORY_DSN, 오케스트레이터 상태의 POSTGRES_ORCH_DSN과 역할을 섞지 않습니다.

asyncpg 공유 풀

부팅 때 등록

from llamon_agent.config import RuntimeEnv
from llamon_agent.core.postgres import PostgresPoolConfig, register_pg_pool
from llamon_agent.graph import END, START, GraphBuilder

from app.nodes import lookup_node


async def build_graph(settings):
    env = RuntimeEnv(source_file=__file__)

    await register_pg_pool(
        name="lookup_db",
        config=PostgresPoolConfig(
            dsn=env.get("POSTGRES_LOOKUP_DSN", ""),
            max_size=5,
            command_timeout=3.0,
        ),
        required=False,
    )

    return (
        GraphBuilder()
        .node("lookup", lookup_node, node_kind="business")
        .edge(START, "lookup")
        .edge("lookup", END)
        .build()
    )

SDK 실행 수명주기가 모든 풀을 닫으므로 별도 종료 hook은 필요하지 않습니다. register_pg_pool()은 초기화 직렬화, 시간 제한, DSN 가림을 함께 처리합니다.

노드에서 사용

from llamon_agent.core.postgres import get_pg_pool


async def lookup_node(state):
    pool = get_pg_pool("lookup_db")
    if pool is None:
        return {"output": {"output_text": "조회 결과가 없습니다.", "output_data": []}}

    async with pool.acquire() as connection:
        rows = await connection.fetch(
            "SELECT id, name FROM my_table WHERE code = ANY($1::text[])",
            state["codes"],
        )
    return {"output": {"output_data": [dict(row) for row in rows]}}

옵션	기본값	의미
`min_size`	`1`	최소 연결 수. PgBouncer에서는 `0` 검토
`max_size`	`5`	이 프로세스가 열 수 있는 최대 연결 수
`command_timeout`	`3.0`	쿼리 시간 제한(초)
`connect_timeout`	`15.0`	연결 수립 시간 제한(초)
`required`	`False`	실패 시 부팅을 막을지 결정

선택 조회가 실패해도 서비스가 계속 동작해야 한다면 required=False로 두고 None을 처리합니다. DB 없이는 답변이 성립하지 않는다면 required=True로 부팅을 중단하세요. 기능 축소를 잘못 적용하면 빈 결과가 정상 응답처럼 보일 수 있습니다.

DB가 여러 개면 이름을 달리해 각각 등록합니다. 풀 이름은 상수로 한곳에 두고 쿼리는 app/internals/나 repository 모듈로 분리하세요. 단위 테스트에서는 set_provider(PostgresPoolRegistry())로 격리된 registry를 주입하고 끝난 뒤 원래 provider를 복원합니다.

SQLAlchemy ORM

모델과 트랜잭션 경계가 필요하면 SQLAlchemy 비동기 엔진을 프로세스당 한 번 만들고 요청마다 짧은 AsyncSession을 엽니다.

uv add "SQLAlchemy>=2.0" asyncpg alembic

권장 구조는 다음과 같습니다.

파일	역할
`app/db.py`	엔진, sessionmaker, commit·rollback, 종료 처리
`app/models.py`	ORM 모델
`app/repositories/*.py`	SQLAlchemy 쿼리
`app/nodes.py`	state 처리와 repository 호출
`.env`	`POSTGRES_URL` 실제 값

from contextlib import asynccontextmanager

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from llamon_agent.config import RuntimeEnv

env = RuntimeEnv(source_file=__file__)
postgres_url = env.get("POSTGRES_URL", "").strip()
if not postgres_url:
    raise RuntimeError("POSTGRES_URL이 설정되지 않았습니다.")
if postgres_url.startswith("postgresql://"):
    postgres_url = "postgresql+asyncpg://" + postgres_url.removeprefix("postgresql://")
elif postgres_url.startswith("postgres://"):
    postgres_url = "postgresql+asyncpg://" + postgres_url.removeprefix("postgres://")
engine = create_async_engine(
    postgres_url,
    pool_size=5,
    max_overflow=0,
    pool_timeout=5,
    pool_pre_ping=True,
    connect_args={"timeout": 10, "command_timeout": 10},
)
sessions = async_sessionmaker(engine, expire_on_commit=False)


@asynccontextmanager
async def session_scope():
    async with sessions() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


async def close_db():
    await engine.dispose()

AsyncSession을 전역으로 공유하지 마세요. 노드는 repository 함수만 호출하고 SQL과 트랜잭션 처리는 repository와 session_scope()에 맡깁니다. graph의 __llamon_shutdown__ 훅에는 close_db()를 연결합니다.

ORM 모델에는 DB 제약이 없더라도 행을 식별할 primary_key=True가 필요합니다. 앱이 소유하는 스키마는 Alembic revision으로 관리하고 운영 코드에서 Base.metadata.create_all()을 호출하지 않습니다.

uv run alembic revision --autogenerate -m "create tables"
uv run alembic upgrade head

SQLAlchemy asyncpg는 prepared statement를 사용합니다. PgBouncer는 직접 연결이나 session 모드를 먼저 검토하고 외부 pooler가 필수라면 배포 환경에서 NullPool을 평가하세요.

실패 처리

DB가 필수면 표준 애플리케이션 오류로 올립니다. 보조 조회라면 metadata에 skipped 또는 failed를 남기고 안전한 빈 결과를 반환해도 됩니다. 어느 정책을 택하든 연결 오류를 정상 데이터와 구분해야 합니다.

메모리 저장소 구성은 멀티턴 메모리, 연결 진단은 문제 해결을 참고하세요.