7+1개 에이전트가 하나의 SQL을 만드는 과정 [2]

왜 에이전트를 나눴나#

1편에서 14개 기업 사례를 보면서도 처음엔 솔직히 좀 과하다고 생각했다. SQL 하나 만드는 데 에이전트를 그렇게 많이 붙여야 하나 싶었다. 프롬프트 하나에 "이 스키마 보고 SQL 만들어"라고 던지면 되는 거 아닌가 하는 쪽이었다.

근데 직접 붙여보니 왜 다들 그 구조로 가는지 바로 보였다.

테이블 10개 정도까지는 한 번에 넣어도 그럭저럭 된다. 그런데 50개가 넘어가면 프롬프트가 수만 토큰이 되면서 LLM이 엉뚱한 테이블을 참조하기 시작한다. 게다가 의도 파악, 테이블 선택, SQL 생성, 검증을 한 번에 시키면 틀렸을 때 어디서 깨졌는지도 안 보인다.

에이전트를 나누고 나니 두 가지가 풀렸다:

1. 각 단계의 입력이 작아져서 정확도가 올라갔다
2. 어디서 틀렸는지 바로 보여서 디버깅이 쉬워졌다

전체 파이프라인#

구조는 LangGraph 기반으로 잡았습니다. 각 에이전트가 노드이고, 공유 상태(GraphState)를 통해 데이터를 주고받습니다.

사용자 질문
    ↓
[Tenant Resolve] → API Key로 테넌트 식별
    ↓
[Intent Agent] → 의도 분류 + 복잡도 판단
    ↓ (SQL_QUERY가 아니면 여기서 빠짐)
[Cache Check] → 동일 질문 캐시 히트 시 바로 반환
    ↓ (Miss)
[Table Agent] → Vector Search → Rerank → 관련 테이블 선별
    ↓
[Column Prune Agent] → 불필요한 컬럼 제거 (COMPLEX만)
    ↓
[Context Assembly] → 샘플 쿼리 + 대화 히스토리 + 용어 사전 조합
    ↓
[Schema Linking] → 테이블 간 관계(FK) 파악
    ↓
[SQL Agent] → SQL 생성 (Chain-of-Thought)
    ↓
[Guardrail] → 위험 구문 필터링
    ↓
[Validation + Self-Correction] → 검증 + 자동 수정 (최대 3회)
    ↓
[Executor] → READ-ONLY 실행 (30초 타임아웃)
    ↓
[Result Agent] → 결과 요약 + 인사이트 + 후속 질문 제안

7개 코어 에이전트 + 1개 선택적 에이전트(Decomposer) 구조입니다. 처음에 블로그 초안을 쓸 때는 12개라고 적어놨는데, 다시 보니 cache_check나 tenant_resolve 같은 non-LLM 노드까지 같이 세고 있었습니다. 실제로 LLM을 호출하는 쪽은 7+1개입니다.

Understand: 의도를 먼저 파악#

Intent Agent#

모든 질문이 SQL로 바꿀 수 있는 건 아닙니다. Intent Agent가 세 가지로 분류합니다:

• SQL_QUERY: 데이터 조회 질문 → 파이프라인 진행
• CLARIFICATION_NEEDED: 모호한 질문 → 사용자에게 되물음
• GENERAL: 일반 대화 → SQL 파이프라인 태우지 않고 직접 응답

여기서 복잡도(SIMPLE / COMPLEX)도 같이 판단합니다. SIMPLE이면 Column Pruning을 건너뛰어서 토큰과 시간을 아낍니다.

Intent Agent에는 비용이 작은 모델(Haiku 4.5)을 씁니다. 의도 분류는 복잡한 추론이 필요 없기 때문에 작은 모델로도 충분하고, 매 요청마다 호출되는 만큼 비용을 줄여야 합니다. 나머지 에이전트는 Sonnet 4.5를 씁니다.

Decomposer Agent (선택적)#

복합 질문이 들어오면 하위 질문으로 분해합니다.

"부서별 주문 건수와 각 부서의 평균 매출을 비교해줘"
    ↓
Sub Q1: "부서별 주문 건수"
Sub Q2: "부서별 평균 매출"
    ↓
각각 Query Subgraph로 병렬 실행 → 결과 병합

항상 활성화되는 건 아니고, Intent Agent가 복합 질문으로 판단했을 때만 동작합니다. 단일 질문이면 건너뜁니다.

