예를 들어 이런 질문이다.

• "저번에 올린 회의록 중 최신 5개 보여줘"
• "pdf 파일 중에서 계약서만 보여줘"
• "A파일 말고 다른 파일 뭐 있었지?"

처음에는 LLM에게 SQL을 생성하게 하면 되겠지. 하고 단순하게 생각했다.

하지만 실제 서비스에 붙이면서 생각보다 많은 문제가 터졌고 새로운 설계가 필요하게 되었다.

핵심은 하나이다.

SQL을 생성하게 하지 말고, 조회 스펙(Query Spec)을 생성하게 하자.

이번 글은 자연어 → SQL 조회 조건을 만드는 에이전트와 그 결과를 RDB / VectorDB에서 공통으로 해석하는 구조를 만들게 된 과정에 대한 기록이다.

자연어 데이터 조회

서비스에는 파일(assets)이 꽤 많이 쌓인다.

유저는 파일명을 정확히 기억하지 않는다. 그래서 보통 이렇게 묻는다.

• "회의록 파일 보여줘"
• "pdf 파일 뭐 있었지?"
• "어제 올린 파일"

맨 처음 이런 요구사항을 처리하기 위해서 두 가지 정도의 방법을 생각했다.

방법 1. 검색용 API를 여러 개 만든다

예를 들면 이런 식이다.

• /files?type=pdf
• /files?name=회의록
• /files?created_after=...

문제는 자연어 질문은 이런 식으로 정형화되어 있지 않다는 것이다.

"저번에 올린 회의록 중 최신 5개"

이 질문은 사실 아래 조건을 동시에 요구한다.

• 파일 이름 필터
• 정렬
• limit

하지만 API만으로 처리하면 예측할 수 없는 수 많은 조합이 생기게 되고 이 방법으로는 지속가능한 서비스를 제공할 수 없을 것이라는 생각이 들었다.

방법 2. LLM이 SQL을 생성한다

그래서 다음으로 LLM에게 SQL을 생성하게 했다.

SELECT *
FROM assets
WHERE asset_type = 'pdf'
ORDER BY created_at DESC
LIMIT 5

처음에는 꽤 잘 되는 것처럼 보였지만 QA단계에서 바로 문제가 생겼다.

SQL 생성 방식이 가진 문제

SQL을 그대로 생성하게 하면 생각보다 문제가 많다.

1. 존재하지 않는 컬럼 생성

LLM은 이런 걸 아무렇지 않게 만든다.

WHERE file_type = 'pdf'

하지만 실제 컬럼은

asset_type

이다.

아무리 테이블 정보를 준다 한들, 확률은 0이 아니라는게 문제다.

2. LIKE 남발

사용자가 파일명을 말하면 LLM은 높은 확률로 이런 sql문을 출력한다.

WHERE file_name LIKE '%회의록%'

하지만 실제로는 이미 파일 ID 목록이 주어져 있고 LIKE문 보다는 ID 접근이 훨씬 정확하다.

3. SQL 표현식 폭주

이 문제가 가장 큰 문제인데 바로 다음과 같은 것들이 튀어나온다는 것이다.

NOW()
INTERVAL
DATE_TRUNC
SUBQUERY

이런 것들은 DBMS에 과하게 종속적이기도 하고 검증하기도 어려울 뿐더러 통제하기도 힘들다.

4. VectorDB와 재사용이 불가능하다

현재 챗봇 서비스는 RDB + VectorDB를 같이 사용한다.

• 메타데이터 조회 → RDB
• 문서 검색 → VectorDB

문제는 SQL 문자열은 VectorDB에서 재사용할 수 없다는 것이다.

여기에 예상치 못한 프롬프트로 인한 SQL Injection 가능성까지 고려해보면 문자열로 생성된 sql문은 서비스에서 바로 사용하기에 부적합하다는 것이 결론이었다.

결국 통제 가능성과 재사용성을 고려한 sql 생성 에이전트를 만들어야하는 상황이 되었다.

LLM 역할 변경

여기서 생각을 바꿨다.

SQL을 생성하게 하지 말자.

대신 조회 스펙(Query Spec) 을 생성하게 하자.

즉 이런 구조다.

User Query
   ↓
LLM
   ↓
Query Spec(JSON)
   ↓
Interpreter
   ↓
SQLAlchemy / Qdrant Filter

LLM은 더 이상 SQL을 생성하지 않는다. 대신 조회 조건을 구조화된 JSON으로 반환한다.

Query Spec DSL 정의

먼저 조회 DSL을 정의했다. 기본적으로 sql에서 사용할 수 있는 대부분의 연산자와 데이터타입을 구성하는걸 목표로 했다.

Operator = Literal[
    "eq", "neq", "gt", "gte", "lt", "lte",
    "like", "ilike",
    "in", "not_in",
    "is_null", "is_not_null",
    "between"
]

Direction = Literal["asc", "desc"]

Scalar = Union[str, int, float, bool, datetime, date]
Value = Union[Scalar, List[Scalar]]

Where 조건은 트리 구조로 만든다. 타겟 컬럼과 연산, 비교값 및 구간을 설정할 수 있다. 우선순위는 value에 있고 만약 없을 경우 구간의 시작과 끝인 start와 end를 확인한다.

class WhereCondition(BaseModel):
    column: str
    operator: Operator
    value: Optional[Value] = None
    start: Optional[Scalar] = None
    end: Optional[Scalar] = None

논리 그룹도 지원한다. 이제 WhereCondition을 or나 and로 묶을 수 있게 되었다.

class WhereGroup(BaseModel):
    op: Literal["and", "or"] = "and"
    conditions: List[Union["WhereGroup", WhereCondition]] = Field(default_factory=list)

정렬은 단순하게 유지한다.

class OrderBy(BaseModel):
    column: str
    direction: Direction = "desc"

최종 조회 스펙은 다음과 같다.

class Statement(BaseModel):
    where: Optional[WhereGroup] = None
    order_by: List[OrderBy] = Field(default_factory=list)
    limit: int = Field(20, ge=1, le=200)
    offset: int = Field(0, ge=0)
    allowed_columns: Optional[Iterable[str]] = None

SQL 에이전트 입력과 출력

LLM이 받는 입력은 다음과 같다.

class SqlInput(BaseModel):
    user_id: str
    query: str
    table_info: str
    data_list: str

여기서 중요한 건 data_list다. 파일 목록을 같이 주기 때문에 LLM은 파일명을 보고 ID 필터로 변환할 수 있다.

출력은 단순하다.

class SqlOutput(BaseModel):
    where: Optional[WhereGroup] = None
    order_by: List[OrderBy] = Field(default_factory=list)
    limit: int = Field(20, ge=1, le=200)
    offset: int = Field(0, ge=0)

LLM은 이제 문자열로 된 SQL을 절대 생성하지 않는다. 오직 JSON만 반환한다.

Query Spec 번역기

이제 이 스펙을 실제 DB 쿼리로 변환해야 한다.

그래서 번역기 를 만들었다.

예외 타입 분리

class QuerySpecError(ValueError):
    pass

쿼리 생성 과정에서 발생하는 오류를 명확히 분리하기 위한 예외 타입이다.

예를 들어 다음 상황에서 사용된다.

• 허용되지 않은 컬럼
• 존재하지 않는 컬럼
• 빈 IN 리스트
• 너무 깊은 WHERE 트리
• 지원하지 않는 operator

컬럼 변환 함수

def _get_col(entity, col_name, allowed_cols=None):

이 함수는 문자열 컬럼 이름을 실제 SQLAlchemy 컬럼 객체로 변환한다.

핵심 역할은 두 가지다.

1. 허용 컬럼 검증

if allowed_cols and col_name not in allowed_cols:
    raise QuerySpecError

외부 입력이 임의의 컬럼을 건드리지 못하게 막는다.

예를 들어 내부 컬럼

• tenant_id
• internal_score
• is_deleted

같은 컬럼은 외부에서 직접 접근하게 하고 싶지 않을 수 있다.

이때 whitelist로 제어한다.

2. ORM / Core 지원

if hasattr(entity, col_name)
if hasattr(entity.c, col_name)

• ORM model → entity.column
• Core table → entity.c.column

둘 다 지원한다.

WHERE 조건 생성

가장 중요한 함수는 build_where_expr이다.

def build_where_expr(...)

이 함수는 다음을 수행한다.

• WhereCondition
• WhereGroup

을 받아서 SQLAlchemy boolean expression으로 변환한다.

트리 복잡도 방어

AI나 외부 시스템이 생성한 조건은 예상보다 복잡해질 수 있다.

그래서 다음 제한을 둔다.

max_nodes
max_depth

• OR 조건 수백 개
• 과도한 중첩

이런 상황을 방지한다. 이는 성능 보호와 악의적 입력 방어 역할을 한다.

재귀 구조

핵심 구현은 재귀 순회다.

def _walk(node, depth):

노드를 순회하면서

• WhereGroup → 논리 연산 조합
• WhereCondition → 실제 연산 해석

을 수행한다.

WhereGroup은 SQLAlchemy의

and_(...)
or_(...)

같은 논리 표현식으로 변환되고, 실제 중요한 로직은 WhereCondition을 해석하는 부분이다.

Operator 해석

위에서 언급했지만 간단하게 다시 정리하자면 WhereCondition의 구조는 다음과 같다.

WhereCondition
 - column
 - operator
 - value

해석 과정은 단순하다.

1. 컬럼을 가져온다

col = _get_col(entity, node.column)

2. operator를 SQLAlchemy 연산으로 변환한다.

이 부분이 사실상 Query Spec DSL을 해석하는 인터프리터 역할을 한다.

비교 연산

지원하는 비교 연산은 다음과 같다.

eq
neq
gt
gte
lt
lte

이에 대한 SQLAlchemy 매핑은 다음과 같다.

col == value
col != value
col > value
col >= value
col < value
col <= value

LLM이 만든 Query Spec은 결국 이 연산들로 변환된다.

문자열 검색

문자열 검색 연산과 그 매핑은 다음과 같다.

like
ilike


col.like(value)
col.ilike(value)

검색 패턴 %keyword% 생성은 해석기가 아니라 에이전트 단계에서 처리한다.

즉 Query Spec에는 이미 패턴 문자열이 들어 있어야 한다.

IN 연산

IN 연산은 다음과 같이 매핑된다.

in
not_in

col.in_(value)
not_(col.in_(value))

여기에 중요한 방어 로직이 들어가는데 다음과 같은 경우다.

id in []

sql 에이전트에서 위와 같은 경우가 나올 수 있는데 이 경우 SQL 의미가 없기 때문에 해석 단계에서 바로 예외 처리한다.

NULL 체크

SQL에서는 NULL 비교가 일반 비교와 다르기 때문에 별도 연산이 필요하다.

is_null
is_not_null

col.is_(None)
col.is_not(None)

BETWEEN

BETWEEN은 두 가지 입력 방식을 지원한다.

value=[start, end]

또는

start / end 필드

해석기는 두 방식을 모두 처리한다.

col.between(start, end)

이 설계는 LLM 출력이 약간 흔들리더라도 복구가 가능하도록 하기 위한 것이다.

Statement 조립

마지막 단계는 build_statement이다.

stmt = select(entity)

여기에 다음 요소를 붙인다.

WHERE

stmt.where(where_clause)

ORDER BY

stmt.order_by(col.asc())

여기서도 _get_col을 사용한다.

