얼마 전 저는 GitHub 이슈를 자동으로 분류(triage)해야 하는 클라이언트 프로젝트를 진행하고 있었습니다. 팀에는 버그, 기능 요청, 질문, 무의미한 노이즈 등 수백 개의 오픈 티켓이 쌓여 있었고, 사람이 직접 확인하기 전에 우선순위에 따라 그룹화해 주는 대시보드를 원했습니다.

저의 첫 번째 본능은 당연하게도 스크립트를 작성하는 것이었습니다. GitHub API를 호출하고, 각 이슈를 Claude에 보내고, 결과를 테이블에 기록하는 방식이었죠. 두 시간 만에 끝냈습니다. 처음에는 아주 잘 작동했습니다. 하지만 두 번째 실행에서 GitHub의 속도 제한(rate-limit)에 걸리고, 스키마(schema)가 변경되고, 두 번째 리포지토리가 추가되면서, 저는 갑자기 거의 동일한 기능을 수행하는 세 개의 별도 스크립트를 유지보수해야 하는 상황에 처했습니다.

그래서 저는 Airbyte가 동기화(sync)를 담당하고, 작은 Python enrichment layer가 AI 분류를 수행하도록 시스템을 재구축했습니다. 이 조합은 훨씬 더 안정적으로 유지되었으며, 그 설정 방법은 다음과 같습니다.

실행 가능한 코드: github.com/Mozi1185/airbyte-claude-enrichment

구축하게 될 내용:

• Airbyte 연결: GitHub 이슈 → Postgres (증분 동기화 (incremental sync))
• Python enricher: 처리되지 않은 행을 읽고, claude-haiku-4-5로 분류한 뒤, 구조화된 결과를 다시 기록
• 실제로 사용할 수 있는 SQL 쿼리

소요 시간: 약 45분

사전 요구 사항: Python 3.10 이상, Docker (로컬 Postgres용), Airbyte Cloud 계정 (무료 티어 가능), Anthropic API 키, GitHub 개인 액세스 토큰 (personal access token)

왜 그냥 스크립트 하나만 작성하지 않는가

하나의 스크립트에서 GitHub API와 Claude를 모두 호출할 수도 있습니다. 대부분의 튜토리얼이 정확히 그렇게 보여줍니다. 문제는 처음 작동한 이후에 발생하는 모든 일들입니다.

GitHub가 속도 제한(rate-limit)을 겁니다. Claude가 파서(parser)가 예상하지 못한 값을 반환합니다. 두 번째 리포지토리를 추가합니다. 동료가 Jira를 추가하고 싶어 합니다. 이제 당신은 원래 만들려고 했던 실제 기능 대신, 이를 연결하기 위한 글루 코드(glue code)를 유지보수하고 있게 됩니다.

작업을 깔끔하게 분리하기: Airbyte는 커넥터 유지보수(connector maintenance), 인증 갱신(auth refresh), 페이지네이션(pagination), 증분 추적(incremental tracking), 스키마 진화(schema evolution)를 처리합니다. 이는 올바르게 구현하기가 정말 까다로운 작업이며, 저는 이를 직접 하고 싶지 않습니다. 강화(enrichment) 레이어는 상태가 없고(stateless) 이식 가능(portable)하게 유지됩니다. 즉, cron으로 실행하거나, 웹훅(webhook)을 통해 트리거하거나, 나중에 Airflow에 넣을 수 있습니다. 강화에 실패하더라도 동기화(sync)를 차단하지 않습니다. 동기화 실패가 강화된 출력물을 손상시키지도 않습니다. 클라이언트가 Zendesk를 추가하면, 저는 Airbyte 연결을 추가하고 동일한 강화 도구(enricher)를 다른 테이블로 지정하기만 하면 됩니다.

1단계: Postgres 구축하기

이미 Postgres 인스턴스가 있다면 이 단계는 건너뛰세요. 로컬 개발을 위한 방법은 다음과 같습니다:

docker run -d \
  --name enrichment-db \
  -e POSTGRES_USER=dev \
...

시작되었는지 확인합니다:

docker exec -it enrichment-db psql -U dev -d enrichment -c "SELECT version();"

2단계: Airbyte를 통해 GitHub를 Postgres에 연결하기