Retrieve: 수백 개 테이블에서 찾기#

Table Agent — Multi-stage Ranking#

테이블 선별이 전체 정확도의 핵심입니다. 잘못된 테이블을 선택하면 이후 단계에서 아무리 잘 해도 틀린 SQL이 나옵니다.

LinkedIn의 Multi-stage Ranking 패턴을 참고해서 2단계 파이프라인을 구성했습니다.

1단계: Vector Search (Qdrant)

질문을 OpenAI text-embedding-3-small(1536차원)로 임베딩하고, Qdrant에서 상위 20개 테이블 후보를 가져옵니다. 이 단계는 재현율(Recall)이 높은 대신 정밀도(Precision)는 낮습니다. 관련 없는 테이블도 섞여 들어오는 게 당연합니다.

테넌트별로 Qdrant 컬렉션을 분리해서 다른 테넌트의 스키마가 섞이지 않게 합니다. 테넌트당 3개 컬렉션: tables_{id}, columns_{id}, sample_queries_{id}.

2단계: Rerank (Cohere)

Cohere rerank-v3.5로 20개 후보를 재순위화해서 상위 5개를 선택합니다. Cross-encoder 기반이라 질문과 테이블 설명 사이의 관련성을 1단계보다 정밀하게 평가합니다.

blog-series-plan 초안에 "3단계(Vector → Rerank → LLM Ranking)"로 적혀있었는데, 코드를 확인해보니 LLM Ranking 단계는 구현하지 않았습니다. 2단계만으로도 충분한 정밀도가 나왔기 때문입니다.

Column Prune Agent#

선택된 테이블에서 질문과 무관한 컬럼을 제거합니다. 테이블 하나에 컬럼이 50개씩 있으면 SQL Agent에 전달되는 컨텍스트가 너무 길어집니다. 실제로 필요한 건 5-10개뿐인데 50개를 다 넣으면 토큰 낭비이고 LLM이 혼란스러워합니다.

SIMPLE 쿼리는 이 단계를 건너뜁니다. 단일 테이블이면 컬럼이 많아도 LLM이 잘 처리하기 때문입니다.

Context Assembly#

SQL 생성에 필요한 컨텍스트를 이 단계에서 한꺼번에 모아 조립합니다:

• 선택된 테이블의 DDL: 컬럼 타입, 제약조건 포함
• 샘플 쿼리: 비슷한 질문에 대한 과거 성공 SQL (RAG에서 검색)
• 대화 히스토리: 이전 질문에서 참조한 테이블 정보
• 용어 사전: 비즈니스 용어 매핑 (예: "활성 사용자" = user_activity 테이블, 최근 30일)

용어 사전은 3계층 구조입니다: User 레벨 > Tenant 레벨 > Base 레벨. 사용자가 직접 정의한 용어가 최우선, 그 다음 테넌트 공통 용어, 마지막으로 시스템 기본 용어입니다.

Schema Linking#

선택된 테이블 간의 FK 관계를 파악합니다. JOIN 조건이 틀리는 경우 대부분 이 단계가 빠졌을 때입니다. FK 1-hop 이웃까지 확장해서, 직접 선택되지 않았지만 JOIN에 필요한 브릿지 테이블도 포함시킵니다.

Generate: SQL 생성 + 검증#

SQL Agent#

Chain-of-Thought 방식으로 SQL을 생성합니다. 단순히 SQL만 출력하는 게 아니라, 추론 과정(thinking), 사용한 테이블, 신뢰도(confidence), 필수 조건 누락 시 명확화 질문까지 함께 반환합니다.

class SQLGenerationResult(BaseModel):
    thinking: str          # 추론 과정
    sql: str               # 생성된 SQL
    tables_used: list[str] # 사용한 테이블
    confidence: float      # 0.0 ~ 1.0
    clarification_question: str | None  # 필수 조건 누락 시

confidence가 낮으면 SQL을 실행하지 않고 사용자에게 되묻습니다. 틀린 결과를 주는 것보다 미리 확인하는 게 낫습니다.

DB 방언(dialect)도 고려합니다. MySQL과 PostgreSQL은 함수명이나 문법이 다른 부분이 있어서, 테넌트의 DB 종류에 맞는 SQL을 생성하도록 방언별 프롬프트를 분리해뒀습니다.

Guardrail#

LLM이 뱉은 SQL을 검증 없이 실행하는 건 당연히 위험합니다. 4개 레이어로 안전장치를 뒀습니다:

