pydantic-ai와 FastAPI를 사용하여 이메일을 Linear로 자동 분류하는 에이전트를 구축한 방법

대부분의 기업에 있는 지원 엔지니어(Support engineers)들은 조용한 좌절감을 공유합니다. 그들은 매일 아침 상당한 시간을 마치 로봇처럼 느껴지는 작업에 소비합니다. 이메일을 읽고, 유형이 무엇인지 결정하고, 우선순위를 추측하고, Linear를 열고, 티켓을 생성하고, 세부 정보를 붙여넣고, 만약 긴급해 보인다면 Slack에서 누군가에게 알림을 보냅니다. 작업 자체는 기계적입니다. 그 작업에 필요한 판단이 항상 사소한 것은 아니지만, 프로세스 자체는 분명히 기계적입니다.

저는 pydantic-ai, FastAPI, Gmail IMAP, Linear API, 그리고 Slack API를 사용하여 이 루프를 제거하는 시스템을 구축했습니다. 이 글에서는 아키텍처(architecture), 핵심 코드 패턴, 그리고 이와 같은 시스템을 프로덕션(production) 환경에서 사용하기 전에 알아야 할 솔직한 트레이드오프(tradeoffs)에 대해 설명합니다.

문제점: 모든 지원 팀에는 여전히 수동 분류 작업이 존재함

자동화가 없을 때 실제로 일어나는 일은 다음과 같습니다. 새벽 2시 47분에 지원 이메일이 도착합니다. 내용은 "결제 흐름 전체가 고장 났습니다. 주문이 전혀 들어가지 않습니다"와 같습니다. 이 메일은 공유 편지함에 머물러 있습니다. 누군가 오전 8시에 이를 확인합니다. 그들은 수동으로 Linear 티켓을 생성하고, P1 라벨을 붙이고, 당직 엔지니어(on-call engineer)에게 할당한 다음, Slack 메시지를 보냅니다. 그 시점에 회사는 잠재적인 매출 회복 기회 5시간을 놓친 상태입니다.

좌절스러운 부분은 대부분의 팀이 이를 해결하려고 _시도해 보았다_는 점입니다. Zapier 규칙은 이메일 제목이 약간만 바뀌어도 깨집니다. 정규 표현식(Regex) 기반 분류기는 새로운 이메일 패턴이 나타날 때마다 지속적인 유지보수가 필요합니다. 전체 LangChain 파이프라인(pipelines)은 과해 보이며, 단순히 구조화된 분류 단계만 필요한 상황에서 상당한 프롬프트 엔지니어링(prompt engineering) 오버헤드를 발생시킵니다.

결과적으로, 기존 통합 방식이 너무 취약하거나 너무 무겁기 때문에 지원 팀은 이메일을 티켓 시스템으로 수동으로 끌어다 놓습니다. 여러분에게 실제로 필요한 것은 이메일을 읽고, 유형과 우선순위에 대해 판단을 내리며, 시간이 흐르며 발생하는 모든 새로운 티켓 카테고리에 대해 커스텀 규칙을 요구하지 않고도 구조화된 조치를 취할 수 있는 가벼운 에이전트(agent)입니다.

그 간극을 메우기 위해 설계된 것이 바로 pydantic-ai입니다.

접근 방식: 접착 계층(Glue Layer)으로서의 구조화된 출력 (Structured Outputs)

여기서 핵심적인 통찰은 pydantic-ai를 사용하면 LLM이 반환하기를 원하는 내용을 라이브러리 수준에서 강제하여 정확하게 정의할 수 있다는 점입니다. 모델이 응답을 올바르게 형식화하기를 기대하거나, Markdown 코드 블록에서 JSON을 파싱할 필요가 없습니다. 모델의 출력은 코드가 확인하기 전에 Pydantic 모델을 통해 검증됩니다.