즉, 정렬 컬럼도 검증된다.

LIMIT / OFFSET

stmt.limit()
stmt.offset()

구조적 장점

이 설계의 가장 큰 특징은

SQL 문자열을 만들지 않는다

는 점이다.

대신 쿼리 스펙 트리를 생성한다.

설계를 하면서 가장 고려했던 부분은 아래와 같다. 다시 말해서 아래와 같은 장점을 가질 수 있다는 말이다.

• SQL injection 위험 감소
• 타입 안정성
• 복잡한 논리 처리 쉬움
• ORM 친화적

정리

아직 완전히 운영단계로 넘어간 설계는 아니다. 그래서 지금도 추가 보완이 이뤄지고 있는 영역이긴 하다.

하지만 기존 자연어 -> sql변환이 필요한 영역의 구현을 온전히 llm에 맡기는 구조에서 통제 가능한 구조로 변형해본 경험은 꽤 재밌었다. 부디 운영단에서 문제없이 돌아갔으면 하는 바람과 함께 글을 마무리해보겠다.

1. 왜 하필 윈도우 서버로 개발하셨죠?

2023년도 하반기에 받았던 질문이다. 잠깐 배경설명을 하자면, 대학원 석사 1차 때 개발용역으로 진행한 프로젝트를 소개할 때 받았던 질문으로 고객사에서는 윈도우 환경에서 실행 가능한 프로그램을 원했다.

고객사는 하드웨어 위주의 회사로 본인들이 판매하는 하드웨어와 호환되는 소프트웨어를 원했고 이러다보니 윈도우 환경을 원했던 것 같다. 솔직히 말하자면 당시에 별로 원하는 프로젝트도 아니었고 제대로 뭘 할 줄 아는 상태도 아니었기에 이것저것 고객사에게 설명하고 좋은 방향으로 이끌기보단 해달라는 것들이 뭔지도 제대로 모르는 채로 개발을 진행했었다.

프로젝트가 끝나고 보니 함께 진행했던 팀원들은 나름 이 프로젝트를 잘 포장해서 어느 정도 덕을 봤다고는 하는데 나는 아니었다. 변명아닌 변명을 해보자면 PM역할을 하면서 개발 자체보다는 일정관리나 의사소통에 좀 더 많은 시간을 할애했고 이렇다보니 이 프로젝트에서 "개발자"로서의 역할은 좀 약했다고 생각한다. 그리고 이런 문제는 면접에서 그대로 드러나버렸다.

윈도우 서버보단 다른 옵션도 있었을 것 같은데...

이것저것 이 프로젝트를 설명하다보니 개발 임원급 되시던 분이 왜 하필 윈도우 서버로 개발을 했냐는 질문을 했다. 그리고 나의 대답은

• 고객사가 원했기 때문에 윈도우 서버에서 개발을 진행할 수 밖에 없었습니다.

예전에 7년차 쯤 되는 개발자분께 이 대답에 대해 어떻게 생각하느냐 물어본적이 있다. 돌아온 답은 "팩트긴 하지만 경우에 따라선 생각없이 프로젝트를 진행한 것처럼 보일 수 있다" 이다. 나도 동의한다. 무엇보다 저 대답을 한 뒤에 돌아온 답은

1. 다른 옵션은 고민해본 적 없느냐
2. 더 선택지가 많았을 것 같은데

였으니까 일단 부정적인 인상이었던 것은 틀림 없어보인다. 아마 내가 해당 프로젝트를 잘 설명 못했던 것도 있겠지만 그 임원분의 입장에서는 기술 분석조차 제대로 안하고 무지성으로 개발하는 사람처럼 보이지 않았을까 싶다. 그렇다면 이제와서 저 질문을 받는다면 과연 어떻게 대답할 수 있을까.

생각해본다면 아마 이렇게 답할 것 같다.

"요구사항을 분석해본 결과 윈도우 서버보단 안정적인 리눅스 서버에서의 개발을 제안드려봤으나 결론적으로는 본인들이 현재 보유 중인 윈도우 소프트웨어와의 호환이 되는 것을 목표로 한다는 것을 알고 윈도우 서버를 구축하기로 했습니다."

실제로 1차적인 개발은 리눅스에서 진행한 뒤 2차 개발로 윈도우 서버로의 포팅작업이 있었고 이를 위해서 고객사와 여러번 회의도 진행했었다. 다만 이런 과정을 면접관들에게 제대로 어필하지 못한 채로 그저 그들이 원했으니까 그렇게 했다라는 식의 대답이 문제였다고 본다. 면접에 정답은 없으니 저런다고 면접관들의 마음에 들지는 확신하지 못하겠지만 적어도 개발 진행에 있어서 더 나은 결과물을 위해 고민한 흔적을 전달할 수 있는 대답이라고 생각한다.

2. CoT에 대해 어떻게 생각하시죠?

2025년 상반기에 받았던 질문이다. 이때 지원 직무는 AI 엔지니어고 주된 업무는 LLM 소프트웨어 개발이었다. 당연히 LLM 프롬프팅 기법 관련 질문과 최근의 트랜드에 대한 질문이 나올줄 예상하고 있었고 아니나 다를까 위의 질문이 들어왔다.

당시 나는 몇몇 LLM 관련 사이드 프로젝트들을 진행하며 프롬프팅 기법이나 MCP나 하는 것들에 대한 공부를 하긴 했으나 결국 내 개인적인 생각은 최신 기법이나 트랜디한 기술들이 중요한게 아니라 필요에 따라 자유롭게 적용할 수 있는, 필요 없다면 과감하게 버릴 수 있는 개발 능력이 더 중요하다는 것이었다. 이는 아직도 변함이 없다만 질문에 대한 나의 대답은 아주 오만하기 그지 없었다.

• 몇 번 적용시켜보고 성능을 비교해보았지만 큰 성능의 변화는 느끼지 못했습니다. 아무래도 LLM 관련 기술들이 비교적 최신의 기법들이라 연구가 많이 이뤄지지 않은 것 같습니다.

지금 다시 읽어보니 이렇게 느껴진다. "더 좋은거 나올 때까지 기다려야죠 뭐." 다시 말해서 스스로 성능을 끌어올리기보단 남들이 연구해놓은거 가져다 쓰겠다는 심보가 눈에 띈다는 소리다.

마침 현재 AI 엔지니어, LLM 서비스 개발 직무에서 일하는 입장에서 생각해보면 이 직무는 개발직으로 느껴지긴 하지만 연구 업무가 없는 것은 또 아니라고 본다. 어떻게하면 지금 있는 데이터에서 성능을 더 끌어올릴 수 있을까. RAG 과정을 좀 더 최적화할 수는 없을까. 데이터를 다른 형태로 가공해볼까. 대학원 시절 해왔던 제안서 작성이나 기초 모델 연구 같은 것들과는 조금 방향성이 다르긴 하지만 어쨌든 연구 비스무리한 업무를 자주 한다. 때문에 저런 대답은 확실히 이 직무에 대한 고민이 전혀 없는 것처럼 들린다. 물론 다른 직무라고 저런 대답을 좋아하진 않겠지만.

결국 이때 지원했던 직무에서 일하고 있기 때문에 과연 다시 저런 질문을 받으면 어떻게 대답할지 크게 생각하진 않았다만 아마 지금 가진 철학과 지속적인 연구의 필요성을 더 어필하지 않았을까 싶다.

3. 그러면 싱글톤이 되는거 맞아요?

이것도 2025년 상반기에 들었던 질문이다. 이 기업에서는 대면 면접 전에 과제 전형이 있었는데 이 때 약간 디자인 패턴에 딥하게 몰입해 있던 터라 아는 것 조금이라도 더 어필해보겠다고 싱글톤 패턴 어쩌고 하면서 구현 방식을 설명했었다. 사실 그게 중요한게 아니었단건 지금에서야 깨달았지만 어쨌든 그랬었다.

해당 직무도 AI, LLM 영역이라 사용 언어가 파이썬이었는데 과제 전형 당시 싱글톤을 구현하기 위해 두 가지 방식을 썼다.

첫 번째는 아래와 같이 직접 객체에 초기화 정보를 주는 방식이다.

class Singleton:
    _initailized = False
    _instanse = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        if not self.__class__._initialized:
            self.__class__._initialized = True

파이썬 문법과 관련된 방법이기 때문에 딱히 설명은 하지 않겠지만 대충 보면 인스턴스가 없다면 인스턴스를 만들고 초기화 플래그를 True로 바꿔서 더이상 인스턴스가 만들어지지 않게 하는 방법이다.

두 번째 방법은 간단하다.

# Singleton module
singleton = Singleton()

# Other module
from singleton_module import singleton
...

그냥 싱글톤으로 쓰고싶은 객체를 전역에서 선언해버리는 것이다. 이건 나도 해당 과제를 진행하기 전까지는 몰랐는데 이렇게 전역에서 인스턴스를 생성하면 파이썬 프로그램의 생명주기 내에 메모리에 계속해서 올라가 있게 되고 직접 메모리를 해제하지 않는 이상 같은 인스턴스가 유지된다. 사실 파이썬 메모리 관련해서 깊게 공부한적이 많지 않아서 이 설명이 맞는지는 모르겠다. 아마 GC와 파이썬 객체 생성 및 메모리 할당 방식을 좀 더 공부해야지 싶다.

아무튼간에 실제로 해시값을 읽어보면 어디서 해당 객체를 읽어오던 같은 값을 가리킨다. 과제 전형에서 나는 두 가지 방식을 혼용해서 썼고 면접에서는 두 번째 방식이 정말로 싱글톤이 되는게 맞냐는 질문을 받았다.

아마 아닐텐데..?

면접관은 두 번째 방식은 틀렸다는 의견을 내놓았고 나는 어떻게 보면 최악의 답변을 했다.

• 아 그렇다면 제가 잘못 알고 있었나봅니다. 다시 공부해보겠습니다.

인정하고 배우려는 자세는 좋다고 생각하지만 내가 면접관의 입장에서 이런 대답을 들었다면 조금 실망했을 것 같다. 면접관이 일부러 속이려고 했든, 진짜로 잘못 알고 있었든, 최소한 내가 왜 그렇게 알고 있었는지, 어디서 어떻게 그런 지식을 얻었는지, 좀 더 나가면 왜 싱글톤을 두 가지 방식으로 구현했는지 까지는 말했어야 한다고 생각한다. 적어도 개발자로서 그것이 비록 잘못 알고 있는 지식이더라도 명확하게 알고 구현하는 것과 "그런것 같다"는 식으로 구현하는 것은 다르다고 보기 때문이다. 그런 면에서 나의 대답은 후자의 뉘앙스를 풍겼을 것 같다.

이외에도 후회되는 답변들이 있긴 하지만 가장 기억에 남고 생각해볼만한 질문들은 여기까지인 것 같다. 혹시 나중에 더 생각난다면 그때 다시 적어보기로 하겠다. 공교롭게도 적어보고나니 질문의 카테고리가 다 다르다.

1번 질문은 프로젝트에 관한 질문, 2번 질문은 직무에 관한 질문, 3번 질문은 개발 지식에 관한 질문이다.