• SQL Filter: DROP, DELETE, UPDATE 같은 위험 구문 차단. 기본적으로 SELECT만 허용합니다
• PII Masker: 결과에 개인정보(이름, 이메일 등)가 포함되면 마스킹
• Audit Log: 모든 질문, 생성된 SQL, 실행 결과를 기록
• Credential Manager: DB 접속 정보를 안전하게 관리

Validation + Self-Correction#

생성된 SQL을 3단계로 검증합니다:

1. Syntax Check: sqlparse로 구문 파싱 가능한지 확인
2. Schema Check: 참조하는 테이블과 컬럼이 실제로 존재하는지 확인 (Hallucination 방지)
3. EXPLAIN: DB에서 실행 계획을 뽑아서 런타임 오류가 없는지 확인

검증에 실패하면 Self-Correction 서브그래프가 동작합니다. 오류 메시지와 원래 질문을 SQL Agent에 다시 넘겨서 수정된 SQL을 생성하고, 다시 검증합니다. 최대 3회까지 시도하고, 그래도 안 되면 UNFIXABLE로 판정합니다.

왜 3회인지 — 경험적으로 3회 안에 수정 안 되면 4, 5번째도 안 될 확률이 높습니다. 무한 루프를 방지하면서도 충분한 재시도 기회를 주는 타협점입니다.

Self-Correction을 LangGraph의 Subgraph로 분리한 이유는, 각 수정 시도를 Langfuse에서 별도 span으로 추적하기 위해서입니다. "3번째 시도에서 어떤 오류를 어떻게 수정했는지"를 나중에 그대로 확인할 수 있어서 디버깅에 꽤 도움이 됐습니다.

Deliver: 실행 + 결과 전달#

Executor#

검증을 통과한 SQL을 실행합니다. READ-ONLY 트랜잭션에서 30초 타임아웃으로 묶어서 잘못된 쿼리가 DB에 부하를 주지 못하게 막고, 최대 결과 행 수도 1000개로 제한합니다.

Result Agent#

실행 결과를 사용자에게 보기 좋게 정리합니다. 단순히 테이블 데이터를 보여주는 게 아니라, 결과에 대한 인사이트와 후속 질문을 함께 제안합니다.

질문: "부서별 주문 건수?"
결과: 개발팀 45명, 영업팀 32명, 경영지원팀 18명
인사이트: "개발팀이 전체 대상자의 38%를 차지합니다"
후속 질문 제안: "개발팀의 월별 매출 추이를 볼까요?"

복잡도별 분기#

모든 질문에 풀 파이프라인을 태우면 낭비입니다. Intent Agent가 판단한 복잡도에 따라 경로가 달라집니다:

SIMPLE 쿼리는 단일 테이블, 단순 필터/집계입니다. Column Pruning을 건너뛰면 LLM 호출이 1회 줄고, 응답 시간도 줄어듭니다.

기술 스택 정리#

숫자로 보는 시스템#

방어 계층은 실제로 눌리나#

여러 테넌트가 붙는 시스템이라, SQL을 잘 만드는 것과 별개로 앞단에서 막아야 할 게 있습니다. 한 IP가 요청을 쏟아붓거나, 한 테넌트가 배정된 예산을 넘기거나, 같은 질문이 반복해서 들어오는 경우입니다. 표에 적힌 숫자는 설계값이고, 실제로 그 방어가 눌리는지는 재봐야 압니다. 그래서 이 계층들을 운영에 올라가 있는 클래스 그대로(verbatim) 떼어다 격리 하네스에 올려놓고 동작을 확인했습니다.

IP 단위 rate limiting

미들웨어 단계에서 IP별 슬라이딩 윈도우를 돌립니다. 운영값인 60초당 60요청 기준으로 한 IP에서 100개를 연달아 던지면, 앞의 60개는 200으로 통과하고 나머지 40개는 429로 떨어집니다. 첫 429가 정확히 61번째 요청에서 나오는 것까지 확인했습니다. 윈도우가 지나면 회복되는지도 봤는데, 짧은 윈도우를 요청으로 가득 채워 막은 뒤 윈도우가 넘어가길 기다렸다가 다시 던지면 전부 통과합니다. 카운트가 시간에 따라 빠진다는 뜻입니다. /health 같은 헬스체크 경로는 카운팅에서 제외해서, 모니터링이 스스로를 막는 일은 없게 했습니다.