이것이 특히 이메일 분류(triage)에서 중요한 이유는 다음과 같습니다. 분류는 다운스트림 시스템(downstream systems)이 이를 안정적으로 소비할 수 있을 때만 유용하기 때문입니다. Linear의 API는 특정 필드 타입을 기대합니다. Slack의 알림 로직은 날짜에 따라 "critical", "Critical", 또는 "very urgent"라고 적힐 수 있는 문자열이 아니라, 불리언(boolean)이나 열거형(enum)을 필요로 합니다. 구조화된 출력(Structured output)은 LLM이 마치 타입이 지정된 함수(typed function)처럼 동작하게 만듭니다.

아키텍처는 간단합니다:

1. FastAPI는 들어오는 이메일 데이터(백그라운드 스케줄러를 통해 IMAP으로 Gmail에서 폴링(polled)된 데이터)를 수신하는 웹훅(webhook) 엔드포인트를 노출합니다.
2. **pydantic-ai 에이전트(agent)**는 원문 이메일 텍스트를 수신하여 엄격한 출력 스키마(output schema)를 가진 LLM을 통해 실행하고, TriageResult 객체를 반환합니다.
3. TriageResult는 Linear의 GraphQL API를 통해 Linear 이슈를 생성하는 데 사용됩니다.
4. 만약 priority가 P1 또는 P2라면, 온콜(on-call) 채널로 Slack 알림이 발송됩니다.

왜 LangChain 대신 이것을 사용하나요? LangChain의 출력 파서(output parsers)도 작동하지만, 실제로 어떤 일이 일어나고 있는지 모호하게 만드는 추상화 계층을 추가합니다. 운영 환경에서 파서가 실패하면 디버깅이 고통스럽습니다. pydantic-ai는 하드웨어에 더 가깝습니다(closer to the metal). Pydantic 모델을 정의하면, 그 모델을 그대로 돌려받습니다. 실패 모드(failure modes)가 명시적이며 처리하기 쉽습니다.

왜 크론 스크립트(cron script) 대신 FastAPI를 사용하나요? 헬스 체크(health check) 엔드포인트, 비동기(async) 지원, 그리고 어떤 컨테이너 환경으로든 쉬운 배포가 가능하기 때문입니다. IMAP 폴링은 백그라운드 작업으로 실행되어 아키텍처를 깔끔하고 테스트 가능하게 유지합니다.

코드 패턴: 타입이 지정된 출력 스키마로 에이전트 정의하기

이것은 개발자들이 무엇보다 먼저 이해해야 할 부분입니다. 전체 시스템은 이 패턴이 올바르게 작동하는지에 달려 있습니다.

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

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

result_type=TriageResult가 바로 마법이 일어나는 지점입니다. pydantic-ai는 모델이 이 스키마(Schema)에 따라 검증되는 응답을 반환하도록 강제하기 위해 프롬프트 스캐폴딩(Prompt scaffolding)을 구축합니다. 만약 검증에 실패하면, 자동으로 재시도합니다 (설정 가능).

requires_immediate_alert 불리언(Boolean) 값은 의도된 설계입니다. 경고 로직을 LLM의 분류(Classification) 내부에 유지하면, 라우팅(Routing) 코드에 조건부 분기(Conditional branches)를 추가하는 대신 시스템 프롬프트(System prompt)를 통해 이를 조정할 수 있습니다. 경고 임계값(Threshold)을 더 엄격하게 하거나 완화하고 싶으신가요? 프롬프트만 업데이트하세요. 코드 변경은 필요 없습니다.

suggested_team 필드는 열거형(Enum)이 아닌 자유로운 문자열(String)입니다. 팀 이름은 조직마다 다르기 때문입니다. 라우팅하기 전에 다운스트림(Downstream)에서 느슨하게 검증합니다.

통합: 이메일 입력, Linear 출력, Slack 알림

데이터 흐름은 다음과 같습니다:

