TuBrief
구독 채널
비디오
커뮤니티

데모에서 멀쩡하던 AI 에이전트가 고객 DB를 만나면 터지는 이유

TuBrief 편집팀
2026년 7월 23일
0
컴퓨터/소프트웨어

원본 영상을 바탕으로 AI의 도움을 받아 작성했습니다. 원본 영상이 기준입니다.

한국어العربية中文FrançaisDeutschРусский日本語EnglishBahasa Indonesia

관련 영상

Ship 26 NYC - 문의 티켓의 90% 자동화: Vercel의 지원 에이전트를 구축한 방법18:15

Ship 26 NYC - 문의 티켓의 90% 자동화: Vercel의 지원 에이전트를 구축한 방법

Vercel

커뮤니티의 다른 글

사내 시스템에 llm api 붙일 때 마주하는 현실적인 한계와 대응법

2026년 9월 13일

레거시 백엔드에 GPT-6 Astra 붙일 때 예산 승인과 보안 통과를 먼저 끝내는 법이 있습니다

2026년 9월 13일

에이전트끼리 대화하다 6천만 원 청구서가 나오는 이유

2026년 9월 13일

사내 RAG 벡터 검색에 Okta 권한 필터를 직접 거는 방법

2026년 9월 13일

브라우저 에이전트에게 내 구글 계정을 통째로 넘기면 안 되는 이유

2026년 9월 12일

Apple Won the AI Race

2026년 9월 12일

댓글 (0)

Log in to leave a comment

아직 작성된 글이 없습니다

© 2026 . All rights reserved.

TuBrief
구독 채널
비디오
커뮤니티
로그인

데모에서 멀쩡하던 AI 에이전트가 고객 DB를 만나면 터지는 이유

데모 환경에서 매끄럽게 질문에 답하던 AI 에이전트를 실제 고객 DB와 붙이는 순간 상황이 일그러집니다. 에이전트가 엉뚱한 데이터를 수정하고, 개인정보를 외부 LLM으로 유출하며, 잘못된 환불 정책을 확답하고, 한 달 만에 API 비용 수천 달러를 청구받는 일은 생각보다 흔하게 일어납니다. 프롬프트에 "절대 환각을 일으키지 마"라고 써봤자 아무 소용 없습니다. 인프라 수준에서 결정론적으로 목을 죄어야 합니다.

1. DB 접근 권한 분리와 PII 마스킹

AI 에이전트에게 DB 접근 권한을 줄 때 가장 무서운 건 프롬프트 인젝션입니다. 악의적인 사용자가 프롬프트로 에이전트를 속여 삭제 쿼리를 실행하도록 유도할 수 있습니다. 자연어 프롬프트 지시문에 의존하는 보안은 무조건 뚫립니다. 물리적인 격리가 필요합니다.

Read Replica 기반 물리적 권한 격리

Model Context Protocol(MCP) 환경에서 데이터를 조회하는 모든 툴은 메인 DB가 아닌 읽기 전용 복제본(Read Replica)에만 연결해야 합니다. PostgreSQL 기준으로 SELECT 권한만 부여한 전용 DB 계정을 따로 생성하세요. 에이전트가 무슨 짓을 해서 INSERT나 UPDATE 쿼리를 날려도 RDBMS 엔진 수준에서 예외를 던지며 트랜잭션을 튕겨냅니다.

PII 마스킹 미들웨어 구축

고객 데이터가 외부 LLM API로 넘어갈 때 이메일이나 전화번호 같은 개인식별정보(PII)가 생으로 넘어가면 법적 문제가 생깁니다. MCP 프록시 앞단에 정규식과 개체명 인식(NER) 모델을 결합한 양방향 PII 마스킹 미들웨어를 배치해야 합니다.

  1. 정규식 패턴으로 이메일, 전화번호, API 키 같은 정형 데이터를 1차로 솎아냅니다.
  2. Microsoft Presidio나 spaCy 모델로 문맥 속에 숨은 고객 이름과 주소를 감지합니다.
  3. 감지된 정보를 [EMAIL_MASKED], [PHONE_MASKED] 형태의 토큰으로 변환한 뒤 LLM으로 보냅니다.
