pydantic-ai, FastAPI, Linear를 사용하여 이메일 자동 분류 시스템을 구축한 방법

고객 지원(Support) 이메일은 선의가 묻히는 무덤과 같습니다. 제가 함께 일했던 모든 팀은 동일한 문제의 변형을 겪고 있었습니다. 공유 편지함(shared inbox)에 이메일이 쌓이면, 누군가가 이를 수동으로 읽고, 이것이 버그인지 결제 관련 질문인지 결정한 뒤, 텍스트를 Linear 티켓으로 복사하고, 직관에 따라 우선순위를 할당하며, 긴급해 보이면 Slack으로 알림을 보냅니다. 이 과정은 운이 좋은 날에도 이메일 한 통당 5~10분이 소요되며, 확장성(scale) 측면에서 매우 좋지 않습니다.

이 글에서는 사람의 개입 없이(without a human in the loop) 수신된 이메일을 분류하고, 구조화된 Linear 이슈를 생성하며, 중요한 사항에 대해 Slack 알림을 보내는 전체 루프를 처리하는 자동 분류 파이프라인의 아키텍처(architecture)와 주요 코드 패턴(code patterns)을 살펴봅니다.

문제점: 수동 분류는 확장할 수 없다

이 프로젝트를 시작하게 된 구체적인 시나리오는 다음과 같습니다.

규모가 작은 SaaS 팀은 하루에 80~150개의 고객 지원 이메일을 받습니다. 여기서 일관되게 중요한 세 가지 카테고리가 있습니다: 버그 (bugs) (고객이 보고한 충돌 또는 기능 고장), 결제 문제 (billing issues) (결제 실패, 잘못된 인보이스), 그리고 기능 요청 (feature requests) (제품 검토가 필요한 추가 기능 제안)입니다. 그 외의 모든 것은 일반 문의이거나 노이즈(noise)입니다.

자동화가 없다면 다음과 같은 일이 발생합니다: 밤사이 이메일이 쌓입니다. 아침에 가장 먼저 출근한 엔지니어는 코드 한 줄을 쓰기도 전에 45분 동안 분류 작업에 시간을 보냅니다. 새벽 2시에 도착한 유료 고객의 P0 버그 보고서는 오전 9시까지 읽히지 않은 채 방치됩니다. 다른 Slack 채널로 전달되어야 할 결제 문제는 엔지니어링 큐(queue)에서 길을 잃습니다. 기능 요청은 아무도 복사-붙여넣기 작업을 하고 싶어 하지 않기 때문에 백로그(backlog)에 포함되지 못합니다.

진정한 비용은 이메일당 소요되는 시간이 아닙니다. 일관성 없게 내려지는 결정, 너무 오래 방치되는 중요한 티켓, 그리고 매일 아침 지원 모드로 컨텍스트 스위칭(context-switching)을 할 때 발생하는 인지 부하(cognitive load)가 진짜 문제입니다. 수동 분류는 실제로 측정해 보기 전까지는 관리 가능한 것처럼 보이는 프로세스입니다.

아키텍처: pydantic-ai + FastAPI를 중추로 사용

여기서 핵심적인 통찰은 이메일 분류(triage)가 생성(generative) 문제가 아니라 구조화된 추출(structured extraction) 문제라는 점입니다. LLM에게 창의적인 무언가를 쓰라고 요청하는 것이 아닙니다. 텍스트를 읽고 카테고리(category), 우선순위(priority), 요약(summary), 권장 담당자(suggested assignee)와 같은 특정 필드로 구성된 양식을 채우라고 요청하는 것입니다. 이것이 바로 pydantic-ai가 설계된 목적입니다.

왜 LangChain이나 일반적인 OpenAI 요청 대신 pydantic-ai를 사용하는가?

LangChain은 필요하지 않은 문제에 대해 너무 많은 추상화(abstraction)를 추가합니다. LangChain의 출력 파서(output parsers)는 마치 나중에 덧붙여진(bolted on) 느낌을 줍니다. 일반적인 OpenAI API 호출은 JSON 스키마(JSON schema) 정의를 수동으로 작성해야 하며, 그 후 출력을 직접 검증해야 합니다. 이는 필연적으로 취약한 문자열 파싱(string parsing) 코드를 작성하게 만듭니다.