Gmail IMAP 폴링 (60초마다)
    -> 원문 이메일 추출 (제목 + 본문)
    -> FastAPI 백그라운드 작업 대기열 등록
...

Linear 통합은 그들의 GraphQL API를 사용합니다. 이슈(Issue)를 생성하는 과정은 대략 다음과 같습니다:

import httpx

LINEAR_API_URL = "https://api.linear.app/graphql"
...

주의해야 할 점 하나: OAuth2를 사용하는 Gmail IMAP은 IMAPClient 라이브러리와 토큰 갱신(Token refresh) 처리가 필요합니다. 만약 단순 비밀번호 인증(Google이 일반 계정에 대해 폐지하고 있는 방식)을 사용한다면, 일부 환경에서 인증 실패가 조용히 발생할 수 있습니다. 토큰 갱신 로직을 사후 고려 사항이 아닌, 첫날부터 구축해 두십시오.

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

이 아키텍처는 잘 정의된 분류(Triage) 시나리오에서는 잘 작동하지만, 배포하기 전에 반드시 이해해야 할 실질적인 한계점들이 있습니다.

대량 처리 시의 LLM 비용: 만약 하루에 수천 통의 이메일을 처리한다면, gpt-4o-mini조차 비용이 누적됩니다. 처리량이 매우 많은 경우에는 LLM 분류 단계에 도달하기 전에 빠른 사전 필터(키워드 매칭 또는 미세 조정된(fine-tuned) 소형 모델)를 추가하는 것이 좋습니다.

환각된 요약 (Hallucinated summaries): summary 필드는 모델이 생성하는 자유 형식의 텍스트입니다. 때때로 원래 이메일을 잘못 나타내는 요약을 생성할 수 있습니다. 만약 Linear 이슈가 시스템의 기록(system of record) 역할을 한다면 이 점은 중요합니다. 원본 이메일 본문을 이슈의 첨부 파일로 저장하는 것을 고려해 보세요.

스레드 인식 불가 (No threading awareness): 이 시스템은 각 이메일을 독립적인 것으로 취급합니다. 답장 체인(Reply chains)과 에스컬레이션(escalations)을 처리하려면 이 템플릿이 다루지 않는 추가적인 로직이 필요합니다.

더 단순한 방식을 선택해야 할 때: 만약 이메일 유형이 진정으로 안정적이라면(절대 변하지 않는 3~4개의 카테고리), 정규 표현식(regex) 매칭을 사용하는 규칙 기반(rule-based) 시스템이 더 저렴하고, 빠르며, 예측 가능할 것입니다. LLM 분류는 입력 공간이 무질서하고 계속 진화할 때 그 복잡성에 대한 가치를 증명합니다.

코드를 가져가고 여러분이 만든 것을 공유해 주세요

저는 이것을 GitHub에 오픈 소스 스캐폴드(scaffold)로 패키징해 두었습니다: https://github.com/Reactance0083/pydantic-ai-email-linear-auto-triage

이 스캐폴드는 핵심 구조를 제공합니다: pydantic-ai 에이전트 정의, FastAPI 앱 스켈레톤(skeleton), 그리고 Linear 및 Slack을 위한 스텁(stub) 통합 기능입니다.

완전한 에러 핸들링, OAuth2 Gmail 인증, 재시도 로직(retry logic), 테스트 커버리지 및 배포 문서가 포함된 전체 프로덕션 버전은 여기에서 확인할 수 있습니다: https://reactance0083.gumroad.com/l/dcror

이미 프로덕션에서 이와 유사한 시스템을 운영 중이거나, 제가 여기서 다루지 않은 예외 케이스(다국어 이메일, CRM 통합, SLA 추적 등)를 경험하셨다면, 댓글을 통해 진심으로 이야기를 듣고 싶습니다. 여기서 내린 설계 결정이 유일하게 유효한 결정은 아니며, 규모에 따라 트레이드오프(tradeoffs)는 다르게 나타날 수 있습니다.