import re
from typing import Dict, Any
from presidio_analyzer import AnalyzerEngine, PatternRecognizer
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

class PIIMaskingMiddleware:
    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()
        self._register_custom_patterns()

    def _register_custom_patterns(self):
        api_key_pattern = PatternRecognizer(
            supported_entity="CUSTOM_API_KEY",
            patterns=[re.compile(r'(?i)(api[_-]?key|secret[_-]?key|bearer)\s*[:=]\s*["\']?([a-zA-Z0-9_\-]{20,})["\']?')]
        )
        self.analyzer.registry.add_recognizer(api_key_pattern)

    def mask_text(self, text: str) -> str:
        if not text or not text.strip():
            return text

        results = self.analyzer.analyze(text=text, language="en")
        anonymized_result = self.anonymizer.anonymize(
            text=text,
            analyzer_results=results,
            operators={
                "EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "[EMAIL_MASKED]"}),
                "PHONE_NUMBER": OperatorConfig("replace", {"new_value": "[PHONE_MASKED]"}),
                "PERSON": OperatorConfig("replace", {"new_value": "[NAME_MASKED]"}),
                "CUSTOM_API_KEY": OperatorConfig("replace", {"new_value": "[API_KEY_MASKED]"}),
                "DEFAULT": OperatorConfig("replace", {"new_value": "[PII_MASKED]"}),
            }
        )
        return anonymized_result.text

    def process_mcp_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        processed = {}
        for key, value in payload.items():
            if isinstance(value, str):
                processed[key] = self.mask_text(value)
            elif isinstance(value, dict):
                processed[key] = self.process_mcp_payload(value)
            elif isinstance(value, list):
                processed[key] = [
                    self.process_mcp_payload(v) if isinstance(v, dict) 
                    else self.mask_text(v) if isinstance(v, str) else v 
                    for v in value
                ]
            else:
                processed[key] = value
        return processed

Human-in-the-Loop 승인 통제

요금제 변경이나 구독 취소처럼 데이터를 실제로 바꾸는 연산은 에이전트가 혼자 처리하게 두면 안 됩니다. Vercel AI SDK를 쓴다면 툴 정의 시 needsApproval: true 옵션을 줘서 UI 단에서 사람이 직접 승인 버튼을 누르도록 만드세요. 승인 대기가 길어지는 작업은 Vercel Workflow SDK의 use workflow와 createWebhook()을 결합해 세션을 일시 중단했다가 승인 이벤트가 수신될 때 재개하는 구조로 풀어야 합니다.

영역 점검 항목 통제 기준
DB 접근 Read Replica 고정 SELECT 전용 계정 연결, CUD 연산 발생 시 DB 단에서 예외 차단
네트워크 Private Subnet MCP 프록시와 DB 통신을 VPC 내부로 한정
PII 보안 양방향 마스킹 정규식 및 Presidio NER 모델로 외부 전송 전 민감 정보 대치
상태 변경 HITL 승인 CUD 툴 호출 시 UI 승인 또는 Workflow Webhook 대기

2. 환각 답변으로 인한 오안내 차단

RAG가 연관성 낮은 문서를 끌어오면 에이전트는 거짓말을 시작합니다. 검색, 생성, 검증 각 단계마다 필터를 걸어야 합니다.

RAG 검색 임계값과 Fallback

벡터 DB를 조회할 때 스코어 기준을 빡빡하게 세우세요. OpenAI text-embedding-3-small 기준 코사인 유사도 0.78 이하인 문서는 그냥 버립니다. 가장 잘 나온 문서의 스코어도 임계값을 못 넘기면 LLM에게 프롬프트를 넘기지 마세요. 곧바로 "확인된 내부 문서가 없습니다. 상담원에게 연결해 드릴까요?"라는 고정 응답을 출력하도록 예외 처리를 해야 합니다.

Temperature 0.0과 Pydantic 검증기

