9가지 Claude AI 기능을 실제 운영 중인 SaaS에 구축한 방법

9가지 AI 기능. 하나의 운영 중인 SaaS. 사용자에게 보여지는 환각 (Hallucination) 데이터 제로. 마지막 부분은 들리는 것보다 훨씬 어렵습니다. 이것은 제가 GDPR을 준수하는 시민 관리 플랫폼인 CitizenApp에 9가지 Claude Haiku 기능을 어떻게 설계, 구축 및 출시했는지에 대한 전체 기술 분석입니다. 아키텍처 (Architecture), 보안 결정, 방어해야 했던 실패 모드 (Failure modes), 그리고 제가 다르게 했을 일들에 대해 다루겠습니다.

Sonnet이나 Opus가 아닌 왜 Haiku인가
CitizenApp의 모든 AI 기능은 claude-haiku-4-5를 사용합니다. 그 이유는 의도적이었습니다: 비용 (Cost). 이것은 멀티 테넌트 (Multi-tenant) SaaS입니다. 각 AI 호출은 테넌트의 크레딧 잔액에서 차감됩니다. Haiku는 Sonnet보다 토큰당 비용이 현저히 저렴합니다. 이는 제가 수익성을 해치지 않으면서도 무료 (Free) 티어에서 의미 있는 AI 기능을 제공할 수 있음을 의미합니다. 속도 (Speed). 대부분의 기능은 대화형입니다 — 사용자가 기다리고 있습니다. 제가 보내는 프롬프트 (Prompt) 크기에 대해 Haiku의 지연 시간 (Latency)은 보통 2초 미만입니다. Sonnet은 그보다 두 배 또는 세 배 더 걸릴 것입니다. 충분한 능력 (Sufficient capability). SQL 생성, 데이터 요약 (Data summarisation), 중복 레코드 비교와 같은 구조화된 작업에 대해 Haiku는 충분히 훌륭합니다. 저는 Opus 수준의 추론 (Reasoning)이 필요하지 않습니다. 저에게 필요한 것은 빠르고, 저렴하며, 신뢰할 수 있는 JSON 출력입니다. 일관된 타임아웃 예산 (Timeout budget). 모든 AI 호출에는 20초의 타임아웃이 설정되어 있습니다. Haiku는 거의 항상 5초 이내에 완료됩니다. Render의 무료 티어에서 실행할 때는 이 점이 중요합니다.

크레딧 시스템 (The Credit System)
기능 자체를 다루기 전에: CitizenApp의 모든 AI 호출에는 크레딧이 소모됩니다. 이는 프론트엔드 (Frontend)가 아닌 API 레이어 (Layer)에서 강제됩니다.

async def deduct_credits ( db : AsyncSession , tenant_id : int , amount : int = 1 ) -> None :
tenant = await db . get ( TenantModel , tenant_id )
if tenant . ai_credits < amount :
raise HTTPException ( status_code = 402 , detail = " Insufficient AI credits " )
tenant . ai_credits -= amount
await db . commit ()

단순합니다. 크레딧이 없으면 AI 호출도 없습니다. 402 Payment Required 응답은 프론트엔드에 업그레이드 프롬프트를 표시하도록 알려줍니다. 크레딧은 Claude API 호출 전에 차감되며, 호출이 실패하면 다시 환불합니다.

이는 사용자가 잘못된 요청으로 인해 크레딧을 소진하는 것을 방지합니다. 통계 쿼리(Stats queries)는 1 크레딧이 소모되며, 실시간 DB 쿼리(Live DB queries)는 2 크레딧이 소모됩니다. 가격 책정은 실제 Claude 토큰 소비량을 반영합니다.