pydantic-ai를 사용하면 기대하는 출력값으로 Pydantic 모델을 정의할 수 있으며, 라이브러리가 프롬프트 전략(prompting strategy)과 검증 루프(validation loop)를 처리합니다. 만약 LLM이 잘못된 형식을 반환하면, pydantic-ai는 검증 오류를 컨텍스트(context)에 포함하여 재시도합니다. 실제로 이는 모든 에이전트 호출로부터 올바른 키가 들어있기를 기대해야 하는 딕셔너리(dictionary) 대신, 타입이 지정되고 검증된 객체(object)를 돌려받는다는 것을 의미합니다.

FastAPI는 이 모든 과정을 웹훅 엔드포인트(webhook endpoint)로 감쌉니다. Gmail은 IMAP 폴링(polling)을 통해 이벤트를 전송하고(또는 푸시 웹훅으로 교체할 수도 있습니다), FastAPI 핸들러는 에이전트를 통해 이메일을 처리한 다음 Linear 및 Slack API 호출을 실행합니다. 이를 통해 파이프라인을 상태가 없는(stateless) 상태로 유지하고 배포하기 쉽게 만듭니다.

핵심 설계 결정: 각 이메일은 완전히 구조화된 분류 객체를 반환하는 단 한 번의 에이전트 호출을 거칩니다. 호출 체인(chain of calls)도 없고, 메모리(memory)도 없으며, 대화 상태(conversation state)도 없습니다. 덕분에 시스템은 예측 가능하고, 실행 비용이 저렴하며, 디버깅이 쉽습니다. 이메일 한 통당 약 300~500개의 입력 토큰(input tokens)이 소모되는데, 현재 GPT-4o-mini 가격 기준으로 이는 1센트의 아주 작은 일부에 불과합니다.

핵심 코드 패턴: pydantic-ai를 이용한 구조화된 분류

다음은 단순화되었지만 실제 작동하는 시스템의 핵심 코드입니다:

from pydantic import BaseModel, Field
from pydantic_ai import Agent
from enum import Enum
...

여기서 설명할 가치가 있는 몇 가지 사항이 있습니다:

각 모델 필드에 사용된 Field(description=...)는 단순한 문서화가 아닙니다. pydantic-ai는 이 설명을 LLM의 출력을 안내하는 스키마 (schema)에 전달합니다. 이는 장황한 퓨샷 (few-shot) 예시를 작성하지 않고도 모델의 동작을 제약할 수 있는 방법입니다. needs_immediate_slack_alert에 작성된 설명은 비즈니스 로직을 타입 정의 (type definition)에 직접 내장합니다.

2,000자에서 본문을 자르는 것은 의도적인 설계입니다. 고객 지원 이메일은 짧거나 (중요한 신호가 첫 번째 단락에 있음), 혹은 매우 길거나 (전달된 스레드, 텍스트로 붙여넣은 첨부 로그 등) 둘 중 하나입니다. 본문을 자름으로써 비용을 예측 가능하게 유지하고, 가끔 발생하는 긴 이메일이 토큰 예산 (token budget)을 모두 소진하는 것을 방지합니다.

system_prompt에는 CRITICAL을 사용하지 않아야 할 때에 대한 명시적인 지침이 포함되어 있습니다. 이 지침이 없으면 LLM은 사용자의 알림 피로도 (alert fatigue) 임계값이 어느 정도인지 알 수 없기 때문에 과도하게 에스컬레이션 (escalate)하는 경향이 있습니다.

통합: Gmail에서 Linear를 거쳐 Slack까지

데이터 흐름은 다음과 같이 작동합니다:

1. FastAPI 백그라운드 태스크 (background task)가 60초마다 IMAP을 통해 Gmail을 폴링 (poll)하여 고객 지원 수신함에서 읽지 않은 이메일을 가져옵니다.
2. 각 이메일은 triage_email()을 거쳐 TriageResult를 반환합니다.
3. 결과는 Linear GraphQL API를 통해 Linear 이슈 (issue)로 매핑됩니다. 카테고리는 라벨 (label)이 되고, 우선순위는 Linear의 1-4 단계로 매핑되며, 요약은 이슈 제목이 됩니다.
4. 만약 needs_immediate_slack_alert가 true라면, 파이프라인은 발신자, 요약, 그리고 새로 생성된 Linear 이슈로의 직접 링크를 포함하여 #critical-support Slack 채널에 메시지를 게시합니다.