Airbyte Cloud에 로그인합니다. 새로운 연결(connection)을 생성합니다.

소스(Source): GitHub

1. Sources → + New source → GitHub로 이동합니다.
2. 저장소(repository)를 입력합니다 — 예: airbytehq/airbyte 또는 본인의 저장소
3. GitHub 개인 액세스 토큰(personal access token, repo 범위 — github.com/settings/tokens에서 생성)을 붙여넣습니다.
4. Set up source를 클릭합니다.

대상(Destination): Postgres

1. Destinations → + New destination → Postgres로 이동합니다.
2. 연결 세부 정보를 입력합니다:
   • Host: localhost
   • Port: 5432
   • Database: enrichment
   • Username: dev
   • Password: localdev
   • Default Schema: github_raw
3. Set up destination을 클릭합니다.

주의 사항: Postgres를 로컬에서 실행 중이라면, Airbyte Cloud는 localhost에 도달할 수 없습니다. 두 서비스가 서로 다른 네트워크에 있기 때문입니다. 저도 이를 깨닫기 전까지 오후 시간을 통째로 허비했습니다. 해결책은 ngrok을 사용하는 것입니다: ngrok tcp 5432. 터널 호스트와 포트(예: 0.tcp.ngrok.io:12345)를 복사하여 대상(destination) 설정에서 localhost 대신 사용하세요.

연결 설정(Connection settings)

1. Streams 항목에서 issues만 선택하세요 — 나머지 항목은 모두 체크 해제합니다.
2. 동기화 모드 (Sync mode): Incremental | Append + Deduped
3. 일정 (Schedule): 6시간마다 (또는 현재는 수동으로 실행)
4. Set up connection을 클릭한 다음, Sync now를 클릭합니다.

첫 번째 동기화가 완료되면 Postgres에 github_raw.issues 테이블이 생성되고 데이터가 채워집니다. 다음 쿼리로 확인하세요:

SELECT id, title, state, created_at
FROM github_raw.issues
LIMIT 5;

body 컬럼은 Claude가 읽게 될 데이터입니다. 일부 행이 비어 있다면 이는 정상입니다. GitHub에서는 본문(body)이 없는 이슈 생성을 허용하기 때문입니다.

Step 3: Enrichment 테이블 생성하기