기능 1: CSV 컬럼 매핑 (CSV Column Mapping)
문제점: 사용자들이 스프레드시트에서 시민 데이터를 가져옵니다. 컬럼 헤더(Column headers)가 일관되지 않습니다 — TC No, tc_no, Kimlik No, ID Number — 이들은 모두 같은 의미를 가집니다. 사용자에게 모든 가져오기 항목을 수동으로 매핑하도록 요청하는 것은 UX(사용자 경험) 측면의 실패입니다.

구현 방식:

async def ai_map_columns ( headers : list [ str ]) -> dict :
    prompt = f """
    You are mapping CSV column headers to a fixed schema.
    Schema fields: tc_no, ad_soyad, dogum_tarihi, cinsiyet
    Headers: { headers }
    Return JSON only: {{ " tc_no " : " header or null " , " ad_soyad " : " header or null " , " dogum_tarihi " : " header or null " , " cinsiyet " : " header or null " }}
    """
    response = await anthropic_client . messages . create (
        model = " claude-haiku-4-5 ",
        max_tokens = 200,
        messages = [{ " role " : " user ", " content " : prompt }]
    )
    return json . loads ( response . content [ 0 ]. text )

배운 점: 스키마(Schema) 설명을 짧게 유지하고 출력 형식을 엄격하게 지정하세요. 초기 버전에서는 Claude에게 "최적의 매핑을 제안하라"고 요청했는데, 이는 너무 모호하여 때때로 JSON 대신 설명 텍스트를 반환하기도 했습니다. Return JSON only:와 함께 정확한 구조를 명시함으로써 이 문제를 해결했습니다.

실패 모드 처리: 만약 json.loads에서 오류가 발생하면, 퍼지 문자열 매칭(Fuzzy string matching)으로 대체합니다. AI는 첫 번째 시도일 뿐, 유일한 방법은 아닙니다.

기능 2: 자연어 검색 (Natural Language Search)
문제점: 사용자는 50,000개 이상의 시민 기록을 필터링해야 합니다. SQL은 사용자 인터페이스가 아닙니다.

구현 방식:

ALLOWED_COLUMNS = { " ad_soyad " , " dogum_tarihi " , " cinsiyet " , " created_at " , " il " , " ilce " }

async def nl_to_sql_filter ( query : str , tenant_id : int ) -> str :
    prompt = f """
    Convert this natural language query to a PostgreSQL WHERE clause.
    Table: vatandas
    Available columns: { ' , ' . join ( ALLOWABLE_COLUMNS ) }
    tenant_id is always { tenant_id } — always include it.
    """