테넌트 쿼터와 예산

IP 단위 위에 테넌트 단위 제어가 한 겹 더 있습니다. 시간당 요청 한도, 일일 한도, 그리고 비용 예산입니다. 예산 쪽이 좀 재밌는데, 월 예산을 $0.10으로 걸어두고 누적 비용을$0.12까지 올려보면 $0.10을 넘어서는 시점에 요청이 차단됩니다. 요청 수가 아니라 실제 LLM 호출 비용을 기준으로 끊는다는 거고, 비활성 처리된 테넌트는 한도 이전에 먼저 걸러집니다.

테넌트 격리 쿼리 캐시

같은 질문이 다시 들어오면 파이프라인을 통째로 다시 태우지 않고 Redis 캐시에서 바로 돌려줍니다. 미스 → 셋 → 히트 순서로 재보면 히트가 0.29ms 수준으로 떨어집니다. 파이프라인 한 바퀴가 LLM 호출 여러 번을 포함한다는 걸 생각하면, 히트 한 건이 아끼는 시간과 비용이 작지 않습니다. 중요한 건 격리입니다. 테넌트 A가 캐시에 넣은 결과를 테넌트 B가 조회하면 미스가 나야 합니다. 캐시 키에 테넌트 식별자를 넣어(cache:{tenant}:{질문 해시 앞 16자}) 이걸 보장하고, 실제로 B가 A의 캐시를 못 보는 것과 무효화(invalidate) 시 해당 테넌트 엔트리 3건이 0건으로 지워지는 것까지 확인했습니다.

여기서 한 가지는 정직하게 적어둬야겠습니다. 위 수치들은 운영 클래스를 verbatim으로 떼어다 격리 하네스에서 잰 값이지, 프로덕션 트래픽을 모니터링해서 뽑은 값이 아닙니다. 코드 자체는 실제로 서비스에 올라가 있지만, 이 측정은 하네스에서 재현한 것입니다. 그리고 rate limiter는 지금 in-memory 카운터(defaultdict) 위에서 돕니다. 인스턴스가 여러 대면 대별로 카운트가 따로 놀아서 분산 환경에서는 정확하지 않고, 재시작하면 카운트가 리셋됩니다. 운영값 60req/60s도 소규모 기준입니다. 이 한계들은 코드의 속성이라 트래픽이 커지면 Redis 기반 분산 카운터로 옮겨야 하는 숙제로 남아 있습니다.

파이프라인을 어떻게 테스트했나#

에이전트가 7+1개이고, GraphState 필드가 44개이고, 외부 서비스가 6개입니다. 이걸 어떻게 테스트하냐는 질문을 자주 받습니다.

통합 테스트 — 10개 의존성을 한꺼번에 mock#

파이프라인 전체가 올바르게 동작하는지 확인하는 통합 테스트가 있습니다. LLM, RAG, Executor, Cache, Feedback, Tracer 등 10개 이상의 외부 의존성을 mock으로 교체하고, 파이프라인에 질문을 넣어서 PipelineResult가 정상적으로 나오는지 검증합니다.

async def test_run_returns_pipeline_result_on_success(self):
    graph_state = make_graph_state()  # 44개 필드 기본값
 
    pipeline = MultiAgentPipeline()
    mock_graph = MagicMock()
    mock_graph.run = AsyncMock(return_value=graph_state)
 
    result = await pipeline.run(
        question="주문 수를 알려줘",
        tenant_id="test_tenant",
    )
 
    assert isinstance(result, PipelineResult)
    assert result.success is True
    assert result.sql == "SELECT COUNT(*) FROM orders\nLIMIT 1000"

make_graph_state() 헬퍼가 44개 필드에 기본값을 채워주고, 테스트마다 필요한 필드만 override합니다. 이 패턴 덕분에 테스트 코드가 짧으면서도 모든 필드를 커버합니다.

의도별 경로 분기 검증#

Intent가 SQL_QUERY / GENERAL / CLARIFICATION_NEEDED 세 가지라서, 각 의도별로 PipelineResult의 success 조건이 다릅니다:

# SQL_QUERY: validated_sql이 있으면 성공
state = make_state(intent="SQL_QUERY", validated_sql="SELECT 1 LIMIT 1000")
assert result.success is True
 