CREATE TABLE IF NOT EXISTS github_raw.issues_enriched (
    issue_id         BIGINT PRIMARY KEY,
    category         TEXT,
...

issue_id에 PRIMARY KEY를 설정하면 재실행 시에도 안전합니다. 행을 중복 생성하는 대신 업서트 (Upsert)를 수행하기 때문입니다.

Step 4: 의존성 설치하기

pip install anthropic psycopg2-binary python-dotenv

.env 파일을 생성합니다:

ANTHROPIC_API_KEY=sk-ant-...
DB_HOST=localhost
DB_PORT=5432
...

Step 5: Enricher 작성하기

# enricher.py
import os
import json
...

실행합니다:

python enricher.py

예상 출력 결과:

Found 47 unenriched issues
  #1823 → bug / high
  #1801 → feature_request / medium
...

Step 6: 결과 쿼리하기

우선순위가 높은 버그를 최신순으로 조회:

SELECT
    i.id,
    i.title,
...

카테고리별 전체 상세 내역 조회:

SELECT
    e.category,
    e.priority,
...

Step 7: 지속적인 실행 유지하기

Enricher는 상태를 저장하지 않는 (Stateless) 구조이므로 어떤 스케줄러에서도 실행할 수 있습니다. 가장 간단한 방법은 다음과 같습니다:

# Cron: 각 Airbyte 동기화(매시 0분 실행) 15분 후에 enrichment 실행
15 */6 * * * cd /path/to/project && python enricher.py >> enricher.log 2>&1

더 긴밀한 결합을 원한다면, Airbyte의 각 연결 설정에는 Webhook Notifications 기능이 있습니다. 웹훅을 run_enrichment_batch()를 호출하는 엔드포인트로 지정하면, 동기화가 완료되는 즉시 Enricher가 실행됩니다. 지연 시간이나 폴링 (Polling)이 필요 없습니다.

Tool use를 통한 파싱 에러 제거

위의 Enricher는 원문 텍스트(raw text)에서 Claude의 JSON을 파싱합니다. 대부분의 경우 잘 작동합니다. 하지만 Claude가 가끔 출력물을 마크다운 코드 펜스(markdown code fence)로 감싸거나, JSON 앞에 설명 문장을 추가하는 경우가 있으며, 이로 인해 json.loads에서 오류가 발생합니다. 루프 내의 json.JSONDecodeError 예외 처리는 해당 행을 건너뜀으로써 이를 처리합니다. 이는 괜찮은 방법이지만, 행을 건너뛴다는 것은 데이터 보강(enrichment) 과정에 공백이 생긴다는 것을 의미합니다.

더 깔끔한 해결책은 Claude의 도구 사용 (Tool use) API를 사용하는 것입니다. 프롬프트에서 JSON을 요청하는 대신, 정확한 스키마(schema)를 가진 도구를 정의하고 Claude가 이를 호출하도록 강제합니다. 응답은 이미 구조화된 상태로 돌아오므로 별도의 파싱 단계가 필요 없습니다.

CLASSIFY_TOOL = {
    "name": "classify_issue",
    "description": "Classify a GitHub issue by category, priority, and summary.",
...

run_enrichment_batch에서 classify_issue를 classify_issue_structured로 교체하고 json.JSONDecodeError 예외 처리를 제거하세요. 이제 더 이상 해당 오류가 발생하지 않을 것입니다.

비용 분석

claude-haiku-4-5는 입력 토큰 100만 개당 $0.80, 출력 토큰 100만 개당 $4입니다. 각 이슈는 대략 300~600개의 입력 토큰과 약 80개의 출력 토큰을 사용합니다.

10,000개의 이슈 기준:

이는 초기 백필 (backfill) 비용입니다. 그 이후에는 Airbyte가 마지막 실행 이후의 새로운 행만 동기화하므로, 지속적인 보강 비용은 전체 이력이 아닌 새로운 데이터 양에 따라 확장됩니다.

입력 비용을 더 절감하고 싶다면 프롬프트 캐싱 (prompt caching)을 사용하세요. 시스템 프롬프트는 모든 호출에서 동일하므로, 이를 캐싱하면 반복 실행 시 입력 비용을 약 90%까지 낮출 수 있습니다.

message = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=256,
...

대규모 리포지토리 (repo)에서 이 작업을 수행한다면 적용할 가치가 충분합니다.

소스 교체

enrichment (데이터 보강) 코드는 GitHub에 대해 전혀 알지 못합니다. 이 코드는 Airbyte가 생성한 테이블로부터 데이터를 읽습니다. Jira로 전환하려면, Jira를 소스(source)로 하고 동일한 Postgres를 목적지(destination)로 하는 새로운 Airbyte 커넥션(connection)을 추가하기만 하면 됩니다. 그러면 Airbyte가 새로운 스키마(schema)와 테이블(table)을 생성합니다. 해당 테이블을 가리키는 get_unenriched_jira_issues 함수를 작성하세요. classify_issue 함수는 그대로 재사용할 수 있습니다.

Airbyte에는 Zendesk, HubSpot, Salesforce, Notion, Linear 등 300개 이상의 커넥터(connectors)가 있습니다. 이 enrichment (데이터 보강) 패턴은 이 모든 곳에 적용됩니다. 이것이 바로 동기화(sync)와 enrichment (데이터 보강)를 분리하여 유지하는 핵심 이유입니다.

결과적으로 당신이 얻게 되는 것은 다음과 같습니다: 정해진 일정에 따라 Airbyte가 데이터를 채워 넣는 Postgres 테이블, 매 동기화 이후 각 새로운 행(row)을 분류하는 enrichment (데이터 보강) 레이어(layer), 그리고 팀에서 사용하는 어떤 SQL 도구로도 쿼리(query)할 수 있는 구조화된 출력값입니다. 동기화와 enrichment (데이터 보강)는 독립적으로 유지됩니다. 즉, 어느 한쪽이 다른 쪽을 망가뜨릴 수 없으며, 새로운 소스(source)를 추가하는 것은 코드 변경이 아닌 Airbyte 설정 변경만으로 가능합니다.

Mozi Mohidien은 Claude API, Airbyte, n8n, Telegram을 활용하여 AI 자동화 시스템을 구축합니다. dev.to/mozimohidien에서 글을 작성하고 있습니다.