결국은 원하는 직무, 원하던 업무를 하고 있긴 하지만 이런 부분들을 더 보완했었더라면 선택지가 더 넓지 않았을까 하는 생각이 든다. 그래도 덕분에 요즘 만드는 서비스들은 더욱 이런 것들을 염두에 두고 만들고 있다. 서비스는 더 나은 방향으로 가고 있는지, 적용시켜볼만한 새로운 기법들은 없는지, 프로그램에 기술적으로 문제되는 부분은 없는지 파악하려 한다. 이런 것들이 쌓여서 다시는 저런 실수들을 하지 않게 될테니까.

이번 글의 핵심 주제는 다음과 같다.

“과연 AI 서비스는 전통적인 개발 아키텍처에 잘 녹아들 수 있는가?”

그리고 이 질문에 답하기까지의 고민 과정이다.

레이어드 패턴의 문제점

흔히들 말하는 레이어드 패턴의 문제점은 이렇다.

• Service 레이어가 비대해진다
• 계층을 무시하고 Controller에서 Repository를 직접 호출한다

하지만 이건 사실 문제의 증상일 뿐, 내가 겪은 진짜 문제는 레이어드 패턴이 전제로 삼는 세계관 자체가 AI 서비스와 맞지 않는다는 점이다.

레이어드 패턴이 가정하는 기본 구조는 간단하다.

Request → Service → DB → Response

• 비즈니스 로직은 단일 요청에 대해 단 한 번 실행되고
• 상태 변화는 대부분 트랜잭션 내부에서 일어나며
• 흐름은 직선적이다

즉, 한 번의 요청 → 한 번의 로직 실행에 특화된 구조다.

그런데 AI의 동작은 이렇지 않았다.

AI 워크플로우는 직선이 아니라 ‘그래프’다

내가 만들고 있는 AI 서비스의 요청 처리를 단계별로 나열해 보면 이렇다.

1. 쿼리 전처리
2. 의도 분리 및 도메인 감지
3. RAG 검색 (벡터 + RDB + 웹)
4. 검색 결과 정리 및 Reranking
5. 프롬프트 생성
6. LLM 호출 (여러 Provider 중 선택)
7. 후처리 및 응답 구성
8. 텔레메트리 기록 및 평가

문제는 위 과정이 단순한 단계의 나열이 아니라는 것이다.

특정 단계는 분기될 수도 있고, 병렬 실행되기도 하고, 조건에 따라 건너뛰기도 하고, 일부 단계는 같은 요청에서 반복되기도 한다.

실험 설정에 따라 흐름 자체가 바뀌기도 하고 무엇보다 더 나은 성능을 위해 지속적으로 구조가 바뀌게 된다.

즉, AI 워크플로우는 본질적으로 그래프(Graph)다.

이걸 억지로 SomeAiService.handle() 같은 하나의 서비스 메서드 안에 꾸겨 넣으면 결과는 뻔하다. Service 레이어는 순식간에 오케스트레이션 덩어리가 된다.

공통 단계는 이곳저곳에 중복되고, 실험을 하려면 하드코딩된 프롬프트와 파라미터를 뒤져서 직접 수정해야 하고, 전체 흐름은 보이지 않고 관리되지 않는다.

이건 레이어드 패턴이 잘못됐다는 게 아니다. 레이어드 패턴은 애초에 이런 종류의 복잡성을 수용하도록 설계된 구조가 아니기 때문이다.

AI 로직을 위한 구조

문제를 해결하기 위해 나는 “레이어”가 아니라 “워크플로우 단위”로 시스템을 바라보기 시작했다.

워크플로우가 가져야 하는 특성을 다음과 같이 정리했다.

1. 각 단계(Node/Step)는 독립된 기능 단위
2. 입력과 출력이 명확
3. 그래프 형태로 서로 연결
4. 분기 / 반복 / 병렬 실행을 자연스럽게 표현
5. 특정 단계만 교체하거나 실험하기 쉬움

그리고 각 단계는 다음 형태 중 하나에 속한다.

• Agent: LLM 호출 및 지능적 판단을 수행하는 노드
• Tool: 검색, DB 접근, 외부 API 호출 등 실제 행동을 수행하는 노드
• Graph: 여러 Agent/Tool을 조합해 전체 흐름을 구성하는 상위 구조

이 지점에서 자연스럽게 헥사고날 아키텍처 이야기가 이어진다.

헥사고날 아키텍처 도입기

사실 레이어드 패턴에 문제점을 인지한 순간 바로 생각난 것이 바로 헥사고날 아키텍처였다. 잠깐 스치듯 공부한 것이라 확신은 없지만 내가 정리한 헥사고날 아키텍처의 철학은 다음과 같다.

• 도메인을 중심에 둔다
• 외부 의존성은 Adapter로 밀어낸다
• 유스케이스는 흐름만 정의하고 기술적인 구현은 Port/Adapter로 분리한다

이 철학 자체는 정말 좋다고 생각한다. 하지만 대부분의 헥사고날 사례는 이런 세계관을 가정한다.

“요청 하나 → 유스케이스 하나 → 비즈니스 규칙 하나”

즉, 전통적인 웹/백엔드 흐름을 기반으로 한다. 그래서 헥사고날을 그대로 AI에 적용하려 하면 레이어드 패턴과 마찬가지로 묘하게 어긋나는 순간들이 생긴다.

• LLM 호출은 도메인 규칙인가? 인프라인가?
• 프롬프트는 데이터인가? 설정인가? 코드인가?
• RAG 검색은 Repository인가? Tool인가?
• Agent는 유스케이스인가? 도메인 객체인가?

경계가 모호해지고 유지보수성은 오히려 나빠진다.

그래서 내가 내린 결론 헥사고날을 그대로 적용하는 게 아니라 재해석해서 AI 로직에 부분적으로 적용하는 것이다.

구체적으로는 다음과 같다.

• 도메인의 규칙/정책은 그래프의 분기 조건으로 녹아든다
• LLM, 검색, DB, 이미지 서버 등 모든 외부 기능은 Adapter로 구성
• Agent와 Tool은 Adapter를 활용해 의사결정, 혹은 필요한 동작을 수행하는 독립 모듈
• Graph는 여러 Agent와 Tool을 조합한 상위 유스케이스

정리하면,

“도메인 서비스 → Application Service” 구조를 “도메인 규칙 → Workflow / Graph”로 치환한 형태

이렇게 보니 AI가 요구하는 복잡성과 변화 속도를 헥사고날이 자연스럽게 품을 수 있게 된다.

헥사고날의 함정

사실 나는 처음에 서비스 전체를 헥사고날로 싹 갈아엎어야 한다고 생각했다. 그러니까, 위의 부분적 도입이 우선이 아니라 무지성으로 '헥사고날 좋아보이니까 갈아엎어!'를 시전한 것이다. 레이어드 패턴에 한계가 보이기 시작했고 “외부 의존성을 Adapter로 몰아넣는 구조”가 이론적으로 너무 매력적이었기 때문이다.

하지만 실제로 적용한 결과는…

• 러닝커브가 꽤 가파르고
• 모든 외부 의존성에 Port를 만들고
• Adapter가 난무하고
• 코드가 레이어들로 과도하게 쪼개지고
• 헥사고날 유지가 개발보다 더 중요한 작업이 되어버렸다.

특히 AI는 외부 의존성이 너무 많다.

1. OpenAI
2. Perplexity
3. Gemini
4. 문서 변환 서버
5. 이미지 생성
6. 벡터 DB
7. 검색 툴
8. 기타 등등 ...

경우에 따라 LLM 서비스 제공자를 바꿔야하고 벡터 DB 입출력도 고려해야하고 여기저기 흩어진 다 다른 형태의 문서들을 하나하나 찾아서 모아줘야한다. 이 모든 걸 Port-Adapter 체계에 넣는 건 아키텍처 설계가 아니라 노동이었다.

그래서 최종적으로 선택한 절충안으로 나는 이렇게 결론 내렸다.

• 전체 시스템은 레이어드로 둔다.
• 하지만 AI 워크플로우만 별도의 헥사고날+그래프 구조로 분리한다.

즉, FastAPI → Controller → Service → Repository로 이어지는 이 전통적 요청/응답 구조는 그대로 유지하되 이 뒤에서 동작하는 AI 처리 로직만 별도 아키텍처로 분리하는 것이다.

이 구조가 결국은 팀 전체가 이미 익숙한 구조를 깨지 않으면서 AI가 가진 복잡성을 독립적으로 다룰 수 있는 방식이라고 생각했다. 다시 돌아와서 결론적으로는 AI 로직에만 부분적으로 헥사고날의 철학을 도입하는 구조가 되었다는 것이다.

workflow 구조

분리된 AI 영역은 크게 두 파트로 나뉜다.

1) interface: 외부 의존성의 추상화 계층

이 영역은 헥사고날의 Adapter/Port 개념을 최소한만 차용한 형태다.

여기서 정의하는 것은 다음과 같다.

Protocol - 반드시 이뤄져야하는 공통된 동작에 대한 기능 명세 Provider - Protocol의 기능을 수행하는 실제 구현체 Factory - Provider를 모아 도메인 로직으로 전달하는 객체

Protocol은 위에서도 언급했지만 기능 명세의 역할을 한다. “LLM은 이런 메서드를 가져야 한다” “RAG는 이런 입력을 받으면 이런 결과를 반환해야 한다” 등의 동작을 적어만 두는 공간이다.

Provider는 외부 의존성에 따라 Protocol의 동작을 구현해 놓는다. 실제 OpenAI 호출 실제 Gemini 호출 실제 검색 API 호출 등등... 각기 다른 외부 의존성의 기능을 각각 구현해 두지만 디테일은 모두 감춘다.

Factory는 도메인 로직에서 사용하고자 하는, 혹은 각 동작의 상황에 따라 적절한 Provider를 선택해 반환한다.

이 덕분에 AI 로직(work)에서는 LLMClient.generate()라는 추상적 개념만 바라보면 된다.

2) work: AI 로직 구현 계층

여기는 AI가 ‘일하는’ 영역으로 서비스의 필요에 따라 다음과 같이 세 구성요소로 나누었다.

Agent - LLM Protocol을 받아 필요한 동작을 수행 Tool - 벡터 DB, 웹 검색 등의 보조적인 동작을 수행 Graph - Agent와 Tool을 하나로 묶어 구현된 일련의 동작을 수행

여기는 다음 원칙을 따른다.

• 외부 의존성을 직접 호출하지 않는다 → interface에서 주입받음
• Step 단위로 쪼개 재사용성 확보
• Graph에서 분기/반복/병렬을 자연스럽게 표현

이제 Service 레이어는 “그래프 하나 실행해줘”만 호출하면 된다. 즉, AI 로직은 완전히 독립된 작은 애플리케이션이 되어버린다.

구조 변경 이후의 장점과 느낀 점

구조를 이렇게 분리해보니, 기존 레이어드 패턴이 가진 단순함은 그대로 유지하면서도 AI 영역에서만 필요한 복잡성과 유연성을 자연스럽게 흡수할 수 있다는 점이 가장 큰 장점이었다. 요청/응답, 인증, 트랜잭션, DB 작업 같은 전통적인 백엔드 흐름은 기존 구조를 그대로 사용하므로 팀 전체가 익숙하게 유지보수할 수 있었고, 굳이 거대한 아키텍처 전환을 강요할 필요가 없었다.

반면 AI 영역은 헥사고날 아키텍처의 장점과 그래프 기반 워크플로우의 장점을 결합해 훨씬 더 유연해졌다. 외부 의존성은 interface 계층에서 캡슐화되어 Provider 교체나 추가가 정말 쉬워졌고, 실제 로직에서는 추상화된 LLMClient·RAGClient만 바라보면 되니 테스트하기도 수월했다. 목업을 붙여 빠르게 실험하는 것도 가능해졌다.