가격 정책이나 계약 조건에 대해 대답할 때는 temperature 파라미터를 0.0으로 고정하세요. LLM이 헛소리를 하지 못하게 억제해야 합니다. 그리고 출력단에는 Pydantic 기반 규칙 검증기를 붙여서 비즈니스 제약조건을 검사합니다.

  1. BaseModel을 상속받아 응답 구조를 정의합니다.
  2. @field_validator로 답변 안에 들어간 할인율이 최대 허용치(예: 15%)를 넘는지, "무조건 환불" 같은 금지 구문이 들어갔는지 검사합니다.
  3. 검증 실패 시 ValidationError 메시지를 힌트로 첨부해 LLM에 최대 3회 재질의합니다.
from pydantic import BaseModel, Field, field_validator
from typing import List, Optional

class CSResponseValidator(BaseModel):
    answer_text: str = Field(description="고객에게 전송될 최종 답변 텍스트")
    referenced_doc_ids: List[str] = Field(description="답변에 참조된 내부 문서 ID")
    offered_discount_rate: Optional[float] = Field(default=0.0, description="답변 내 언급된 할인율")

    @field_validator("offered_discount_rate")
    @classmethod
    def validate_discount_cap(cls, value: float) -> float:
        MAX_DISCOUNT = 0.15
        if value > MAX_DISCOUNT:
            raise ValueError(f"허용된 최대 할인율({MAX_DISCOUNT * 100}%)을 초과하여 안내할 수 없습니다: {value * 100}%")
        return value

    @field_validator("answer_text")
    @classmethod
    def validate_prohibited_terms(cls, text: str) -> str:
        prohibited_phrases = ["무조건 환불", "100% 데이터 복구 보장", "평생 무료 전환"]
        for phrase in prohibited_phrases:
            if phrase in text:
                raise ValueError(f"비즈니스 금지 보장 구문이 포함되었습니다: '{phrase}'")
        return text

깨진 링크 검증과 샘플링 검수

답변을 스트리밍하기 직전, 마크다운 링크 [text](url) 패턴을 정규식으로 뽑아낸 뒤 비동기 HEAD 요청을 날리세요. 1.5초 내에 200 OK 응답이 안 오면 죽은 링크로 판단하고 URL을 메인 도움말 페이지로 자동 치환합니다.

전체 대화 중 무작위 5%와 함께 부정 피드백이 들어온 대화, RAG 스코어 경계선(0.78 ~ 0.82)에 걸친 대화를 100% 수집해서 주기적으로 오안내율을 집계해야 합니다.

오안내율 (%)=(환각 및 오안내 답변 수전체 샘플링 대화 수)×100\text{오안내율 (\%)} = \left( \frac{\text{환각 및 오안내 답변 수}}{\text{전체 샘플링 대화 수}} \right) \times 100오안내율 (%)=(전체 샘플링 대화 수환각 및 오안내 답변 수​)×100

3. 토큰 비용을 줄이는 시맨틱 캐싱

B2B SaaS CS 문의는 패턴이 뻔합니다. 질문의 30~50%는 들어왔던 질문이 또 들어오는 구조입니다. 단순히 문자가 똑같은지 비교하는 캐싱은 히트율이 10%도 안 나옵니다. 의미가 같은지 판단하는 시맨틱 캐시를 써야 합니다.

RedisVL 기반 시맨틱 캐시

RedisVL을 활용해 유입된 질문의 임베딩과 기존 질문 DB 사이의 코사인 유사도를 계산하세요. 코사인 유사도가 0.95 이상이면 LLM API를 아예 호출하지 않고 Redis에 저장된 기존 대답을 20ms 안에 바로 뱉어냅니다. 멀티테넌트 환경이라면 Redis FT.SEARCH 쿼리에 tenant_id 필터를 걸어 다른 고객사의 데이터가 섞여 나가는 일을 막아야 합니다.

from redisvl.extensions.llmcache import SemanticCache
from redisvl.query.filter import Tag

class CSCSemanticCacheManager:
    def __init__(self, redis_url: str):
        self.cache = SemanticCache(
            name="cs_semantic_cache",
            redis_url=redis_url,
            distance_threshold=0.05,
            ttl=604800
        )

    def get_cached_response(self, user_query: str, tenant_id: str) -> dict:
        tenant_filter = Tag("tenant_id") == tenant_id
        results = self.cache.check(
            prompt=user_query,
            filter_expression=tenant_filter,
            return_fields=["prompt", "response", "metadata"]
        )

        if results:
            return {
                "hit": True,
                "response": results[0]["response"],
                "latency_ms": 15
            }
        return {"hit": False, "response": None}

    def store_response(self, user_query: str, llm_response: str, tenant_id: str):
        self.cache.store(
            prompt=user_query,
            response=llm_response,
            metadata={"tenant_id": tenant_id}
        )