async def process_email(email: ParsedEmail):
    triage = await triage_email(email.subject, email.body, email.sender)

...

알아두어야 할 주의사항 (The gotcha): Linear의 GraphQL API를 사용하려면 이슈를 생성하기 전에 팀 ID (team IDs)와 라벨 ID (label IDs)를 먼저 가져와야 합니다. 이 ID들은 워크스페이스 (workspace)별로 다르며 사람이 읽을 수 있는 형식이 아닙니다. 프로덕션 버전에서는 매 이메일마다 이를 가져오는 대신 시작 시점에 이를 캐싱 (cache)합니다. 이는 장애 발생 후 20개의 이메일이 한꺼번에 몰려 처리해야 할 때 매우 중요합니다.

트레이드오프 및 한계점 (Tradeoffs and Limitations)

이 접근 방식은 이메일 양이 비교적 일정하고 카테고리가 잘 정의된 팀에게는 효과적입니다. 하지만 다음과 같은 몇 가지 사항은 깔끔하게 처리하지 못합니다:

스레드 문맥 (Thread context)이 유실됩니다. 각 이메일은 독립적으로 처리됩니다. 만약 고객이 기존 스레드에 답장을 보낸다면, 시스템은 기존 이슈에 내용을 추가하는 대신 Linear 이슈를 중복해서 생성하게 됩니다. 이를 해결하려면 이메일 스레딩 로직 (제목 또는 Message-ID 헤더를 통한 매칭)이 필요한데, 이는 상당한 복잡성을 추가합니다.

LLM 분류에는 오차 범위가 존재합니다. 테스트 결과, 약 3~5%의 이메일에서 카테고리가 잘못 지정되었습니다. 모호한 이메일 ("당신의 도구가 제 데이터를 모두 삭제했지만, 환불을 요청하고 싶으면서 동시에 엔터프라이즈 플랜에 대해서도 묻고 싶습니다")은 모델이 우선순위를 두는 카테고리로 할당됩니다. 따라서 HIGH 우선순위 미만의 모든 항목에 대해서는 여전히 사람이 검토하는 큐 (queue)가 필요합니다.

IMAP 폴링 (polling)은 대량 처리에 이상적이지 않습니다. 하루에 수천 통의 이메일을 처리해야 한다면, Gmail의 Pub/Sub 푸시 알림이나 적절한 이메일 처리 서비스로 전환해야 합니다. 60초마다 폴링하는 방식은 대부분의 고객 지원 편지함에는 충분합니다.

이메일 양이 매우 적다면, 이 방식은 아마도 과잉 설계 (over-engineered)일 것입니다. 단순한 필터 규칙과 Zapier 워크플로우를 사용하는 것이 더 적절한 선택일 수 있습니다.

마치며 (Closing)

이 파이프라인은 테스트에 참여한 팀의 아침 분류 작업 (triage ritual)을 없애주었습니다. 엔지니어들은 이메일을 읽으며 하루를 시작하는 일을 멈췄습니다. 중요한 티켓들은 몇 시간 뒤가 아니라 도착 후 2분 이내에 Slack으로 들어오기 시작했습니다.

저는 이 프로젝트를 오후 한나절이면 배포할 수 있는 오픈 소스 템플릿으로 패키징했습니다:

GitHub 스캐폴드 (scaffold): https://github.com/Reactance0083/pydantic-ai-email-linear-auto-triage

스캐폴드는 핵심 아키텍처를 제공합니다. 적절한 에러 핸들링 (error handling), 재시도 로직 (retry logic), 이메일 스레드 중복 제거 (email thread deduplication), 테스트 스위트 (test suite) 및 배포 설정이 포함된 전체 프로덕션 버전은 여기서 확인할 수 있습니다:

전체 프로덕션 코드 (Full production code): https://reactance0083.gumroad.com/l/dcror

만약 여러분도 이와 유사한 것을 구축했거나, 프로덕션 환경에서 LLM (Large Language Model) 기반 분류 작업을 수행하며 다른 예외 케이스 (edge cases)를 겪었다면, 댓글을 통해 진심으로 이야기를 듣고 싶습니다. 특히 스레드 매칭 (thread-matching) 문제를 깔끔하게 해결한 분이 계신지 매우 궁금합니다.