가장 체감이 컸던 부분은 복잡한 AI 워크플로우를 그래프 형태로 표현할 수 있게 되면서 분기, 반복, 병렬 실행 같은 흐름을 코드 구조 자체에서 자연스럽게 담아낼 수 있었다는 점이다. 이전처럼 Service 레이어가 거대한 오케스트레이션 덩어리가 되어버릴 일도 없었다. 이제 Service는 "그래프를 실행해줘"라는 한 줄로 모든 AI 처리를 넘길 수 있고, 복잡한 내부 동작은 모두 workflow 안에서 관리된다.

결국 느낀 점은 이것이다. AI 시스템은 본질적으로 전통적 레이어드 패턴 안에 잘 들어가지 않는다. AI 워크플로우는 직선이 아니라 그래프이고, 도메인 규칙과 외부 도구들이 유기적으로 엮여 있으며, 실험과 변화가 너무 빈번하다. 이런 특성은 기존의 “서비스 메서드 하나에 로직을 몰아넣는 구조”로는 결코 만족스럽게 표현할 수 없다.

그래서 나는 전체 시스템은 레이어드를 유지하되, AI 워크플로우만 독립된 그래프·에이전트·툴 기반 구조로 분리하는 방식을 선택했다. 이 방식은 헥사고날 아키텍처의 철학과도 잘 맞고, 기존 웹 백엔드의 단순함과 AI의 복잡성을 공존시키는 가장 현실적인 해답처럼 느껴진다. 지금으로서는 이 구조가 확장성·유연성·가독성 면에서 가장 균형 잡힌 선택이었다고 생각한다.

그래서 이제 코드리뷰도 AI의 도움을 받자는 생각이 들었다. 조금 다른 얘기지만 코드 작성도, 리팩토링도, 리뷰도 이젠 AI의 도움을 받는다니 이제 정말 단순 코더들은 다 죽어나가겠구나 하는 생각이 든다. 다시 돌아와서, 코드리뷰를 위한 AI는 생각보다 꽤 많다. CodeRabbit, Github Copilot, 기타 등등 몇 가지 비교해보면서 내린 결론은 Gemini Code Assist이다. 가장 큰 이유는 비용 부담 없이 가장 빠르게 실무에 도입할 수 있다는 점이었다. 물론 Gemini 자체에 대한 신뢰도 있었다. 하여튼 그래서 오늘의 글은 Gemini Code Assist를 도입하면서 생긴 크고 작은 변화들에 대한 기록이다.

Gemini Code Assist는 좀 기니까 GCA라고 부르겠다. GCA는 깃허브 마켓플레이스에서 무료로 설치할 수 있다.

Gemini Code Assist

설치 후 적용할 레포지토리를 연결해주면 간단하게 설정이 끝난다. 이제 해당 레포지토리에 PR을 올리면 간략한 코드 변경 요약과 리뷰가 출력된다.

변경 요약은 체감상 거의 즉시 ~ 1분 내로 올라온다. 실제 리뷰는 약 5분 정도 지나야 올라오는 느낌이다.

리뷰에는 Critical, High Priority, Medium Priority, Low Priority로 중요도 레벨이 같이 달린다. 기본적으로 Medium Priority까지만 출력이 되도록 설정이 되어있다. 실무에서도 너무 과한 리뷰를 방지하기 위해 기본 설정을 그대로 가져갔다.

실제로 GCA를 도입하고 나서 느낀 점들을 좀 정리해보면, 생각보다 얻어가는 부분이 꽤 많았다.

첫 번째로, 팀 전체의 스타일가이드를 훨씬 명확하게 정리할 수 있게 됐다. 사람마다 기준이 조금씩 다른 부분을 GCA에 명시적으로 정의해두면, 그게 사실상 팀의 공식 컨벤션처럼 작동한다. 기존에는 구두로만 전달되거나 PR 리뷰 중간에 슬쩍 이야기되는 것들이 많았는데, 이제는 명확한 문서와 자동화된 리뷰가 동시에 역할을 하니까 기준이 흐트러질 일이 거의 없다.

두 번째로, 자잘한 컨벤션 리뷰에 시간을 낭비하지 않아도 된다는 점이 상당히 크다. 띄어쓰기, 변수명 일관성, import 정렬, 스타일 포맷팅 같은 것들은 솔직히 사람이 매번 리뷰할 필요가 없다. ‘이런 것까지 내가 직접 리뷰해야 하나?’ 싶은 것도 많았는데, 이제는 그런 부담을 거의 GCA가 가져간다. 기본적인 스타일 문제는 머신이 걸러주고 리뷰어는 진짜 논의해야 하는 로직, 아키텍처, 설계 쪽에 집중할 수 있다.

세 번째로, 안티패턴을 찾는 데도 확실히 도움이 된다. 특정 프레임워크나 언어에서 흔히 발생하는 비효율적인 패턴, 잘못된 추상화, 리소스 누수 가능성 같은 것들을 꽤 잘 잡아준다. 사람이 놓치기 쉬운 부분을 자동으로 탐지해주니 코드 품질 관리가 한결 수월해진 느낌이다.

그리고 마지막으로, 이건 아직 확신할 정도는 아니지만, 장기적으로 팀 전체의 컨벤션을 강제하고 신규 인력이 합류해도 온보딩이 빨라질 것 같다. 팀마다 고유한 규칙이나 철학이 있는데, 그걸 새 팀원이 학습하는 데 꽤 시간이 걸린다. 그런데 GCA가 그 기준을 계속 자동으로 피드백해주면, 자연스럽게 팀의 코드 스타일을 따라가게 된다. 문서 + 자동화 리뷰 조합이 온보딩 시간을 줄여줄 것 같다는 느낌이 있다.

장점들을 쭉 이야기했는데, 반대로 쓰다 보니 몇 가지 아쉬운 점도 분명 있었다. 이 부분도 같이 적어두는 게 균형이 맞을 것 같아 정리해보면 아래 정도다.

첫째, PR을 닫았다가 다시 열지 않는 이상 자동 리뷰는 최초 한 번만 진행된다. 물론 수동으로 “리뷰 다시 해줘”라고 요청할 수는 있지만, 여전히 흐름이 끊기고 번거롭다. 코드가 계속 업데이트되는 상황에서는 자동으로 다시 리뷰가 붙어주면 좋겠다는 생각이 자꾸 든다.

둘째, 재리뷰를 받을 때는 생성형 AI의 한계가 그대로 드러난다. 어떻게든 뭔가 새로운 피드백을 만들어내려는 특유의 억지 리뷰가 종종 보인다. 이미 해결된 부분인데 다른 말로 또 언급한다든가, 실제로는 별 의미 없는 포인트를 중요해 보이게 말하는 식의 현상이 조금 있다. 이런 부분은 사람이 읽을 때 피곤해진다.

셋째, 기본 설정에서는 과거 리뷰 맥락을 모르고 동작한다. 과거 리뷰나 팀의 누적된 컨벤션을 학습시키는 옵션이 있기는 하지만, 설정도 까다롭고 리뷰 시간이 길어질 가능성이 높다. 여기에 비용 문제까지 고려하면 굳이 그렇게까지 해야 하나 싶은 마음이 든다. 말 그대로 배보다 배꼽이 커질 것 같은 느낌이다.

물론 단점보단 장점이 많고 실제 경험을 따져보면 실무에 들여올만한 가치는 충분이 있다고 본다. 아직은 시범적인 도입단계라 단순 리뷰에만 사용하고 있지만 좀 더 이런식의 코드리뷰 문화가 조직에 정착되면 개인적으로 해보고 싶은 것들이 더 있다.

뭐, 그건 좀 나중의 이야기기 때문에 일단 오늘 기록은 여기까지.

1. 필요한 api 라우터를 만들고
2. 대응하는 Service 로직을 구성한 뒤
3. 필요한 데이터를 Repository에서 전달하는 함수를 구현

이렇게 진행하는 이유는 쓸데없는 코드 구성을 생략하고 정말 필요한 로직만 골라서 구현할 수 있기 때문이다. 다만 실제 흐름은 Repository -> Service -> Controller 순으로 진행되기 때문에 기록은 이 순서를 따라가는게 좋겠다는 생각이다.

session maker 구성

당연하게도 SQLAlchemy를 사용하려면 Session maker부터 구성해야한다. 별건 아니지만 전역 변수 선언과 의존성 주입 구성의 일부분을 담당하기 때문에 살짝 언급하고 싶다.

# database/session.py
import os
from functools import lru_cache
from sqlalchemy.ext.asyncio import (
    AsyncEngine, create_async_engine, async_sessionmaker, AsyncSession
)

@lru_cache(maxsize=1)
def _async_database_url() -> str:
    return (
        f"mysql+asyncmy://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}"
        f"@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT','3306')}/{os.getenv('DB_DATABASE')}?charset=utf8mb4"
    )

def _make_async_engine() -> AsyncEngine:
    return create_async_engine(
        _async_database_url(),
        pool_size=int(os.getenv("DB_POOL_SIZE", 10)),
        max_overflow=int(os.getenv("DB_MAX_OVERFLOW", 20)),
        pool_timeout=int(os.getenv("DB_POOL_TIMEOUT", 30)),
        pool_recycle=int(os.getenv("DB_POOL_RECYCLE", 3600)),
        pool_pre_ping=True,
        pool_use_lifo=True,
        isolation_level=os.getenv("DB_ISOLATION", "READ COMMITTED"),
    )

database 폴더는 세션구성을 위한 설정을 담당한다.

# infra/global_state.py
from functools import lru_cache

@lru_cache(maxsize=1)
def get_async_engine() -> AsyncEngine:
    return _make_async_engine()

@lru_cache(maxsize=1)
def get_async_sessionmaker() -> async_sessionmaker[AsyncSession]:
    return async_sessionmaker(
        bind=get_async_engine(),
        expire_on_commit=False,
        autoflush=False,
        autocommit=False,
        class_=AsyncSession,
    )

설정을 받아서 생성한 session maker는 인프라 레벨에서 딱 한번만 준비되고 공통된 session을 UoW가 받아서 트랜잭션을 관리하는 형태이다. 즉, 세션 생성/해제 책임은 UoW, 연결 수명과 풀 설정은 Session maker가 담당한다.

Repository 구성

세션이 구성됐다면 다음은 실제 DB와 통신을 하는 Repository를 구성할 차례이다. Repository는 Orm 객체를 통해 데이터에 직접 접근한다. Orm 객체는 외부로 노출되면 예상치 못한 데이터 수정이 일어날 수 있기때문에 Service 레이어로 데이터를 전달할 때는 Dto를 통해 전달한다. 따라서 Repository를 구성하기 위해서는 Orm, Dto, Session이 필요하다.

Orm

# database/model/user_model.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'

    id = Column(
        Integer,
        primary_key=True,
        autoincrement=True,
        nullable=False,
        comment="자동 증가 기본 키"
    )
    name = Column(
        String(36),
        nullable=True,
        comment="사용자 이름"
    )

Orm 객체는 DeclarativeBase를 상속받아 구성된다. 예전에 Declarative mapping과 Imperative mapping에 대해 공부한 적이 있는데 생각이 잘 안나서 나중에 따로 또 공부를 해야겠다.

