beautifulsoups.dev
Published on

LangChain+OpenAI로 PII 자동마스킹 RAG 구축

Authors

서버 로그, CS 티켓, 계약서, 상담 녹취 전사본 같은 비정형 문서에는 이름, 전화번호, 이메일, 주소, 계좌번호 등 PII가 섞이는 경우가 흔합니다. 이런 데이터를 그대로 벡터DB에 넣거나 LLM 프롬프트로 보내면 컴플라이언스 리스크가 커지고, 한 번 유출되면 회수가 어렵습니다.

이 글은 LangChain + OpenAI 조합으로 PII 자동 마스킹을 선행한 RAG 파이프라인을 만드는 방법을 다룹니다. 핵심은 다음 두 가지입니다.

  • 인덱싱 단계: 원문에서 PII를 탐지하고 마스킹한 텍스트만 임베딩 및 저장
  • 질의/생성 단계: 검색 결과 컨텍스트에도 PII가 섞이지 않게 보장하고, 모델 출력에서도 재노출을 막는 가드레일 적용

운영 관점에서 지연과 비용도 중요합니다. 벡터 검색 튜닝은 RAG 품질과 지연을 동시에 좌우하므로, 필요하다면 HNSW 파라미터까지 손보는 접근이 유효합니다. 관련해서는 Rust+Qdrant RAG - HNSW 튜닝으로 지연 50%↓도 함께 참고하면 좋습니다.

목표 아키텍처: PII 안전한 RAG의 기준선

일반적인 RAG 흐름을 PII 관점으로 다시 그리면 다음과 같습니다.

  1. 수집(Ingest): 파일, DB, S3, 티켓 시스템 등에서 문서 로드
  2. 전처리(Preprocess): 정규화, 청크 분할, 메타데이터 부착
  3. PII 탐지(Detect): 규칙 기반 + 모델 기반 하이브리드
  4. 마스킹(Mask): 토큰 치환, 부분 마스킹, 혹은 가명화(pseudonymization)
  5. 인덱싱(Index): 마스킹된 텍스트만 임베딩하고 벡터DB에 저장
  6. 검색(Retrieve): 질의 임베딩 후 top-k 검색, 필요시 rerank
  7. 생성(Generate): 마스킹된 컨텍스트로만 답변 생성, 출력 필터링
  8. 감사(Audit): 어떤 문서가 검색되었고 어떤 마스킹이 적용되었는지 추적

여기서 가장 중요한 원칙은 원문을 LLM 입력으로 보내지 않기입니다. 원문이 필요하면 별도 보안 저장소에 두고, RAG는 마스킹된 파생 데이터로만 동작하게 만듭니다.

PII 마스킹 전략: 규칙 vs 모델, 그리고 하이브리드

PII 탐지는 크게 두 축이 있습니다.

1) 규칙 기반(정규식, 룰 엔진)

  • 장점: 빠르고 예측 가능, 비용 0에 가깝고 재현성 높음
  • 단점: 케이스가 늘수록 룰이 복잡해지고 누락이 생김

예: 전화번호, 이메일, 주민번호 형태 등은 정규식이 강합니다.

2) 모델 기반(NER, LLM)

  • 장점: 문맥 기반으로 이름, 회사명, 주소 등 변형에 강함
  • 단점: 비용과 지연, 그리고 오탐/누락에 대한 통제가 어려움

3) 하이브리드(권장)

  • 1차로 규칙 기반으로 확실한 패턴을 잡고
  • 2차로 NER 또는 LLM을 사용해 문맥형 PII를 보완
  • 마지막에 마스킹 결과를 검증하는 후처리(예: 이메일 패턴이 남아있으면 실패 처리)

운영에서는 오탐보다 누락이 더 위험한 경우가 많습니다. 따라서 “마스킹이 조금 과해도 안전”한 정책을 기본으로 두고, 검색 품질 하락이 크면 예외 정책을 좁게 추가하는 방식이 안정적입니다.

LangChain 파이프라인 설계: 인덱싱과 질의 흐름 분리

RAG는 크게 Indexing과 Query 두 파이프라인으로 분리하는 게 관리에 유리합니다.

  • Indexing: 문서를 청크로 나누고 PII를 마스킹한 뒤 임베딩 저장
  • Query: 사용자 질의에 대해 검색하고, 컨텍스트로 답변 생성

PII 안전성은 Indexing에서 대부분 결정됩니다. Query 단계에서 아무리 가드레일을 세워도, 이미 벡터DB에 원문이 들어가 있으면 사고 가능성이 남습니다.

예제 코드: PII 마스킹 후 벡터DB 인덱싱

아래 예시는 Python 기준이며, LangChain은 버전 변화가 잦으니 패키지 버전을 고정하는 것을 권장합니다.

pip install \
  langchain==0.3.19 \
  langchain-openai==0.3.6 \
  langchain-community==0.3.18 \
  qdrant-client==1.13.2

1) 간단한 PII 마스커(정규식 기반)