질의: " { query } " ' WHERE '로 시작하는 WHERE 절만 반환하세요. 설명은 생략합니다. """ response = await anthropic_client.messages.create(model="claude-haiku-4-5", max_tokens=300, messages=[{"role": "user", "content": prompt}]) clause = response.content[0].text.strip() validate_sql_clause(clause) # 위험한 패턴이 발견되면 예외 발생 return clause

가장 중요한 부분 — validate_sql_clause:
BLOCKED = [ " drop ", " delete ", " update ", " insert ", " alter ", " exec ", " -- ", " ; ", " union ", " sleep ", " pg_ " ]

def validate_sql_clause(clause: str) -> None:
lower = clause.lower()
if not lower.startswith(" where "):
raise ValueError("Invalid clause")
for pattern in BLOCKED:
if pattern in lower:
raise ValueError(f"Blocked pattern: {pattern}")

SQL과 관련된 LLM (Large Language Model) 출력은 절대 신뢰해서는 안 됩니다. 허용된 컬럼 목록은 프롬프트(prompt)에 포함되어 있습니다. 만약 Claude가 해당 목록에 없는 컬럼을 생성하면, SQLAlchemy는 실행 전 컬럼을 찾을 수 없다는 오류(column-not-found error)를 발생시킬 것입니다. 심층 방어 (Defense in depth) 전략입니다.

기능 3: 이상 탐지 (Anomaly Detection)
문제점: 잘못된 데이터가 축적됩니다. 미래의 생년월일, 1890년에 태어난 시민, 레코드의 40%에서 누락된 성별 등입니다. 사용자는 보고서를 직접 작성하지 않고도 이를 찾아낼 수 있어야 합니다.

구현 방법: 저는 이상 징후를 탐지하는 데 AI를 사용하지 않고, Python에서 다섯 가지 결정론적(deterministic) 체크를 실행합니다:

def detect_anomalies(citizens: list[dict]) -> list[dict]:
issues = []
today = date.today()
for c in citizens:
if c["dogum_tarihi"] and c["dogum_tarihi"] > today:
issues.append({"id": c["id"], "type": "future_dob", "severity": "high"})
if c["dogum_tarihi"] and c["dogum_tarihi"].year < 1900:
issues.append({"id": c["id"], "type": "implausible_dob", "severity": "medium"})
if not c["cinsiyet"]:
issues.

append({ "id": c["id"], "type": "missing_gender", "severity": "low" }) return issues

Claude의 역할은 결과를 요약하는 것입니다:

summary_prompt = f """
비기술직 관리자를 위해 다음 데이터 품질 이슈를 2~3문장으로 요약하세요:
{json.dumps(issue_counts)}
숫자를 구체적으로 명시하세요. 가장 우선순위가 높은 해결책을 먼저 제안하세요.
"""

이것이 올바른 업무 분담입니다. Python은 결정론적(deterministically)으로 이슈를 찾아냅니다. Claude는 이를 쉬운 언어로 설명합니다. AI는 "이것이 이상치(anomaly)인가"라는 이진적(binary) 판단을 절대 내리지 않습니다. AI는 단지 결과를 전달할 뿐입니다.

기능 4: AI 일일 브리핑 (AI Daily Briefing)
아키텍처 측면에서 가장 단순한 기능입니다. 매일 아침, 관리자 대시보드에는 Claude가 생성한 3~4문장의 브리핑이 표시될 수 있습니다:

stats = { "total_records": 12847, "missing_gender": 342, "missing_dob": 89, "potential_duplicates": 23, "created_last_7_days": 156 }

prompt = f """
당신은 데이터 어시스턴트입니다. 관리자를 위해 현재 시민 데이터베이스의 상태에 대한 3~4문장의 브리핑을 작성하세요. 간결하고 실행 가능해야 합니다.
통계: {json.dumps(stats)}
"""

교훈: API에 가공되지 않은 개인정보(PII, Personally Identifiable Information)를 보내지 마세요. 통계 데이터는 안전합니다. 개별 시민의 기록은 안전하지 않습니다. 저는 전송하기 전에 데이터를 집계(aggregate)합니다.

기능 5: AI 중복 병합 (AI Duplicate Merge)
이 기능은 제대로 구현하기 가장 복잡한 기능이었습니다.

문제점: 중복 탐지(duplicate detection)는 유사도 점수(similarity score)를 통해 잠재적 중복을 식별합니다. 하지만 이를 병합하려면 판단이 필요합니다. 어떤 기록의 철자가 정확할까요? 어떤 생년월일이 더 정확할 가능성이 높을까요?

구현 방식:

async def ai_merge_suggest(record_a: dict, record_b: dict) -> dict:
# Claude로 보내기 전에 주민등록번호(TC no)를 마스킹(mask) 처리합니다.
safe_a = {k: v for k, v in record_a.items() if k != "tc_no"}
safe_b = {k: v for k, v in record_b.items() if k != "tc_no"}

prompt = f """
이 두 시민 기록을 비교하여 가장 최선의 병합 버전을 제안하세요.
기록 A (id= {record_a['id']}): {json.dumps(safe_a)}
기록 B (id= {record_b['id']}): {json.

dumps ( safe_b ) } JSON 반환: {{ " keep_id " : <기본 레코드로 유지할 레코드의 id>, " merged " : {{ " ad_soyad " : " ... " , " dogum_tarihi " : " YYYY-MM-DD " , " cinsiyet " : " E/K " }}, " confidence " : " high|medium|low " , " reasoning " : " 한 문장 설명 " }} """ TC 식별 번호 (TC kimlik no)는 Claude로 전송되기 전에 항상 마스킹(masking) 처리됩니다. 이는 타협할 수 없는 원칙입니다. 이는 개인정보(PII)이며 규제 대상이고, 병합 결정에 필요하지 않기 때문입니다. 응답의 keep_id는 백엔드에 어떤 레코드를 승격시킬지 알려주며, Claude는 TC 번호를 절대 보거나 결정하지 않습니다. Claude의 출력은 자동으로 적용되지 않습니다. 이는 관리자에게 표시되는 제안이며, 관리자가 "적용(Apply)"을 클릭해야 확정됩니다. AI는 실행자가 아니라 조언자입니다.

기능 6: 자연어 보고서 (Natural Language Reports)
문제점: 관리자들은 맞춤형 데이터 내보내기가 필요합니다. "1960년에서 1970년 사이에 앙카라에서 태어난 모든 여성 시민을 보여줘"와 같은 요청은 UI가 미리 예측할 수 있는 필터가 아닙니다.
구현 방식: 전용 /reports 페이지가 자연어 쿼리(natural language query)를 수락하고, 이를 Claude로 보내 SQL 필터(SQL filter)를 생성하게 합니다 (기능 2와 동일한 패턴). 그 후 미리보기 테이블을 보여주고, Blob과 URL.createObjectURL을 사용하여 클라이언트 측에서 CSV/TSV 내보내기를 허용합니다. 백엔드는 필터링된 데이터셋을 생성합니다. 다운로드는 브라우저에서 이루어집니다. 서버 측 파일 생성, 임시 파일 정리, 저장 비용이 전혀 발생하지 않습니다.

기능 7 & 8: AI 채팅 — 통계 및 실시간 DB 쿼리 (AI Chat — Stats and Live DB Query)
플로팅 채팅 위젯(floating chat widget)은 기술적으로 가장 흥미로운 기능이었습니다. 하나의 엔드포인트(endpoint)에 두 가지 모드가 있습니다:
통계 모드 (Stats mode) — Claude는 집계 통계(aggregate statistics)만을 바탕으로 질문에 답합니다. "생년월일이 누락된 레코드는 몇 개인가요?"와 같은 질문입니다. 답변은 실시간 쿼리가 아닌 미리 계산된 통계로부터 나옵니다.
DB 쿼리 모드 (DB query mode) — 통계로 답변할 수 없는 질문의 경우, Claude는 백엔드가 직접 실행할 SQL SELECT 문을 생성합니다. DB 쿼리 모드에는 세심한 안전 공학(safety engineering)이 필요했습니다: SAFE_QUERY_RULES = """ PostgreSQL SELECT 쿼리를 생성하십시오.