컨텍스트 압축과 호출 제한

대화가 길어지면 토큰 수가 비대해집니다. 최근 3회 대화(6개 메시지)만 원본 형태로 유지하고, 그 이전 대화는 "요청사항: 요금제 변경, 상태: 완료" 형태로 핵심만 요약해 프롬프트에 집어넣으세요. 이렇게만 해도 프롬프트 토큰을 절반 이상 줄일 수 있습니다. API 게이트웨이 앞단에는 Redis 기반 Sliding Window Rate Limiter를 걸어 IP나 계정당 분당 15회 이상 요청이 들어오면 차단해야 합니다.

월 10,000건의 문의가 발생하는 환경에서 시맨틱 캐시를 도입했을 때의 체감 효과는 명확합니다.

지표 도입 전 도입 후 (유사도 0.95)
평균 캐시 히트율 8% (단순 문자열) 48% (의미 기반)
LLM API 월 비용 $3,000 $1,620
평균 응답 속도 1,200 ms 320 ms

4. 사람 상담사 무중단 이관 파이프라인

에이전트가 모든 걸 해결할 수는 없습니다. 복잡한 버그 리포트나 화가 난 고객의 문의는 억지로 쥐고 있지 말고 즉시 사람에게 넘겨야 합니다.

이관 스코어링 산식

이관 여부는 키워드, 감정 점수, 대화 길이를 더해서 결정합니다.

이관 스코어=S키워드+S감정+S지속시간\text{이관 스코어} = S_{\text{키워드}} + S_{\text{감정}} + S_{\text{지속시간}}이관 스코어=S키워드​+S감정​+S지속시간​
  • 키워드(S키워드S_{\text{키워드}}S키워드​): "환불", "해지", "상담원" 포함 시 +0.4
  • 감정 점수(S감정S_{\text{감정}}S감정​): VADER 분석 결과 부정 점수가 0.75 이상이면 +0.35
  • 지속 시간(S지속시간S_{\text{지속시간}}S지속시간​): 10턴 이상 미해결 시 +0.3

누적 스코어가 0.4를 넘거나 MCP 툴 호출이 3회 연속 실패하면 즉시 제어권을 멈추고 상담원 시스템으로 넘깁니다.

Helpdesk 연동 페이로드

Zendesk나 Intercom으로 맥락을 넘길 때는 에이전트가 어디까지 진행하다 실패했는지 기록을 한꺼번에 보내야 합니다. 상담사가 고객에게 "어떤 문제가 있으신가요?"라고 다시 물어보는 순간 사용자 경험은 망가집니다.

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "AgentHumanHandoffPayload",
  "type": "object",
  "properties": {
    "event_id": { "type": "string", "format": "uuid" },
    "timestamp": { "type": "string", "format": "date-time" },
    "session_id": { "type": "string" },
    "customer_info": {
      "type": "object",
      "properties": {
        "user_id": { "type": "string" },
        "email": { "type": "string" },
        "plan_tier": { "type": "string" }
      },
      "required": ["user_id", "email"]
    },
    "handoff_reason": {
      "type": "string",
      "enum": ["CRITICAL_KEYWORD", "HIGH_NEGATIVE_SENTIMENT", "TOOL_EXECUTION_FAILURE", "VALIDATION_LOOP_EXCEEDED"]
    },
    "conversation_summary": {
      "type": "object",
      "properties": {
        "issue_category": { "type": "string" },
        "key_entities": { "type": "object" },
        "condensed_history": { "type": "string" }
      },
      "required": ["issue_category", "condensed_history"]
    },
    "technical_trace": {
      "type": "object",
      "properties": {
        "failed_tool_name": { "type": "string" },
        "error_message": { "type": "string" },
        "retry_count": { "type": "integer" }
      }
    }
  },
  "required": ["event_id", "timestamp", "session_id", "customer_info", "handoff_reason", "conversation_summary"]
}