Dto

# schemas/dto/user.py
from pydantic import BaseModel
from typing import Optional

class UserDto(BaseModel):
    id: int
    name: Optional[str]

    model_config = {
        "from_attributes": True
    }

Dto는 Orm 객체에서 매핑할 컬럼을 대상으로 구성한다. 기본적으로 전체 컬럼을 전부 대응시켜고 좋지만 회사 프로젝트의 경우 데이터가 많아질 경우를 대비해 필요한 정보만 매핑할 수 있는 Dto를 여러개 구성해놨다.

from_attributes 는 Pydantic에서 제공하는 매핑 기능을 사용하기 위해 항상 True로 둔다. Dto가 많아질 수록 매번 model_config에 from_attributes 를 설정해주기는 힘들다. 때문에 BaseDto를 만들어 모든 Dto는 BaseDto를 상속하게 한다.

# schemas/dto/base_dto.py
from pydantic import BaseModel

class BaseDto(BaseModel):
    model_config = {
        "from_attributes": True
    }

Repository

마지막 실제 레포지토리를 구성한다.

from typing import Optional

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from database.model.user_model import User
from schemas.dto.user import UserDto

class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_user_by_id(self, user_id: str) -> Optional[UserDto]:
        result = await self.db.execute(
            select(User)
            .where(User.id == user_id)
        )

        user_orm = result.scalars().one_or_none()

        if not user_orm:
            return None

        return UserDto.model_validate(user_orm)

레포지토리는 초기화 시, 세션을 받으며 해당 세션을 통해 쿼리문을 날린다. 단, 세션은 오직 레포지토리에만 주입되며 트랜잭션은 모두 UoW가 관리하게 된다.

UoW(Unit of Work) 구성

트랜잭션 단위

Spring에서는 @Transactional 어노테이션으로 Service 레이어에서 트랜잭션을 관리할 수 있게 해줬다. 파이썬은 그런거 없다. 대신 우리는 디자인 패턴의 힘을 빌려 비슷하게 구성할 수 있다.

# uow.py
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from crud.repositories.user_repository import UserRepository

class UnitOfWork:
    def __init__(self, sf: async_sessionmaker[AsyncSession]):
        self.sf = sf
        self.session: AsyncSession | None = None

    async def __aenter__(self):
        self.session = self.sf()
        return self

    async def __aexit__(self, exc_type, *_):
        try:
            if exc_type is None:
                await self.session.commit()
            else:
                await self.session.rollback()
        finally:
            await self.session.close()

    @property
    def user(self) -> UserRepository:
        return UserRepository(self.session)

context manager를 통해 자연스럽게 with 구문 내에서 진입과 탈출을 구성할 수 있다. 모든 Commit은 UoW가 닫힐 때 자연스럽게 이뤄지고 중간에 예외가 발생한다면 바로 롤백이 이뤄질수 있다. 모든 레포들은 객체 속성으로 선언된다. 즉, UoW에 등록하고 싶은 객체들은 @propery로 선언하면 된다.

Repository Registry

UoW를 운영하기에 문제가 하나 있다. 레포가 많아질수록 UoW에 등록되는 레포가 자연스럽게 많아진다는 것이고 그럼 레포를 구성할때마다 항상 UoW를 구성하는 파일이 변경되어야 한다는 것이다. 이런 패턴은 협업에 있어서 항상 문제를 일으킨다. 예컨데 동시에 같은 파일에 접근해서 충돌을 만든다거나 하는 그런 것들. 소규모에서는 그러려니하고 마는데 서비스 덩치가 커지면 커질수록 이는 엄청난 개발 오버헤드를 일으킨다. 그래서 UoW를 위한 레지스터를 만들 수 밖에 없었다.

# repo_registry.py
from typing import Dict,Type, TypeVar

T = TypeVar('T')

class RepoRegistry:
    _registry: Dict[Type[T], Type[T]] = {}

    @classmethod
    def register(cls, repo_cls: Type[T]):
        if repo_cls in cls._registry:
            raise ValueError(f"Repository {repo_cls} already registered")
        cls._registry[repo_cls] = repo_cls

    @classmethod
    def get(cls, repo_cls: Type[T]):
        try:
            return cls._registry[repo_cls]
        except KeyError:
            raise ValueError(f"Repository {repo_cls} not found (available: {list(cls._registry.keys())})")

def register_repo():
    def decorator(repo_cls: Type[T]):
        RepoRegistry.register(repo_cls)
        return repo_cls
    return decorator

RepoRegistry는 캐싱을 위한 레포지토리 딕셔너리를 속성으로 갖는다. 클래스 매서드로 register와 get을 구성해 인스턴스 없이 전역으로 레지스트리에 접근 가능하게 구성한다.

마지막으로는 register_repo() 데코레이터를 구성해 레포지토리 클래스에서 간단하게 레지스트리에 등록할 수 있도록 구성한다.

이제 아래와 같이 UoW에 간단하게 등록가능하다.

@register_repo()
class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

레지스트리에 맞춰 UoW도 조금 코드를 바꿔준다.

T = TypeVar('T')

class UnitOfWork:
    def __init__(self, sf: async_sessionmaker[AsyncSession]):
        self.sf = sf
        self.session: AsyncSession | None = None
        self._cache = {}
        self._active = False

    async def __aenter__(self):
        self.session = self.sf()
        self._active = True
        return self

    async def __aexit__(self, exc_type, *_):
        try:
            if exc_type is None:
                await self.session.commit()
            else:
                await self.session.rollback()
        finally:
            await self.session.close()
            self._active = False
            self.session = None
            self._cache.clear()

    def _ensure_active(self):
        if not self._active or self.session is None:
            raise RuntimeError("UnitOfWork is not active. Use 'async with uow_factory() as uow:'")

        return self.session

    def get_repo(self, repo_cls: Type[T]) -> T:
        if repo_cls not in self._cache:
            session = self._ensure_active()
            repo_cls = RepoRegistry.get(repo_cls)
            self._cache[repo_cls] = repo_cls(session)
        return self._cache[repo_cls]

달라진 점은 레포지토리 캐시와 활성화 여부를 체크해주는 private 속성이 추가되었다는 것이다.

캐시를 통해 컨텍스트 내에서 한번 초기화된 레포지토리는 다음에 좀 더 빠르게 접근이 가능해지고 캐시에 등록되지 않은 레포지토리는 레지스트리에서 가져와서 캐시에 등록한다.

UoW의 활성화 여부에 대한 체크는 개발단계에서의 실수를 줄이기 위한 설계로 async context manager를 통한 접근을 강제한다.

마지막으로 async with문이 닫히면 UoW를 비활성화하고 캐시를 비움으로써 메모리의 부담을 줄인다.

실제 UoW는 아래와 같이 사용할 수 있다.

async with UnitOfWork(get_async_sessionmaker()) as uow:
    user_repo = uow.get_repo(UserRepository)

    user_dto = user_repo.get_user_by_id(user_id)

FastAPI DI 구성

구성된 UoW를 실제 라우터에 전달하기 위한 두 가지의 방법이 있다.

1. 활성화된 UoW 객체를 라우터에 직접 주입
2. UoW 객체를 생성할 수 있는 Factory를 주입

UoW 객체 주입

deps.py에 UoW 객체를 활성화 시킨 상태로 구성해두고 라우터에서 Depens()를 통해 주입받는 형태이다.

# api/deps.py
async def get_uow():
    async with UnitOfWork(get_async_sessionmaker()) as uow:
        yield uow

yield를 통해 Generator를 구성한다. Depends()는 Generator를 구성하는 Callable 객체를 전달 받을 시 라우터의 동작이 끝나면 Generator를 구성했던 자원들을 정리한다. 즉, 라우터가 종료될 때 자연스럽게 UoW의 커밋이 이뤄진다는 것이다. 때문에 라우터 단위의 트랜잭션을 구성할 수 있다.

@router("/user/{user_id}")
async def get_user(
    request: Request,
    user_id: int = Path(),
    user_service: UserService = Depends(get_user_service),
    uow: UnitOfWork = Depends(get_uow)
):
    ...

이 방식의 경우 단순히 UoW를 주입하는 것만으로 트랜잭션 단위를 묶을 수 있다. 단, 라우터가 무사히 끝나야만 커밋이 이뤄진다. 덕분에 만약 외부 Api와 통신을 하는 동작이 라우터 내부에서 이뤄진다면 예기치 못한 데이터 불일치가 발생할 수 있다.

예를 들면, "S3에 데이터 저장 -> DB에 저장 위치 및 메타데이터 기록 -> 나머지 라우터 동작" 이라는 로직이 있다고 가정해보자. 각각 a, b, c 동작이라 할때 커밋은 c가 끝나고 난 뒤에 이뤄진다. 따라서 a (성공) -> b (성공) -> c (에러) 와 같은 상황이 발생하면 c단계에서 UoW는 롤백이 이뤄진다. 덕분에 S3에는 데이터가 성공적으로 업로드 되었지만 b단계에서 DB에 저장된 정보는 c의 에러로 인해 롤백이 되어 데이터 불일치가 일어날 수 있다.

이를 방지하기 위해서는 외부 I/O 동작에 대한 추가적 예외처리, 혹은 명시적 커밋 동작이 필요하다.

UoW Factory 주입

첫 번째 방법과 다르게 UoW를 구성하는 Factory를 구성해서 이를 라우터 혹은 서비스에 전달하는 방식이다.

# api/deps.py
def get_uow_factory() -> Callable[[], UnitOfWork]:
    def factory() -> UnitOfWork:
        return UnitOfWork(get_async_sessionmaker())
    return factory

마찬가지로 라우터에서 Factory를 전달 받는다.

@router("/user/{user_id}")
async def get_user(
    request: Request,
    user_id: int = Path(),
    user_service: UserService = Depends(get_user_service),
    uow_factory: Callable[[], UnitOfWork] = Depends(get_uow_factory)
):
    ...

다만 Factory의 경우 자체적으로 async with를 통해 트랜잭션을 구성해야한다.

async uow_factory() as uow:
    ...

이 덕분에 트랜잭션 단위를 좀 더 세부적으로 조절할 수 있게된다.

더욱이 이렇게되면 굳이 라우터에서 UoW를 주입받을 이유가 없다. 라우터는 애초에 Orm에 접근해서는 안되니 실수를 방지하기 위해서라도 서비스에 직접 Factory를 주입하도록 구성한다.

# api/deps.py
def get_uow_factory() -> Callable[[], UnitOfWork]:
    def factory() -> UnitOfWork:
        return UnitOfWork(get_async_sessionmaker())
    return factory

def get_user_service() -> UserService:
    return UserService(get_uow_factory())

이제 라우터는 단순히 서비스만 주입받아 동작할 수 있다. 물론 서비스 레이어에서 UoW를 사용할 때는 반드시 async with문을 구성해야한다. 좀 귀찮을 수 있지만 회사 프로젝트에서는 첫 번째 방식의 단점이 더 크다고 판단해 거의 모든 서비스에서 Factory를 명시적으로 주입받도록 했다.

물론 외부 I/O 없이 단순한 동작만 있는 경우에는 UoW를 직접 주입받아도 상관없다. 결국 중요한 건 어떤 레이어가 어디까지 책임을 가져가는지, 트랜잭션 경계를 어떻게 잡는지, 협업과 유지보수에 어떤 영향을 주는지를 이해하고, 그 장단점을 알고 서비스 형태에 따라 유연하게 선택하는 것이라고 생각한다.