엄격한 규칙: - 오직 vatandas 테이블만 쿼리할 것 - tc_no 컬럼은 절대 포함하지 말 것 - 항상 필터링할 것: WHERE tenant_id = {tenant_id} - 항상 추가할 것: LIMIT 100 - 서브쿼리(subqueries) 금지, 다른 테이블과의 조인(JOINs) 금지 - 교차 테넌트(cross-tenant) 데이터를 노출할 수 있는 집계(aggregations) 금지

def validate_generated_query ( sql : str , tenant_id : int ) -> str :
    sql_lower = sql . lower (). strip ()
    assert sql_lower . startswith ( " select " ), " SELECT 문이어야 합니다 "
    assert " tc_no " not in sql_lower , " tc_no는 차단되었습니다 "
    assert f " tenant_id = { tenant_id } " in sql_lower , " tenant_id 필터가 필요합니다 "
    assert " limit " in sql_lower , " LIMIT가 필요합니다 "

    # 제한 사항 추출 및 강제 적용
    limit_match = re . search ( r " limit\s+(\d+) " , sql_lower )
    if limit_match and int ( limit_match . group ( 1 )) > 100 :
        sql = re . sub ( r " limit\s+\d+ " , " LIMIT 100 " , sql , flags = re . IGNORECASE )
    return sql