실무에서는 Microsoft Presidio 같은 라이브러리나 상용 DLP를 붙이는 경우가 많지만, 개념을 보여주기 위해 최소 구현을 둡니다.

import re
from dataclasses import dataclass

@dataclass
class MaskResult:
    masked_text: str
    hits: list[dict]

EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
PHONE_RE = re.compile(r"\b(01[016789]|02|0[3-6][1-5])[- ]?\d{3,4}[- ]?\d{4}\b")

def mask_pii(text: str) -> MaskResult:
    hits = []

    def _mask(pattern, label, repl):
        nonlocal text
        for m in list(pattern.finditer(text)):
            hits.append({
                "type": label,
                "start": m.start(),
                "end": m.end(),
                "value": m.group(0),
            })
        text = pattern.sub(repl, text)

    _mask(EMAIL_RE, "EMAIL", "[EMAIL]")
    _mask(PHONE_RE, "PHONE", "[PHONE]")

    return MaskResult(masked_text=text, hits=hits)

여기서 중요한 포인트는 hits를 남겨 감사 로그로 활용하는 것입니다. 운영 중 “왜 이 문서가 검색되었지” 같은 이슈를 추적할 때, 마스킹 여부와 결과를 함께 봐야 원인을 빨리 찾습니다.

2) 문서 로드, 청크 분할, 마스킹 적용

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = TextLoader("./data/support_tickets.txt", encoding="utf-8")
docs = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=120,
)
chunks = splitter.split_documents(docs)

masked_chunks = []
for d in chunks:
    r = mask_pii(d.page_content)
    d.page_content = r.masked_text
    d.metadata = {
        **d.metadata,
        "pii_types": sorted({h["type"] for h in r.hits}),
        "pii_hit_count": len(r.hits),
    }
    masked_chunks.append(d)

메타데이터에 pii_hit_count를 넣어두면, 검색 결과에 대해 “PII가 많이 포함된 문서 청크는 점수를 낮추거나 제외” 같은 정책을 후속으로 적용할 수 있습니다.

3) 임베딩과 Qdrant 저장

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Qdrant
from qdrant_client import QdrantClient

emb = OpenAIEmbeddings(model="text-embedding-3-large")

client = QdrantClient(url="http://localhost:6333")

vs = Qdrant.from_documents(
    documents=masked_chunks,
    embedding=emb,
    url="http://localhost:6333",
    collection_name="support_kb_masked",
)

운영 팁:

  • 임베딩 모델 변경은 재인덱싱을 의미합니다. 컬렉션을 버전으로 나누고 단계적으로 스위칭하세요.
  • 검색 지연이 커지면 HNSW 튜닝, payload 인덱스, 필터링 전략을 함께 보세요. 대규모에서 병목이 자주 터지는 지점입니다.

예제 코드: 검색과 답변 생성, 그리고 출력 가드레일

1) Retriever 구성

retriever = vs.as_retriever(search_kwargs={"k": 6})

여기서 k는 품질과 비용에 직결됩니다. 너무 작으면 근거가 부족하고, 너무 크면 컨텍스트가 길어져 비용과 환각 가능성이 늘어납니다.

2) 프롬프트: PII 비노출을 명시

프롬프트에 “PII를 출력하지 말라”는 문장을 넣는 것은 필요하지만, 이것만으로는 부족합니다. 그래도 최소한의 정책으로는 반드시 넣는 편이 낫습니다.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system",
     "너는 사내 지식베이스 도우미다. 제공된 컨텍스트만 근거로 답하라. "
     "개인정보(이메일, 전화번호, 주소, 계좌 등)는 어떤 경우에도 원문 그대로 출력하지 말고 "
     "반드시 [EMAIL], [PHONE] 같은 마스킹 토큰으로 유지하라."),
    ("user", "질문: {question}\n\n컨텍스트:\n{context}")
])

3) RAG 체인

from langchain_core.runnables import RunnablePassthrough

def format_docs(docs):
    return "\n\n".join(
        f"[source={d.metadata.get('source','unknown')}]\n{d.page_content}"
        for d in docs
    )

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
)

answer = rag_chain.invoke("고객이 비밀번호 재설정 링크를 못 받았다고 할 때 대응 절차 알려줘")
print(answer.content)

4) 최종 출력 필터(후처리)

프롬프트만으로는 누락이 생길 수 있으니, 마지막에 정규식으로 한 번 더 필터링하는 방식을 추천합니다.

def enforce_no_pii(text: str) -> str:
    text = EMAIL_RE.sub("[EMAIL]", text)
    text = PHONE_RE.sub("[PHONE]", text)
    return text

safe_output = enforce_no_pii(answer.content)
print(safe_output)

이 후처리는 완벽하지 않지만, 최소한 “명백한 패턴” 재노출을 한 번 더 막아줍니다. 더 강하게 하려면 NER 기반 검출을 후처리에 추가하고, 검출되면 응답 자체를 차단하거나 재생성시키는 정책을 둡니다.