마무리

이번 글에서는 세션 구성, Repository, UoW, DI 방식까지 한 번에 훑으면서 데이터 접근과 트랜잭션을 어떻게 다룰 것인가에 초점을 맞춰봤다. 다음 글에서는 이 기반 위에서 Service 레이어를 어떻게 설계하고, 어디까지 역할을 줘야 하는지를 구체적인 예제와 함께 풀어보도록 하겠다.

특히 협업에 있어서 아키텍처의 부재는 큰 문제로 다가왔는데 프로젝트에 참여하는 인원의 언어 및 프레임워크 이해도, 개발 스타일, 심지어 변수 명을 짓는 방식까지 달라서 프로젝트가 진행되면 될수록 불어나는 덩치에 비례해 점점 유지보수에 대한 부담이 올라가기 시작했다. 기본적으로 Full Stack FastAPI Template을 따라 개발을 진행하긴 했다만 템플릿이 제시하는 개발 방향과 실제 필요한 개발 방향은 다르기 때문에 잘 지켜지지 않은 부분도 있었다. 결론적으로 더 이상 공통된 아키텍처가 없이 개발을 지속하는 것은 문제가 많다는 것이 팀의 전체 의견이었다.

기본 아키텍처

고려사항

아키텍처 설계를 고민하기에 앞서 몇 가지 우선적으로 고려할 사항이 있었다.

1. 코드 구조만 보고 바로 개발에 들어갈 수 있을 것
2. 프레임워크에 종속적이지 않을 것
3. 안티패턴이 바로 눈에 띌 것

기본적으로는 MVC 패턴을 따라하는 것으로 결정했다. 대부분의 개발자에게 익숙하기 때문에 협업에 빠르게 도입할 수 있다는게 주된 이유이다. 단, 어디까지나 개발 패턴을 따라하는 것이기 때문에 FastAPI에 맞게 재구성할 필요는 있었다. 그 결과 아래와 같은 구조로 뼈대를 세웠다.

├── app/ 
│ ├── main.py # FastAPI 앱 인스턴스 및 진입점 
│ ├── api/ 
│ │ ├── v1/ 
│ │ │ ├── users.py # 사용자 API 
│ │ │ └── items.py # 아이템 API 
│ │ ├── deps.py # FastAPI 의존성 어댑터 (DI 경계) 
│ │ └── router.py # 라우터 일괄 등록 
│ ├── core/ 
│ │ └── user_service.py # 비즈니스 규칙(프레임워크 비종속) 
│ ├── crud/ 
│ │ ├── repositories/ # Repository 구현(ORM 의존) 
│ │ └── uow.py # UnitOfWork (요청 단위 트랜잭션) 
│ │ └── repo_registery.py # 레포지토리 레지스트리 (선택)
│ ├── database/ 
│ │ └── session.py # 세션/엔진 설정(ORM) 
│ ├── schemas/ 
│ │ ├── dto/ # 내부 계층 간 DTO 
│ │ ├── request/ # API 입력 스키마 
│ │ └── response/ # API 출력 스키마 
│ │ ├── base_schema.py # 기본 스키마 형태 (선택)
│ ├── infra/ 
│ │ ├── config/ # 설정 로딩(Env, 파일) 
│ │ ├── utils/ # 범용 유틸 
│ │ ├── middlewares/ # 미들 웨어 설정 
│ │ └── global_state.py # 전역 싱글톤(프레임워크 비종속) 
│ └── template/ 
│ └── template.html 
├── tests/ 
├── requirements.txt 
└── README.md

api - Controller 레이어

api 폴더는 Controller를 담당한다. 고려사항에서 프레임워크에 종속적이지 않는 것을 강조했지만 이 부분은 실제 유저의 요청과 응답 부분을 담당하므로 어쩔수 없이 종속적일수 밖에 없어진다. 대신, 모든 프레임워크 관련 의존성을 이 레이어 안에 한정시켰다. deps.py는 이 구조의 핵심으로 FastAPI의 의존성 주입(Dependency Injection) 기능을 활용해, 라우터가 비즈니스 로직을 직접 알지 않고 필요한 객체를 주입받을 수 있게 한다. 즉, 프레임워크 경계에서 다른 계층으로 연결되는 단 하나의 진입점 역할을 수행한다.

core - Service 레이어

core 폴더는 Service를 담당한다. 앞서 말했든 해당 폴더는 오로지 비즈니스 로직에만 집중하고 전혀 FastAPI를 모른다. 모든 입출력은 DTO를 통해 이뤄지며 Repository를 통해 필요한 데이터를 전달받을 수 있도록 한다.

crud - Repository & UoW

crud 폴더는 Repository를 포함, ORM을 다루는 역할을 한다. 예전 Spring 개발을 할 때는 트랜잭션 경계와 관심사 분리 원칙 때문에 DTO 변환을 Service 레이어가 담당했지만 설계중인 아키텍처에서는 트랜잭션을 UoW(Unit of Works)가 담당한다. 또한 Lazy Loading을 통해 Service 레이어로 프록시 객체를 전달하는 Spring과는 다르게 SQLAlchemy에서는 ORM 객체를 직접 전달하기 때문에 혹시 모를 Service 단에서의 ORM 오염을 막기 위해 Repo에서 직접 DTO로 변환하는 방식을 택했다.

infra - 인프라 계층

나머지 모든 구성요소(설정, 미들웨어, 유틸 등)는 infra 폴더로 모았다. 이 계층은 프로젝트의 운영 환경을 담당하며, 비즈니스 로직과는 무관한 기술적 관심사를 분리하는 역할을 한다.

global_state.py는 전역 싱글턴 객체를 관리하는 모듈로, Spring의 Bean과 유사한 개념으로 이해할 수 있다. 실제 동작 방식은 조금 다르지만 전역 의존성을 중앙에서 일괄 관리한다는 목적은 동일하다. 이 부분은 이후에 더 자세히 다룰 예정이다.

참조 트리

폴더 구성만 빡세게 해놓고 참조 규약을 안만들면 그것만큼 의미없는 일이 없다고 생각한다. 예를 들어 api폴더 내 라우터들에서 core를 안거치고 직접 crud에 접근한다거나 하는 그런 일들 말이다. 그래서 참조 규약을 만들었다.

의도한 참조 방향

main.py
↓
api/router
↓
api/routes
↓
api/deps
↓
infra/global_state
↓
infra/utils
↓
infra/config

---

api/deps
↓
infra/global_state
↓
database

---

api/deps
↓
uow
↓
crud

---

api/v1/routes
↓
core
↓
schemas

모든 참조는 위에서 아래로만 가능하며 참조의 맨위에는 항상 api가 존재한다. FastAPI의 종속성은 main.py와 api 내부 경계 안쪽에만 존재해야 한다.

설계 철학

• 프레임워크 종속은 api 경계 안에 가둔다.
  • FastAPI의 요청, 응답, DI는 오직 main.py와 api 안쪽에서만 다룬다.
• 글로벌 싱글턴은 한곳에서만 관리한다.
  • 환경 설정, DB 세션, 외부 클라이언트(Qdrant, Redis 등) 등 모든 전역 객체는 이 파일을 통해 관리된다.
  • core나 crud가 직접 전역 객체를 import 하는 것은 금지된다.
• deps.py가 모든 계층 간 연결의 경계다.
  • deps.py는 FastAPI의 DI 레이어이자 경계 관리자 역할을 한다.
  • core와 crud, global_state.py가 여기에 주입되며 라우터는 오직 deps.py를 통해 필요한 객체를 의존성으로 전달받는다.

결국 라우터는 비즈니스 로직이 어디에 있는지를 몰라도 되며, FastAPI를 알고 있는 계층은 deps.py 이하로 한정된다.

금지 참조 예시

다음과 같은 참조는 모두 금지된다.

마무리

사실 그동안 프로젝트를 하면서 이렇게 본격적으로 직접 아키텍처를 설계한 경험은 처음이다. 때문에 그 밑바닥의 철학을 단단하게 구성하는게 무엇보다 중요하다는 생각을 했다. FastAPI, 파이썬 코드 치고는 지켜야할게 많고 그게 오히려 파이썬스럽지 않은 코드가 될 수도 있겠다는 생각도 했지만 돌고 돌아 어떤 언어든 간에 제일 중요한 원칙은 유지보수성이라는 결론이다.

다음 글에서는 이번에 정의한 구조를 실제 코드로 구현하면서

• deps.py를 통한 의존성 주입 흐름
• UnitOfWork의 트랜잭션 경계 처리
• Repository에서 DTO로의 변환 방식
• global_state를 통한 전역 객체 관리

등을 구체적인 예시 코드와 함께 다뤄보려고 한다. 만관부 -

덕분에 어떻게든 기능을 맞춰 넣고 돌아가게 만드는 게 우선이었고,
“일단 굴러가면 된다”가 나를 포함한 모두의 공통된 목표였다.
물론, 어디까지나 내 개인적인 생각이지만.

서비스는 결국 출시됐다. 지금은 적당히 안정적으로 돌아가고 있다.
목표 수준에는 못 미치지만, 조금씩 사용자도 늘고 있다.

다만, 촉박한 개발 일정 속에서 점점 덩치를 불려간 코드,
제대로 지켜지지 않은 아키텍처 설계,
그리고 협업이라기엔 너무 달랐던 각자의 개발 철학이 맞물리며
이제는 어디서부터 손대야 할지 감도 잡히지 않는 상태가 되어버렸다.

특정 비즈니스 로직이 어디에 있는지,
새로운 기능을 추가하려면 어느 폴더의 어느 파일을 수정해야 하는지조차 명확하지 않았다.

“잘 돌아가는데 뭐가 문제야?”

위에서 말한 모든 것이 문제였다.

솔직히 말하자면, 어떻게든 개발은 이어나갈 수 있다.
불편하고 복잡하더라도 하나씩 뜯어보면서 조금 고치고, 조금 때우면 일단 서버는 굴러간다.
하지만 그게 맞는 걸까?

서비스는 돌아가지만, 가독성도, 아키텍처도, 코드 스타일도 제각각이었다.
같은 기능을 구현하면서도 사용하는 라이브러리가 다르고,
한 파일은 Pydantic 모델을, 다른 파일은 dataclass를 쓰는 식이었다.
“기능이 되니까 됐다”는 방식으로 쌓여온 결과였다.

백엔드와 AI 서버가 한 배를 탔을 때

특히 서비스 구조상 백엔드와 AI 서버가 한 컨테이너 안에서 함께 돌아가고 있었다.
처음엔 편했다. 로컬 환경에서도 한 번에 올릴 수 있었고, 배포도 단순했다.
하지만 운영이 길어질수록 이 구조가 점점 발목을 잡기 시작했다.

AI 모듈은 LLM API 호출, 외부 통신, 비동기 작업 등 불안정한 요소가 많았다.
이쪽에서 문제가 생기면 백엔드 전체 요청이 지연되거나, 심한 경우 서버가 통째로 내려갔다.
단순히 “AI 응답이 늦다” 수준이 아니라, 로그인이나 단순 데이터 조회 같은 기본 기능까지 영향을 받았다.
결국 AI의 불안정성이 서비스 전체의 불안정성으로 전이되는 구조였다.