# CLARIFICATION_NEEDED: SQL 없어도 성공 (명확화 질문이 있으니까)
state = make_state(intent="CLARIFICATION_NEEDED", validated_sql=None,
                   clarification_question="어떤 데이터를 원하시나요?")
assert result.success is True
 
# GENERAL: SQL 없어도 성공 (일반 대화)
state = make_state(intent="GENERAL", validated_sql=None)
assert result.success is True

의도별로 "성공"의 정의가 다르다는 걸 테스트로 명시합니다. 이게 없으면 CLARIFICATION_NEEDED에서 SQL이 없다고 실패로 처리하는 버그가 생기기 쉽습니다.

직렬화 — DB가 반환하는 타입들#

SQL 실행 결과를 JSON으로 직렬화할 때, DB가 반환하는 타입이 문제가 됩니다. datetime, Decimal, bytes 같은 타입은 json.dumps()가 처리 못 합니다.

def test_datetime_serialized_to_isoformat(self):
    dt = datetime.datetime(2024, 1, 15, 10, 30, 0)
    rows = [{"created_at": dt}]
    result = _make_serializable(rows)
    assert result[0]["created_at"] == "2024-01-15T10:30:00"
 
def test_decimal_serialized_to_float(self):
    rows = [{"amount": Decimal("99.99")}]
    result = _make_serializable(rows)
    assert result[0]["amount"] == pytest.approx(99.99)
 
def test_bytes_serialized_to_string(self):
    rows = [{"data": b"hello"}]
    result = _make_serializable(rows)
    assert result[0]["data"] == "hello"

프로덕션에서 MySQL의 DECIMAL 컬럼이 Python Decimal로 오는데, 프론트엔드에서 처리 못 해서 빈 화면이 뜨는 버그를 실제로 겪었습니다. 이후에 이 테스트를 추가했습니다.

SSE 스트리밍 이벤트 순서#

실시간 스트리밍은 이벤트 순서가 중요합니다. start가 먼저, 중간에 단계별 이벤트, 마지막에 complete.

async def test_stream_yields_sse_events(self):
    events = [e async for e in pipeline.run_stream(question="주문 수는?", ...)]
 
    assert events[0]["event"] == "start"
    assert events[-1]["event"] == "complete"
    event_names = [e["event"] for e in events]
    assert "intent" in event_names or "sql_generation" in event_names

complete 이벤트에는 전체 timing breakdown도 포함됩니다. 각 에이전트별 소요 시간을 Tracer에서 수집해서 클라이언트에 전달합니다.

async def test_stream_complete_event_includes_timing(self):
    complete_event = [e for e in events if e["event"] == "complete"][0]
    assert "timing" in complete_event["data"]
    assert complete_event["data"]["timing"]["total_ms"] == 120

초기화 실패 시 안전한 에러 반환#

DB 연결이 끊기거나 API 키가 없는 상황에서 파이프라인이 크래시하면 안 됩니다. 예외가 발생해도 PipelineResult(success=False)를 반환하고, 스트리밍 모드에서는 error 이벤트를 yield해야 합니다.

async def test_run_returns_failure_when_initialization_raises(self):
    with patch.object(pipeline, "_ensure_initialized",
                      side_effect=RuntimeError("TARGET_DB_URL이 설정되지 않았습니다")):
        result = await pipeline.run(question="주문 수는?", tenant_id="t1")
 
    assert result.success is False
    assert "초기화" in result.error
 
async def test_stream_yields_error_event_on_init_failure(self):
    with patch.object(pipeline, "_ensure_initialized",
                      side_effect=RuntimeError("DB 연결 실패")):
        events = [e async for e in pipeline.run_stream(question="테스트", tenant_id="t1")]
 
    assert len(events) == 1
    assert events[0]["event"] == "error"

이 테스트들이 있어야 "에러가 나도 시스템이 그냥 죽지 않고, 사용자한테는 의미 있는 응답을 준다"는 걸 코드로 보장합니다.

돌아보면 이 글에서 말한 구조의 핵심은 멋있는 멀티 에이전트 구성이 아닙니다. 질문을 작게 나누고, 각 단계를 따로 검증 가능하게 만들고, 실패했을 때 어디서 잘못됐는지 바로 보이게 만드는 쪽이 더 중요했습니다.

다음 글에서#

이 파이프라인에서 가장 까다로웠던 부분 — 수백 개 테이블에서 정확한 테이블을 찾아내는 RAG + Reranking 검색 아키텍처를 3편에서 자세히 다루겠습니다.