쿼리는 데이터베이스 수준에서 5초의 문장 타임아웃(statement timeout)과 함께 실행됩니다. 만약 Claude가 느린 쿼리를 생성하면 실행이 취소됩니다.

쿼리 결과 표시: 결과는 접이식 QueryBlock에 표시되며, 데이터 테이블 상단에 SQL이 노출됩니다. 사용자는 정확히 어떤 쿼리가 실행되었는지 확인할 수 있습니다. AI가 데이터베이스를 다룰 때는 투명성(Transparency)이 중요합니다.

기능 9: AI 감사 추적 설명기 (AI Audit Trail Explainer)
감사 로그(audit log)는 citizen_created, bulk_import, login_failed, 2fa_enabled 등 30개 이상의 이벤트 유형을 기록합니다. 기술적 지식이 없는 관리자는 가공되지 않은 이벤트 스트림을 해석할 수 없습니다.

async def explain_audit_events ( events : list [ dict ]) -> str :
    # 내부 ID 및 기술적 필드 제거
    safe_events = [
        { " type " : e [ " event_type " ], " user " : e [ " user_email " ], " time " : e [ " created_at " ]}
        for e in events [: 20 ] # 최근 20개 이벤트만 포함
    ]
    prompt = f """
    이 감사 로그를 기술적 지식이 없는 관리자를 위해 쉬운 언어로 요약해 주세요.
    특이한 패턴(여러 번의 로그인 실패, 대량 작업, 권한 변경 등)이 있다면 언급해 주세요.
    최대 2~4문장으로 작성하세요.
    이벤트: { json . dumps ( safe_events ) }
    """

이벤트는 20개로 제한합니다. 전체 로그를 보내지 마세요. 토큰 예산(token budget)을 예측 가능하게 유지해야 합니다.

다르게 했을 점
1.

처음부터 구조화된 출력 (Structured Output)을 추가하세요. Anthropic의 도구 사용 (Tool use) / 구조화된 출력 (Structured output) API는 프롬프트 엔지니어링 (Prompt engineering)보다 깔끔합니다. JSON만 반환하도록 설정하세요. 저는 일부 기능에 이를 사후 적용했습니다. 처음부터 기본값으로 설정했어야 했습니다. 2. 더 공격적으로 캐싱 (Cache)하세요. 데일리 브리핑 (Daily briefing)은 페이지를 로드할 때마다 생성될 필요가 없습니다. 백그라운드 새로고침 (Background refresh) 기능이 포함된 6시간 단위의 Redis 캐시를 사용하면 AI 비용을 크게 절감할 수 있습니다. 현재는 대시보드를 열 때마다 매번 다시 생성됩니다. 3. AI 서비스 레이어 (Service layer)를 분리하세요. 9가지 기능 모두 app/services/ai.py에 있는 얇은 래퍼 (Thin wrapper)를 통해 Claude를 호출합니다. 이는 중앙 집중식 에러 처리 (Error handling) 및 로깅 (Logging)에 유용합니다. 더 일찍 추가했어야 했던 점은, 프롬프트가 버전 관리되고 엔드포인트 (Endpoint) 로직과 독립적으로 테스트 가능하도록 하는 적절한 프롬프트 레지스트리 (Prompt registry)입니다. 4. AI 호출뿐만 아니라 AI 출력 (Output)을 테스트하세요. 제 테스트 스위트 (Test suite)는 Anthropic 클라이언트를 모킹 (Mock)합니다. 이는 필요하지만 불충분합니다. 프롬프트 퇴보 (Prompt regression)를 잡아내지 못하기 때문입니다.