처음엔 단순히 “AI 쪽 에러 처리를 더 탄탄하게 하자” 정도로 접근했다.
하지만 코드를 살펴볼수록 문제는 단순한 예외 처리를 넘어 있었다.
서비스의 경계가 모호했고, 책임이 얽혀 있었다.

AI 로직은 “생성”과 “추론”에 집중해야 하고,
백엔드는 “요청 흐름”과 “데이터 관리”에 집중해야 한다.
하지만 지금 구조는 이 두 역할이 한 덩어리로 섞여 있었다.

결국 장애가 나면 “이게 AI 문제인지, DB 세션 문제인지, 네트워크 문제인지”
명확히 판단하기 어려웠고, 이는 운영 리스크로 이어졌다.

구조는 결국 “시간”과 “책임”의 문제

이쯤 되니 단순히 코드를 조금 정리하는 수준으로는 해결되지 않았다.
문제는 기술적인 차원이 아니라, 구조적인 차원이었다.
더 정확히 말하면, 시간과 책임의 문제였다.

그동안 코드가 이렇게 된 이유는 단순했다 — 시간이 없었기 때문이다.
하지만 근본적으로는 각 기능이 맡아야 할 역할과 책임이 명확히 정의되지 않았기 때문이다.
명확한 경계가 없으면, 시간이 지나면서 모든 코드가 서로 기대게 된다.
결국 한 부분의 변경이 다른 부분을 깨뜨리는 구조가 되어버린다.

그래서, 분리를 시작했다

결론은 명확했다.
AI와 백엔드를 완전히 분리하자.

Docker Compose 파일을 나누고, 서비스 단위를 다시 정의했다.
AI 서버는 “추론과 응답 생성”에만 집중하도록 하고,
백엔드는 “비즈니스 로직과 데이터 흐름”을 담당하도록 역할을 나눴다.

자, 설계는 끝났고 이제 구현만 남았다.
하지만 막상 손을 대보니 이게 생각처럼 단순하지 않았다.
컨테이너를 두 개로 나눈다고 끝나는 일이 아니었다.
각 서비스가 어떤 식으로 통신해야 하는지,
환경 변수를 어디서 관리해야 하는지,
공통으로 접근해야 하는 모듈은 어떻게 공유할지 —
모든 게 다시 설계의 영역이었다.

하다 보니 깨달았다.
Docker에 대해서도, Python에 대해서도, FastAPI에 대해서도 여전히 모르는 게 많았다.
그동안 써왔던 기술이 아니라,
“왜 그렇게 썼는가”를 다시 묻는 단계였다.

그래서 이 시리즈를 쓰기로 했다.
이건 단순히 Compose 파일을 나누는 법을 정리한 글이 아니다.
짧은 일정 속에서 구조적 문제를 마주하고,
그걸 해결하기 위해 부딪히며 배운 과정을 기록하는 글이다.

누군가에게는 당연한 내용일 수도 있지만,
적어도 나에게는 “잘 돌아가는 서비스”와 “유지보수 가능한 구조”의 차이를
처음으로 실감하게 해준 경험이었다.

LLM 오케스트레이션 도구의 중요성

ChatGPT의 등장은 AI 개발 환경의 판도를 크게 바꿔놓았다. 이제는 사내 챗봇부터 고객상담, 자동화 업무까지 거의 모든 AI 프로젝트에 LLM이 핵심 컴포넌트로 자리잡았다. 하지만 LLM 하나만으로 실무 시스템을 구성하기엔 여전히 넘어야 할 장벽이 많다. 프롬프트 설계, 입력/출력 파싱, 다양한 외부 데이터 연동, 오류 처리 등 수많은 작업이 남아 있다.

이런 현실적인 요구를 해결하기 위해 다양한 오케스트레이션 프레임워크들이 등장했고, 그 중에서도 LangChain은 빠른 시기에 가장 널리 사용되는 표준처럼 자리잡았다. 실제로 LangChain은 LLM을 여러 도구와 체인 형태로 손쉽게 연결할 수 있게 해줘서, 프로토타이핑 단계에서는 분명 큰 이점을 줬다.

하지만 막상 실무에서 장기적으로 서비스 운영을 하다 보면 “잘 돌아가는 것”과 “잘 유지보수되는 것”은 전혀 다르다는 걸 체감하게 된다. 단순한 예제에서는 문제 없던 구조가, 실제 운영 환경에서는 복잡해지고 점점 디버깅과 확장, 유지보수에 발목을 잡게 되는 것이다.

이 포스팅에서는 이런 배경에서 실제로 느꼈던 LangChain의 한계, 그리고 그 대안으로써 최근 주목받고 있는 Pydantic AI를 비교 분석해보려 한다. 개발자로서 실제 운영·개발 과정에서 마주한 경험을 바탕으로, 어떤 상황에 어떤 선택이 더 맞았는지 솔직하게 이야기해보고자 한다.

LangChain의 한계

과도한 추상화와 복잡성

LangChain의 가장 큰 문제는 제목에서도 알 수 있듯이 "과도한 추상화"에 있다.

처음에는 각 LLM 호출이나 외부 툴 연결, 입력과 출력을 체인으로 손쉽게 엮을 수 있다는 점이 매력적으로 보인다. 하지만 프로젝트가 조금만 복잡해지면, 오히려 이 추상화가 발목을 잡는다. 실제 현업에서는 새로운 기능을 추가하거나 기존 체인을 조금만 수정하려 해도, 체인 전체 구조를 따라가며 일일이 확인해야 한다. 입력값이 어떻게 변환되는지, 어느 단계에서 어떤 데이터가 주고받아지는지 일관성 있게 추적하는 것도 쉽지 않다.

이처럼 단순한 데모나 프로토타이핑 단계에서는 LangChain의 추상화가 빠른 개발에 도움을 주지만, 실무 환경에선 디버깅과 유지보수의 허들이 된다. 에러가 발생하면 어디서 문제가 생긴 건지 파악하려면 추상화 레이어를 하나씩 뜯어봐야 하고, LangChain 내부 구조를 깊이 이해하지 않으면 원인 분석도 어려워진다. 또, 파이썬 개발자라면 당연히 기대하는 “직관적인 코드 흐름”보다는, LangChain에서 요구하는 스타일에 맞춰 코드를 작성해야 한다는 점도 꽤 답답하다.

간단한 예시를 가져와봤다.

from langchain.chains import SequentialChain, LLMChain
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

# 모델 선언
llm = OpenAI()

# 문서 요약 프롬프트
summary_prompt = PromptTemplate(
    input_variables=["text"],
    template="다음 문서를 3줄로 요약해줘: {text}"
)

summary_chain = LLMChain(llm=llm, prompt=summary_prompt, output_key="summary")

# 키워드 추출 프롬프트
keyword_prompt = PromptTemplate(
    input_variables=["summary"],
    template="다음 요약에서 핵심 키워드 5개를 뽑아줘: {summary}"
)
keyword_chain = LLMChain(llm=llm, prompt=keyword_prompt, output_key="keywords")

# 체인 연결
overall_chain = SequentialChain(
    chains=[summary_chain, keyword_chain],
    input_variables=["text"],
    output_variables=["keywords"]
)

result = overall_chain.run({"text": "LangChain은 문제가 많다."})
print(result["keywords"])

간단하게 문서 요약 → 키워드 추출로 이어지는 체인이다. 이 간단한 체인을 위해서도 각 프롬프트를 LLMChain으로 감싸고, 다시 SequentialChain으로 한 번 더 감싸야 한다.

여기에 이어서 각 LLMChain에는 input/output key를 일일이 맞춰줘야 하고, 체인 전체의 입력/출력 변수도 별도로 정의해줘야 한다. 단순히 두 번의 LLM 호출을 연결하는 것뿐인데, 코드 구조는 점점 복잡해지고, 중간 결과를 추적하거나 에러가 발생했을 때 어느 단계에서 문제가 생겼는지 확인하는 것도 쉽지 않다.

실제로 프로젝트가 커질수록 “프롬프트 추가/수정이 필요할 때마다 체인 전체 구조를 다시 검토해야 한다” “체인 간 데이터 전달 과정에서 변수명이 꼬이거나 누락되는 일이 반복된다” “디버깅이나 테스트도 LangChain의 추상 구조를 따라가야 하니 부담이 크다” 는 점이 발목을 잡는다.

이처럼 아주 단순한 파이프라인에서도 LangChain의 추상화 레이어가 오히려 개발자의 생산성과 코드의 명확성을 저해하는 상황이 반복된다. 결국 실무에서는, “이 정도 로직이라면 그냥 파이썬 함수 두 개로 처리하는 게 훨씬 낫지 않을까?” 라는 생각이 들 수밖에 없다. 실제로 LangChain을 쓰지 않고 직접 OpenAI API를 사용해 아래 코드와 같이 간단하게 끝낼 수 있다.

from openai import OpenAI
client = OpenAI()

def get_response(prompt: str):
    client = OpenAI()
    response = client.responses.create(
        input=prompt
    )
    return response.output_text

def get_summary(text: str) -> str:
    prompt = f"다음 문서를 3줄로 요약해줘: {text}"
    # 실제 LLM 호출
    return get_response(prompt)

def get_keywords(summary: str) -> list[str]:
    prompt = f"다음 요약에서 핵심 키워드 5개를 뽑아줘: {summary}"
    return get_response(prompt)

summary = get_summary("LangChain은 문제가 많다.")
keywords = get_keywords(summary)
print(keywords)

이 경우 함수 기반으로 코드가 짜여져있어 각 단계를 명확히 분리해 원하는 방식대로 호출/테스트/디버깅이 가능해진다. 즉, 프레임워크가 강제하는 구조가 아니라 파이썬의 설계 방식과 동일하게 유지보수가 가능해진다는 것이다. 결국 중요한건 "프레임워크가 모든 걸 다 해주지 않는다" 라는 것이다. 누군가 100줄 코드로 LangChain 구성하기라는 제목으로 같은 문제점을 지적한 글이 있는데 링크를 남겨둘테니 관심있다면 읽어보는 것이 좋겠다.

Re-implementing LangChain in 100 lines of code

문서와 커뮤니티의 장벽

위의 예시와는 다르게 개발 자체와는 거리가 좀 있는 문제이긴 하지만 어쨌거나 개발자 입장에서 지나칠 수 없는 문제가 바로 부실한 공식문서이다. LangChain은 빠르게 성장하는 오픈소스 프로젝트이다 보니, 공식 문서가 자주 바뀌고, 주요 기능이나 API 사용법이 제대로 정리되지 않은 경우가 많다. 내가 참고했던 예제가 어느새 구버전이 되어버리거나, 실제로는 동작하지 않는 코드가 여전히 공식 문서에 남아있는 경우도 심심치 않게 발견된다.

또한 실무에서 자주 마주치는 복잡한 사례나 에러 케이스에 대한 설명은 거의 찾아보기 어렵다. 단순한 예제는 많은데, 실제로 서비스에 붙였을 때 발생하는 문제(예: 입력/출력 타입 불일치, 체인 중간 단계 오류, 외부 데이터 연동 등)에 대한 해결책이나 경험담은 공식 문서나 커뮤니티에서도 찾기 힘들다.

결국, 제대로 된 정보를 얻으려면 깃허브 이슈를 뒤지거나, 커뮤니티 포럼/슬랙 등에서 비슷한 사례를 찾아서 직접 코드를 분석해봐야 하는 상황이 자주 발생한다. 이런 과정은 개발 생산성을 떨어뜨리고, 온보딩된 팀원이나 초보자들에게는 진입장벽으로 작용한다.