운영에서 터지는 문제 6가지와 대응

1) 마스킹 누락: 변형 포맷과 문맥형 PII

  • 문제: 01012345678, 010.1234.5678, 홍 길동, 서울 강남구 같은 변형
  • 대응: 규칙 기반을 촘촘히 + NER 보완 + “마스킹 검증 단계” 추가

검증 단계 예:

  • 이메일 패턴이 남아있으면 인덱싱 실패 처리
  • 전화번호 패턴이 남아있으면 해당 청크를 폐기하거나 재마스킹

2) 과도한 마스킹으로 검색 품질 하락

  • 문제: 제품명/에러코드/호스트명까지 PII로 오탐하면 검색이 망가짐
  • 대응: 엔티티 타입별 정책 분리(예: PERSON, EMAIL은 강제 마스킹, ORG는 조건부)

3) 마스킹 토큰의 일관성 부족

  • 문제: [EMAIL], [E-MAIL], [MAIL]처럼 토큰이 흔들리면 검색과 분석이 어려움
  • 대응: 토큰 사전 고정, 타입 체계 고정, 버전 관리

4) 원문 재구성 위험(가명화의 함정)

  • 문제: 동일 인물을 항상 같은 토큰으로 치환하면 조합으로 재식별 가능
  • 대응: 보안 요구가 높으면 문서 단위 랜덤화, 혹은 완전 마스킹을 선택

5) 감사 추적 부재

  • 문제: 사고가 났을 때 “어떤 문서가 LLM에 들어갔는지” 증명 불가
  • 대응: 요청 ID 단위로 retrieved_doc_ids, pii_hit_count, model, prompt_version를 로깅

6) 지연과 비용 폭증

  • 문제: 청크가 많고 k가 크면 컨텍스트 길이와 호출 비용이 급증
  • 대응: 청크 크기 최적화, MMR, rerank 최소화, 캐시 적용

데이터 처리 파이프라인에서 병목을 진단하는 습관은 중요합니다. 예를 들어 조인/머지로 행이 폭증하는 류의 실수는 RAG 인덱싱에서도 그대로 반복됩니다(문서 중복 적재, 청크 중복 생성 등). 이런 데이터 이슈를 빠르게 진단하는 감각은 pandas merge 후 행 폭증·NaN - 키 중복 5분 진단 같은 글의 접근이 그대로 도움이 됩니다.

고급 패턴: “민감 원문”과 “검색용 파생본” 분리

보안 요건이 높은 조직에서는 다음처럼 이원화합니다.

  • 벡터DB: 마스킹된 검색용 텍스트만 저장
  • 원문 저장소: 접근 통제된 별도 스토리지에 암호화 저장
  • 앱 서버: 검색 결과의 doc_id만 받아서, 권한이 있는 사용자에게만 원문 일부를 보여줌

즉, LLM은 끝까지 마스킹된 컨텍스트만 보게 하고, 사람이 UI에서 원문을 확인해야 하는 경우에만 권한 체크 후 보여주는 구조입니다.

배포 체크리스트: 실무에서 필요한 최소 안전장치

  • 인덱싱 단계에서 PII 검출 실패 시 fail closed 정책(저장하지 않음)
  • 문서 중복 적재 방지(해시 기반 dedup)
  • 컬렉션 버전 관리(임베딩 모델 변경, 마스킹 정책 변경 시)
  • 프롬프트 버전 관리 및 감사 로그
  • 출력 후처리 필터 + 재생성/차단 정책
  • 키 관리: OpenAI API 키는 KMS/Secret Manager에 저장, 로컬 파일 하드코딩 금지

쿠버네티스 환경이라면 시크릿/네트워크 정책/아웃바운드 제어까지 같이 봐야 합니다. 특히 장애가 나면 원인이 애플리케이션이 아니라 클러스터 네트워킹인 경우도 많아서, 기본적인 클러스터 트러블슈팅 글을 옆에 두는 것이 좋습니다. 예: EKS kubelet NotReady - CNI plugin not initialized 해결

마무리: PII 마스킹은 RAG 품질의 적이 아니라 전제조건

PII 자동 마스킹을 붙이면 검색 품질이 떨어질 것 같지만, 실제로는 “운영 가능한 RAG”를 만들기 위한 전제조건에 가깝습니다. 중요한 것은 마스킹을 단순한 문자열 치환으로 끝내지 않고,

  • 인덱싱 전 강제 적용
  • 누락 검증
  • 메타데이터로 추적 가능하게 기록
  • 생성 결과에도 후처리 가드레일 적용

까지 한 묶음으로 설계하는 것입니다.

다음 단계로는 (1) Presidio 같은 NER 기반 엔진 결합, (2) 문서 타입별 마스킹 정책 분리, (3) 검색 성능을 위한 벡터 인덱스 튜닝, (4) 평가셋을 만들어 누락률과 오탐률을 수치화하는 쪽을 추천합니다.

Discuss on Twitter • View on GitHub