실무에서 “구체적인 레퍼런스가 필요한 순간”마다 “이렇게 빠르게 바뀌는 생태계에서 안정적으로 코드를 운영할 수 있을까?” 라는 불안감이 커질 수밖에 없다.

아래는 실제 reddit에서 LangChain에 신나게 데이고 극대노한 개발자의 생생한 증언이다.

이렇게까지 꾹꾹 눌러담은 분노라니 새삼 reddit의 번역기능이 놀라울 따름이다.

디버깅과 유지보수의 어려움

다시 개발얘기로 돌아오자면, 첫 번째 문제와 이어지는 디버깅의 문제가 있겠다. 앞서 언급했듯이, LangChain의 핵심은 여러 추상화 레이어를 겹겹이 쌓아서 복잡한 LLM 워크플로를 간편하게 조립하는 것이다. 하지만 실전에서 이 구조가 오히려 “문제가 발생했을 때 원인을 빠르게 파악하는 데 심각한 장애물”로 작용한다.

예를 들어,

체인 내부에서 오류가 발생해도, 스택 트레이스는 LangChain 라이브러리 깊은 곳까지 들어가 있다가 터진다. 내가 작성한 프롬프트에 오타가 있었는지, 중간 단계에서 데이터가 잘못 넘어갔는지 바로 알 수 없다.

체인 사이 데이터 전달이 꼬이면, output_key/input_key 실수, 프롬프트 포맷 에러, 예상치 못한 응답 파싱 오류 등이 한꺼번에 겹쳐서 에러의 근본 원인(프롬프트? 체인 연결? 외부 API?)을 추적하는 데 한참을 허비하게 된다.

단순히 프롬프트 한 줄을 바꿔도 “이게 어느 체인에, 어떤 변수명으로 연결되어 있었지?” “중간 결과가 정상적으로 전달되고 있는 게 맞나?” 이런 걸 체인 전체 코드를 하나씩 확인하면서 추적해야 하는 일이 반복된다.

뭐, 이미 추상화 얘기에서 비슷한 얘기를 많이 한것 같아 이 부분은 대충 넘어가도록 하겠다.

Pydantic AI는 좀 다른가?

Python 개발자라면 언젠가 한번은 들어봤을 유효성 검사 라이브러리 Pydantic이 LLM 오케스트레이션 도구를 만들었다. 바로 Pydantic AI다. 솔직히 말하자면 내가 Pydantic AI를 처음 접하고 쓰기 시작한 것은 LangChain을 대신할 라이브러리를 찾고 있었기 때문이 아니라 순전히 호기심 때문이었다. LangChain이 짜증나긴 하지만 마땅한 대책이 없으니 울며 겨자먹기로 써가던 와중에 정말 우연찮게 Pydantic AI의 존재를 알았고 몇번 시험삼아 써보니 어느 순간 LangChain의 존재는 까마득하게 잊혀져 있었다. 사실 공개된지 얼마 안된 라이브러리기도 하고 LangChain 못지않게 빠르게 개발이 진행되면서 이것저것 많이 바뀌고 있는 상황이긴 하지만 그 모든 것들을 제치고서 사용성이 말도 안되게 좋다는 이유 하나로 이 글을 쓰게되었다.

타입 기반의 구조화된 입출력

LangChain의 단점에서 따로 이야기하진 않았지만 LLM을 사용한 개발에 있어서 항상 고민되는 문제가 있다. 바로 입출력 구조가 100% 보장되느냐는 점이다.

일반적인 소프트웨어 개발에서는 함수의 입력과 출력 타입이 명확하게 정의되고, 이 타입에 어긋나는 데이터가 들어오면 곧바로 예외가 발생한다. 하지만 LLM을 사용하는 환경에서는 프롬프트에서 아무리 명확하게 JSON 형식, 혹은 특정 스키마로 답하라고 요구해도 실제로 반환되는 값이 미묘하게 다르거나, 필드가 빠지거나, 예상과는 전혀 다른 형태로 응답하는 경우가 생기지 않는다고는 못하겠다.

LangChain을 사용할 적엔 기본적으로 프롬프트 내부에 답변 형식을 강제하는 프롬프트를 적어야했다. 이 방식은 어디까지나 LLM의 "착실함"에 의존하는 방식이다. 물론 최근 OpenAI나 Perplexity나 구조화된 답변을 받도록 강제하는 Tool이 제공되긴한다만, LangChain에서 이를 사용하는지는 잘 모르겠다.

하여간에 이런 문제는 개발 과정에서 잘 동작하던 체인이 실제 서비스에서는 예외 상황에서 바로 오류를 뱉어내거나 전체 워크플로우가 중단되는 경우를 야기할 수 있다. Pydantic AI의 가장 큰 장점이 바로 이 지점에서 온다. Pydantic이 유효성 검사로 이름을 날린 라이브러리라 그런지 Pydantic AI도 입출력 구조화에 진심이다. LLM을 호출할 때의 입력 데이터는 물론이고 모델이 반환하는 출력도 Pydantic 모델로 엄격히 관리한다.

예를 들어 아래 코드와 같이 입출력을 선언해두면,

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext
from dataclasses import dataclass

@dataclass
class AgentDeps:
    text: str

class AgentOutput(BaseModel):
    summary: str

agent = Agent(
            model = model,
            deps_type = AgentDeps,
            output_type = AgentOutput
        )

이제 입력에 필요한 구조는 AgentDeps를 통해 관리하고 출력은 AgentOutput이 관리하게 된다. 물론 실제 쓰임세는 조금씩 다르긴 하지만 그건 나중에 실제 사용하는 방법을 포스팅할때 얘기하도록 하고 지금은 이 정도로 이해하고 넘어가면 좋겠다. 내부적으로 라이브러리를 뜯어보니 동작은 위에서 언급했던 구조화된 답변을 받도록 강제하는 Tool을 명시적으로 불러오도록 되어있는 것 같다. 뭐 어쨌거나 덕분에 우리는 LLM이 기대한 형식에 맞지 않는 응답을 보내면 곧바로 예외를 통해 감지할 수 있고 중간 데이터 흐름을 명확하게 추적할 수 있게 된다.

함수형 설계의 장점

Pydantic AI는 복잡한 체인, 에이전트, 툴 등 추상화 레이어를 과하게 쌓지 않고, LLM 호출을 타입이 명확한 파이썬 함수처럼 다룰 수 있게 해준다.

이 구조의 가장 큰 장점은 각 단계를 완전히 독립적인 함수 단위로 분리할 수 있다는 것이다. 예를 들어, "문서 요약 → 키워드 추출 → 결과 정제" 같은 파이프라인을 구현한다고 할 때, 각 단계별 함수가 입력과 출력 타입을 명확히 갖추고 있으니 특정 단계만 따로 수정하거나, 테스트를 따로 돌리거나, 필요하면 순서를 바꾸거나, 중간에 새로운 기능(예: 추가적인 LLM 호출, 외부 API 연동)을 끼워넣는 것도 매우 쉽고, 안전하다.

LangChain은 그럼 이런 것들이 불가능한가? 아니다 분명 이것들은 LangChain을 쓰면서도 충분히 할 수 있는 작업들이다. 다만 작업의 난이도가 다르다. 입출력을 조금 바꿔야하는 경우에도 LangChain은 끄적끄적 프롬프트를 손봐야한다. 중간에 새로운 기능을 추가하려면 기존 체인을 박살내고 다시 연결시켜줘야한다. 이런 작업들은 어려운게 아니라 귀찮고 피곤하다는게 문제다.

실제로 팀 프로젝트를 하다 보면, "여기 요약 프롬프트만 살짝 바꿔주세요", "여기에 추가 데이터 전처리 코드를 넣고 싶어요" 같은 요청이 빈번하게 들어온다. 함수형으로 에이전트를 관리하는 방식이라면, 해당 함수만 수정하고, 입출력 타입만 확인하면 되기 때문에 예상치 못한 버그가 터질 일이 거의 없다.

이어지는 장점인데 팀원 온보딩이나 협업도 훨씬 수월하다. LangChain처럼 체인-에이전트-툴의 추상화된 구조와, 각종 output_key, input_key, 체인 연결 규칙을 처음부터 이해해야 하는 게 아니라 "이 함수는 어떤 입력을 받고, 어떤 출력을 준다" "이 중간 단계가 문제면, 해당 함수만 보면 된다" "전체 파이프라인은 파이썬 함수 호출 순서와 거의 같다" 라는 직관적인 개발 방식이 유지된다.

즉, 새로운 팀원이 합류해도 코드 흐름을 한 번에 파악할 수 있고, 각 단계별 책임이 명확하니 서로 충돌 없이 개발을 병렬로 진행하기도 쉽다.

그래서 Pydantic AI는 무적인가?

아니다. 다시 한번 말하지만 프레임워크가 모든 걸 다 해주지 않는다. 개인적으로 느끼는 Pydantic AI의 단점은 지원되는 LLM API 서비스가 그닥 많지 않다는 것이다. 물론 이건 다른 라이브러리도 마찬가지겠지만 쓰고자하는 API가 호환이 안된다면 직접 구현하는 수밖에 없다. 가만 생각해보면 OpenAI, Gemini, Ollama 정도만 지원되면 장땡 아닌가 싶기도 하지만 어쨌든 특수한 상황이 없는건 아니기 때문이다.

당장 최근에만 해도 Perplexity를 연동해야하는 상황이 있었는데 Pydantic AI에서는 아직 지원되는 형식이 아니어서 당황했던 적이 있다. 다행인건 OpenAI와 같은 규격의 API를 사용하면 쓸수 있다는 문서가 있어 문제없이 넘어가긴 했다만 세상 모든 LLM이 OpenAI의 API와 호환되는건 아닐거라 생각한다. 그런 부분에 있어서 당분간 Pydantic AI에서는 "OpenAI와 호환되는 서비스를 이용하라" 라고 안내를 하고있다. 이게 LLM 분야 전부가 공유하는 내용인지는 모르겠지만 어쨌든 호환 안되는 모델을 써야한다면 결국 구현해야하는건 나다.

그럼에도 당분간, 어쩌면 한동안은 LLM 오케스트레이션 도구로는 Pydantic AI를 채택할 것 같다. 지금까지의 경험을 돌아보면, LangChain은 프로토타입 단계나 다양한 외부 툴 연동이 필요할 때에는 확실한 강점이 있다. 하지만 실서비스, 특히 구조화된 데이터 흐름과 안정성, 유지보수를 우선시해야 하는 프로젝트라면 Pydantic AI와 같이 “타입 중심”의 심플하고 명확한 설계를 제공하는 프레임워크가 훨씬 큰 만족을 주는 듯하다.

으로 더 많은 LLM API가 표준화되고, Pydantic AI도 더 다양한 모델과 연동이 쉬워진다면 LLM을 다루는 개발자들에게 점점 더 매력적인 선택지가 될 것이라 생각한다.

여담이지만 해외에서는 많이들 Pydantic AI를 사용하는 것 같은데 아직 한국에서는 잘 알려지지 않은 것 같다. 혹시라도 나중에 Pydantic AI가 한국에서도 유명해지면 당장 이 글을 보여주면서 내가 선구자 중에 한명이었다! 라고 자랑할 수 있었으면 좋겠다.