LangChain Middleware 미들웨어 완벽 가이드

LangChain에서 제공하는 미들웨어 기능을 활용하여 LLM 호출을 더욱 안정적이고 효율적으로 관리하는 방법을 알아봅니다. 자동 재시도, 폴백, 호출 제한, 개인정보 탐지 등 실무에서 필수적인 미들웨어 패턴을 다룹니다.

---

목차

1. 미들웨어_개념과_활용_사례
2. Tool_Retry_자동_재시도
3. Model_Fallback_대체_모델
4. Model_Call_Limit_호출_제한
5. PII_Detection_개인정보_탐지
6. before_model_after_model_훅
7. 커스텀_미들웨어_작성

---

1. 미들웨어 개념과 활용 사례

김개발 씨는 회사에서 LangChain을 활용한 AI 챗봇 프로젝트를 진행하고 있습니다. 그런데 OpenAI API가 가끔 타임아웃 에러를 뱉고, 사용자들의 개인정보가 그대로 로그에 남는 문제가 발생했습니다.

이 모든 문제를 한 번에 해결할 방법은 없을까요?

미들웨어는 한마디로 요청과 응답 사이에 끼어들어 무언가를 처리하는 중간 계층입니다. 마치 공항의 보안검색대처럼, 승객이 비행기에 타기 전에 짐을 검사하고, 내릴 때도 특정 절차를 거치게 하는 것과 같습니다.

LangChain에서 미들웨어를 활용하면 LLM 호출 전후로 로깅, 에러 처리, 보안 검사 등 다양한 작업을 자동화할 수 있습니다.

다음 코드를 살펴봅시다.

from langchain_core.callbacks import BaseCallbackHandler
from langchain_openai import ChatOpenAI

# 미들웨어 역할을 하는 콜백 핸들러 정의
class LoggingMiddleware(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        # LLM 호출 전에 실행됩니다
        print(f"[LOG] LLM 호출 시작: {len(prompts)}개 프롬프트")

    def on_llm_end(self, response, **kwargs):
        # LLM 호출 후에 실행됩니다
        print(f"[LOG] LLM 호출 완료: 토큰 사용량 확인")

# 미들웨어를 적용한 LLM 인스턴스 생성
llm = ChatOpenAI(callbacks=[LoggingMiddleware()])
result = llm.invoke("안녕하세요!")

김개발 씨는 입사 6개월 차 주니어 개발자입니다. 오늘도 열심히 AI 챗봇 코드를 작성하던 중, 이상한 현상을 발견했습니다.

프로덕션 서버에서 가끔씩 API 호출이 실패하고, 로그에는 사용자의 주민등록번호가 그대로 남아있었습니다. 선배 개발자 박시니어 씨가 다가와 코드를 살펴봅니다.

"아, 여기가 문제네요. 미들웨어 패턴을 사용하지 않아서 생긴 문제예요." 그렇다면 미들웨어란 정확히 무엇일까요?

쉽게 비유하자면, 미들웨어는 마치 공항의 보안검색대와 같습니다. 승객이 비행기를 타기 전에 짐 검사를 받고, 도착해서도 세관 검사를 거치듯이요.

이처럼 미들웨어도 요청이 목적지에 도달하기 전과 후에 특정 작업을 수행하는 역할을 합니다. 미들웨어가 없던 시절에는 어땠을까요?

개발자들은 LLM을 호출할 때마다 일일이 로깅 코드를 작성해야 했습니다. 에러가 발생하면 각 호출 지점에서 개별적으로 처리해야 했고, 코드가 중복되기 일쑤였습니다.

더 큰 문제는 보안 검사나 성능 모니터링 같은 횡단 관심사를 일관되게 적용하기가 매우 어려웠다는 점입니다. 바로 이런 문제를 해결하기 위해 미들웨어 패턴이 등장했습니다.

LangChain에서 미들웨어를 사용하면 관심사의 분리가 가능해집니다. 비즈니스 로직은 비즈니스 로직대로, 로깅은 로깅대로, 보안은 보안대로 각자의 역할에 집중할 수 있습니다.

또한 재사용성도 크게 높아집니다. 한 번 만든 미들웨어를 여러 LLM 인스턴스에 적용할 수 있으니까요.

위의 코드를 살펴보겠습니다. 먼저 BaseCallbackHandler를 상속받아 커스텀 핸들러를 만듭니다.

이것이 미들웨어의 기본 구조입니다. on_llm_start 메서드는 LLM 호출이 시작되기 전에 실행됩니다.

on_llm_end 메서드는 LLM 호출이 완료된 후에 실행됩니다. 마지막으로 이 핸들러를 callbacks 파라미터로 전달하면 미들웨어가 적용됩니다.

실제 현업에서는 어떻게 활용할까요? 예를 들어 금융 서비스 챗봇을 개발한다고 가정해봅시다.

모든 LLM 호출에 대해 감사 로그를 남겨야 하고, 개인정보가 포함된 응답은 마스킹 처리해야 합니다. 미들웨어를 활용하면 이런 요구사항을 비즈니스 로직을 건드리지 않고도 깔끔하게 구현할 수 있습니다.

하지만 주의할 점도 있습니다. 초보 개발자들이 흔히 하는 실수 중 하나는 미들웨어에서 너무 많은 작업을 수행하는 것입니다.

미들웨어는 가볍고 빠르게 동작해야 합니다. 무거운 작업은 비동기로 처리하거나 별도의 서비스로 분리하는 것이 좋습니다.

다시 김개발 씨의 이야기로 돌아가 봅시다. 박시니어 씨의 설명을 들은 김개발 씨는 고개를 끄덕였습니다.

"아, 그래서 여기저기 중복된 로깅 코드가 있었군요!"

실전 팁

💡 - 미들웨어는 가볍게 유지하고, 무거운 작업은 비동기로 처리하세요

• 여러 미들웨어를 조합할 때는 실행 순서를 명확히 파악해두세요
• 디버깅을 위해 미들웨어 자체의 로깅도 추가해두면 좋습니다

---

2. Tool Retry 자동 재시도

김개발 씨가 만든 챗봇이 프로덕션에 배포되었습니다. 그런데 새벽 시간에 유독 "서버 에러"가 많이 발생합니다.

알고 보니 OpenAI API가 간헐적으로 타임아웃을 뱉고 있었습니다. 사용자에게 "잠시 후 다시 시도해주세요"라고 안내하는 것 말고 더 좋은 방법은 없을까요?

Tool Retry는 도구 호출이 실패했을 때 자동으로 재시도하는 미들웨어 패턴입니다. 마치 전화를 걸었는데 상대방이 안 받으면 몇 번 더 시도해보는 것과 같습니다.

네트워크 불안정이나 일시적인 서버 과부하 같은 문제를 우아하게 처리할 수 있습니다.

다음 코드를 살펴봅시다.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import time

# 재시도 로직이 포함된 LLM 설정
llm = ChatOpenAI(
    model="gpt-4",
    max_retries=3,           # 최대 3번 재시도
    request_timeout=30,      # 30초 타임아웃
)

# 커스텀 재시도 래퍼 함수
def invoke_with_retry(llm, message, max_attempts=3, delay=1):
    for attempt in range(max_attempts):
        try:
            return llm.invoke([HumanMessage(content=message)])
        except Exception as e:
            if attempt < max_attempts - 1:
                print(f"재시도 {attempt + 1}/{max_attempts}: {e}")
                time.sleep(delay * (attempt + 1))  # 점진적 대기
            else:
                raise e

김개발 씨는 새벽 3시에 걸려온 전화에 잠을 깼습니다. 운영팀에서 "챗봇이 자꾸 에러를 뱉는다"며 급하게 연락한 것입니다.

로그를 확인해보니 OpenAI API에서 간헐적으로 503 Service Unavailable 에러가 발생하고 있었습니다. 다음 날 출근한 김개발 씨에게 박시니어 씨가 물었습니다.

"혹시 재시도 로직 넣어두셨어요?" 그렇다면 재시도 로직이란 무엇일까요? 쉽게 비유하자면, 재시도는 마치 바쁜 맛집에 전화 예약하는 것과 같습니다.

처음 전화했을 때 통화 중이라면, 조금 기다렸다가 다시 전화하고, 또 안 되면 조금 더 기다렸다가 다시 시도하는 것이죠. 대부분의 경우 몇 번 시도하면 결국 연결이 됩니다.

재시도 로직이 없으면 어떻게 될까요? API가 일시적으로 실패할 때마다 사용자는 에러 메시지를 보게 됩니다.

사실 1초만 기다렸다가 다시 호출하면 정상적으로 동작했을 텐데 말이죠. 이런 일시적인 실패 때문에 사용자 경험이 크게 나빠지고, 고객 문의가 쏟아지게 됩니다.

LangChain에서는 재시도를 어떻게 구현할까요? 가장 간단한 방법은 LLM 인스턴스 생성 시 max_retries 파라미터를 설정하는 것입니다.

이렇게 하면 LangChain이 내부적으로 지정된 횟수만큼 재시도를 수행합니다. 추가로 request_timeout을 설정하면 너무 오래 기다리는 것을 방지할 수 있습니다.

더 세밀한 제어가 필요하다면 커스텀 래퍼 함수를 만들 수 있습니다. 위 코드의 invoke_with_retry 함수를 보면, 실패할 때마다 대기 시간을 점점 늘리는 지수 백오프 패턴을 적용했습니다.

첫 번째 실패 후에는 1초, 두 번째 실패 후에는 2초를 기다립니다. 이렇게 하면 서버에 과도한 부하를 주지 않으면서도 안정적으로 재시도할 수 있습니다.

실제 현업에서는 어떤 상황에서 재시도가 필요할까요? 네트워크 일시 장애, API 서버의 순간적인 과부하, DNS 해석 실패 등 다양한 원인으로 일시적인 실패가 발생합니다.

특히 대용량 트래픽을 처리하는 서비스에서는 이런 문제가 더 자주 발생합니다. 재시도 로직은 이런 상황에서 시스템의 안정성을 크게 높여줍니다.

하지만 모든 에러에 재시도가 적절한 것은 아닙니다. 401 Unauthorized나 400 Bad Request 같은 에러는 재시도해도 소용없습니다.

API 키가 잘못되었거나 요청 자체가 잘못된 것이니까요. 따라서 재시도 가능한 에러와 그렇지 않은 에러를 구분하는 것이 중요합니다.

김개발 씨는 재시도 로직을 추가한 후 새벽 전화가 뚝 끊겼습니다. 대부분의 일시적 에러가 자동으로 복구되었기 때문입니다.

실전 팁

💡 - 재시도 횟수는 3-5회 정도가 적당합니다. 너무 많으면 응답 시간이 길어집니다

• 지수 백오프를 적용하여 서버 부하를 줄이세요
• 재시도 불가능한 에러(인증 실패, 잘못된 요청 등)는 즉시 실패 처리하세요

---

3. Model Fallback 대체 모델

어느 날 OpenAI API가 전면 장애를 일으켰습니다. 김개발 씨의 챗봇은 완전히 멈춰버렸고, 수천 명의 사용자가 서비스를 이용하지 못하게 되었습니다.

"하나의 API에 의존하는 게 이렇게 위험한 줄 몰랐어요..." 김개발 씨는 한숨을 쉬었습니다.

Model Fallback은 주 모델이 실패했을 때 자동으로 대체 모델로 전환하는 패턴입니다. 마치 단골 식당이 문을 닫았을 때 옆집 식당에 가는 것처럼, 서비스의 연속성을 보장합니다.

이를 통해 단일 장애점을 제거하고 시스템의 가용성을 크게 높일 수 있습니다.

다음 코드를 살펴봅시다.

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage

# 주 모델과 대체 모델 설정
primary_llm = ChatOpenAI(model="gpt-4")
fallback_llm = ChatAnthropic(model="claude-3-sonnet-20240229")

# with_fallbacks를 사용한 폴백 체인 구성
llm_with_fallback = primary_llm.with_fallbacks([fallback_llm])

# 사용 예시 - GPT-4 실패 시 자동으로 Claude로 전환
try:
    response = llm_with_fallback.invoke([
        HumanMessage(content="안녕하세요, 오늘 날씨 어때요?")
    ])
    print(response.content)
except Exception as e:
    print(f"모든 모델 실패: {e}")

김개발 씨는 그날의 악몽을 잊을 수가 없습니다. OpenAI가 4시간 동안 장애를 겪는 동안, 회사의 AI 챗봇 서비스는 완전히 마비되었습니다.

고객 문의는 폭주하고, 매출 손실은 눈덩이처럼 불어났습니다. 박시니어 씨가 조용히 말했습니다.

"이런 상황을 대비해서 폴백 전략을 세워뒀어야 했어요." 폴백이란 무엇일까요? 쉽게 비유하자면, 폴백은 마치 비상구와 같습니다.

건물의 주 출입구가 막혔을 때 비상구로 탈출할 수 있듯이, 주 모델이 응답하지 않을 때 대체 모델이 그 역할을 대신합니다. 사용자 입장에서는 어떤 모델이 응답했는지 알 필요도 없고, 서비스가 끊기지 않는다는 것만 중요합니다.

단일 모델에 의존하면 어떤 위험이 있을까요? 첫째, 가용성 문제입니다.

해당 API가 장애를 겪으면 전체 서비스가 멈춥니다. 둘째, 비용 문제입니다.

특정 모델의 가격이 급등하거나 할당량이 소진되면 서비스를 지속하기 어렵습니다. 셋째, 성능 문제입니다.

특정 시간대에 API 응답이 느려지면 사용자 경험이 나빠집니다. LangChain에서는 with_fallbacks 메서드를 제공합니다.

위 코드를 보면, GPT-4를 주 모델로, Claude를 대체 모델로 설정했습니다. GPT-4가 어떤 이유로든 실패하면 자동으로 Claude가 호출됩니다.

이 과정은 완전히 자동으로 이루어지며, 개발자가 별도로 에러 처리 로직을 작성할 필요가 없습니다. 폴백 체인은 여러 단계로 구성할 수도 있습니다.

예를 들어 GPT-4가 실패하면 GPT-3.5를 시도하고, 그것도 실패하면 Claude를 시도하는 식으로 구성할 수 있습니다. 비용이 저렴한 모델을 먼저 배치하거나, 응답 품질이 높은 모델을 우선 배치하는 등 비즈니스 요구사항에 맞게 전략을 세울 수 있습니다.

실제 현업에서는 어떻게 활용될까요? 금융권에서는 서비스 중단이 곧 금전적 손실로 이어집니다.

따라서 대부분의 금융 AI 서비스는 최소 2개 이상의 LLM 제공업체와 계약하고, 폴백 체인을 구성해둡니다. 이를 멀티 벤더 전략이라고 합니다.

주의할 점도 있습니다. 대체 모델의 응답 품질이 주 모델과 크게 다르면 사용자 경험에 일관성이 없어질 수 있습니다.

따라서 폴백 모델도 충분히 테스트하고, 응답 품질을 모니터링해야 합니다. 김개발 씨는 폴백 전략을 적용한 후 마음이 한결 편해졌습니다.

이제 한 API가 장애를 겪어도 서비스는 계속 돌아가니까요.

실전 팁

💡 - 폴백 모델도 주기적으로 테스트하여 정상 작동하는지 확인하세요

• 폴백이 발생했을 때 알림을 받을 수 있도록 모니터링을 설정하세요
• 비용과 품질의 균형을 고려하여 폴백 순서를 정하세요

---

4. Model Call Limit 호출 제한

월말이 되자 김개발 씨는 깜짝 놀랐습니다. OpenAI API 사용료가 예상의 3배가 청구된 것입니다.

알고 보니 악의적인 사용자가 봇을 만들어 챗봇에 수만 건의 요청을 보낸 것이었습니다. "이걸 어떻게 막을 수 있을까요?"

Model Call Limit은 LLM 호출 횟수를 제한하는 미들웨어 패턴입니다. 마치 뷔페 식당에서 1인당 음식을 가져갈 수 있는 횟수를 제한하는 것처럼, API 남용을 방지하고 비용을 통제합니다.

사용자별, 시간대별로 호출 제한을 설정하여 서비스의 안정성을 보장합니다.

다음 코드를 살펴봅시다.

from langchain_core.callbacks import BaseCallbackHandler
from datetime import datetime, timedelta
from collections import defaultdict

class RateLimitMiddleware(BaseCallbackHandler):
    def __init__(self, max_calls=100, window_minutes=60):
        self.max_calls = max_calls
        self.window = timedelta(minutes=window_minutes)
        self.call_history = defaultdict(list)

    def on_llm_start(self, serialized, prompts, **kwargs):
        user_id = kwargs.get("tags", ["anonymous"])[0]
        now = datetime.now()
        # 윈도우 시간 내의 호출 기록만 유지
        self.call_history[user_id] = [
            t for t in self.call_history[user_id] if now - t < self.window
        ]
        if len(self.call_history[user_id]) >= self.max_calls:
            raise Exception(f"호출 한도 초과: {self.max_calls}회/시간")
        self.call_history[user_id].append(now)

김개발 씨는 청구서를 보며 식은땀을 흘렸습니다. 한 달 예산의 3배가 넘는 금액이 찍혀 있었습니다.

로그를 분석해보니 특정 IP에서 하루에 10만 건이 넘는 요청이 들어온 것을 발견했습니다. 박시니어 씨가 말했습니다.

"레이트 리미팅을 적용했어야죠. API를 공개하면 이런 일이 꼭 생겨요." 레이트 리미팅이란 무엇일까요?

쉽게 비유하자면, 레이트 리미팅은 마치 놀이공원의 패스트패스와 같습니다. 인기 있는 놀이기구는 1인당 하루에 탈 수 있는 횟수가 제한되어 있죠.

이렇게 해야 모든 방문객이 공평하게 이용할 수 있고, 놀이기구도 과부하 없이 안전하게 운영됩니다. 호출 제한이 없으면 어떤 문제가 생길까요?

첫째, 비용 폭증입니다. 악의적인 사용자나 버그가 있는 클라이언트가 무한 루프로 API를 호출하면 순식간에 수백만 원의 비용이 발생할 수 있습니다.

둘째, 서비스 품질 저하입니다. 특정 사용자가 모든 리소스를 독점하면 다른 사용자들의 응답 속도가 느려집니다.

셋째, API 제공업체의 제재입니다. OpenAI 같은 제공업체도 자체적인 레이트 리밋이 있어서, 이를 초과하면 계정이 일시 정지될 수 있습니다.

위 코드는 어떻게 동작할까요? RateLimitMiddleware 클래스는 사용자별로 호출 기록을 저장합니다.

LLM이 호출될 때마다 on_llm_start 메서드가 실행되어 현재 사용자의 최근 호출 횟수를 확인합니다. 만약 설정된 한도를 초과하면 예외를 발생시켜 호출을 차단합니다.

슬라이딩 윈도우 방식을 사용한 것에 주목하세요. 단순히 "시간당 100회"라고 하면 1시 59분에 100회, 2시 1분에 100회를 호출하여 2분 사이에 200회 호출하는 편법이 가능합니다.

슬라이딩 윈도우는 현재 시점을 기준으로 지난 60분간의 호출 횟수를 계산하므로 이런 편법을 방지할 수 있습니다. 실제 현업에서는 더 정교한 제한이 필요합니다.

예를 들어 무료 사용자는 시간당 10회, 유료 사용자는 시간당 1000회처럼 요금제별로 다른 한도를 적용할 수 있습니다. 또한 Redis 같은 외부 저장소를 사용하여 여러 서버 인스턴스 간에 호출 기록을 공유하는 것이 일반적입니다.

김개발 씨는 레이트 리미팅을 적용한 후 다음 달 청구서를 안심하고 열어볼 수 있게 되었습니다.

실전 팁

💡 - 프로덕션 환경에서는 Redis 같은 분산 캐시를 사용하여 호출 기록을 관리하세요

• 사용자에게 남은 호출 횟수를 응답 헤더로 알려주면 좋습니다
• 한도 초과 시 429 Too Many Requests 상태 코드와 함께 재시도 가능 시간을 안내하세요

---

5. PII Detection 개인정보 탐지

보안팀에서 긴급 연락이 왔습니다. "챗봇 로그에서 사용자들의 주민등록번호와 카드번호가 발견되었습니다." 김개발 씨는 심장이 철렁했습니다.

개인정보보호법 위반은 회사에 엄청난 벌금과 평판 손실을 가져올 수 있기 때문입니다.

PII Detection은 개인식별정보를 탐지하고 마스킹하는 미들웨어입니다. 마치 공문서에서 민감한 정보를 검은 펜으로 가리는 것처럼, 주민등록번호, 전화번호, 이메일 등의 개인정보가 로그에 남거나 외부 API로 전송되는 것을 방지합니다.

GDPR, 개인정보보호법 준수를 위해 필수적인 보안 계층입니다.

다음 코드를 살펴봅시다.

import re
from langchain_core.callbacks import BaseCallbackHandler

class PIIDetectionMiddleware(BaseCallbackHandler):
    def __init__(self):
        # 한국 개인정보 패턴 정의
        self.patterns = {
            "주민등록번호": r"\d{6}-[1-4]\d{6}",
            "전화번호": r"01[0-9]-\d{3,4}-\d{4}",
            "카드번호": r"\d{4}-\d{4}-\d{4}-\d{4}",
            "이메일": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
        }

    def on_llm_start(self, serialized, prompts, **kwargs):
        for prompt in prompts:
            for pii_type, pattern in self.patterns.items():
                if re.search(pattern, prompt):
                    raise ValueError(f"개인정보 탐지됨: {pii_type}")

    def mask_pii(self, text):
        for pii_type, pattern in self.patterns.items():
            text = re.sub(pattern, f"[{pii_type} 마스킹]", text)
        return text

김개발 씨는 보안팀의 연락을 받고 밤새 로그를 분석했습니다. 사용자들이 챗봇에 자신의 개인정보를 입력하고, 그 정보가 그대로 로그 파일과 OpenAI 서버로 전송되고 있었습니다.

박시니어 씨가 심각한 표정으로 말했습니다. "PII 필터링은 AI 서비스의 기본 중의 기본이에요.

이건 정말 큰 문제가 될 수 있어요." PII란 무엇일까요? PII는 Personally Identifiable Information의 약자로, 개인을 식별할 수 있는 정보를 뜻합니다.

주민등록번호, 전화번호, 이메일, 신용카드 번호, 주소 등이 여기에 해당합니다. 쉽게 비유하자면, PII는 마치 집 열쇠와 같습니다.

다른 사람 손에 들어가면 큰 피해를 입을 수 있는 민감한 정보입니다. 왜 PII 탐지가 중요할까요?

첫째, 법적 의무입니다. 한국의 개인정보보호법, 유럽의 GDPR 등은 개인정보의 무단 수집과 전송을 엄격히 금지합니다.

위반 시 수십억 원의 과징금이 부과될 수 있습니다. 둘째, 신뢰 문제입니다.

개인정보 유출 사고가 발생하면 기업의 평판은 회복하기 어렵습니다. 셋째, 보안 위험입니다.

외부 API로 전송된 개인정보가 어디에 저장되고 누가 접근할 수 있는지 알 수 없습니다. 위 코드는 어떻게 동작할까요?

PIIDetectionMiddleware는 정규표현식을 사용하여 다양한 개인정보 패턴을 탐지합니다. 주민등록번호는 "123456-1234567" 형식, 전화번호는 "010-1234-5678" 형식 등을 인식합니다.

LLM이 호출되기 전에 프롬프트를 검사하여, 개인정보가 발견되면 예외를 발생시켜 호출을 차단합니다. mask_pii 메서드는 개인정보를 마스킹 처리합니다.

완전히 차단하는 대신, 개인정보 부분만 마스킹하여 나머지 요청은 정상적으로 처리하고 싶을 때 사용합니다. 예를 들어 "제 전화번호는 010-1234-5678입니다"라는 입력은 "제 전화번호는 [전화번호 마스킹]입니다"로 변환됩니다.

실제 현업에서는 더 정교한 탐지가 필요합니다. 정규표현식만으로는 "영일공-일이삼사-오육칠팔"처럼 한글로 작성된 전화번호나, 문맥상 개인정보인 경우를 탐지하기 어렵습니다.

따라서 Microsoft Presidio나 AWS Macie 같은 전문 PII 탐지 서비스를 함께 사용하는 것이 일반적입니다. 주의할 점도 있습니다.

너무 엄격한 필터링은 정상적인 서비스 이용을 방해할 수 있습니다. 예를 들어 "1234-5678"이라는 숫자가 항상 개인정보인 것은 아닙니다.

오탐을 줄이면서도 보안을 유지하는 균형점을 찾는 것이 중요합니다. 김개발 씨는 PII 탐지 미들웨어를 적용한 후 보안팀의 승인을 받고 서비스를 재개할 수 있었습니다.

실전 팁

💡 - 정규표현식 외에도 ML 기반 PII 탐지 라이브러리를 함께 사용하세요

• 탐지된 개인정보는 로그에도 남기지 않도록 주의하세요
• 사용자에게 개인정보 입력 자제를 안내하는 UI 문구도 도움이 됩니다

---

6. before model after model 훅

김개발 씨는 LLM 호출 전후에 다양한 작업을 수행해야 했습니다. 호출 전에는 프롬프트를 정제하고, 호출 후에는 응답을 가공해야 했습니다.

매번 같은 코드를 반복 작성하다 보니 지치기 시작했습니다. 더 깔끔한 방법은 없을까요?

@before_model과 @after_model 훅은 LLM 호출 전후에 자동으로 실행되는 데코레이터 패턴입니다. 마치 무대 뒤에서 공연 전 조명을 켜고, 공연 후 무대를 정리하는 스태프처럼, 개발자가 직접 호출하지 않아도 자동으로 필요한 작업을 수행합니다.

이를 통해 관심사를 깔끔하게 분리하고 코드 재사용성을 높일 수 있습니다.

다음 코드를 살펴봅시다.

from langchain_core.callbacks import BaseCallbackHandler
from functools import wraps
import time

# 훅 데코레이터 구현
def before_model(func):
    @wraps(func)
    def wrapper(prompt, *args, **kwargs):
        # 모델 호출 전 프롬프트 전처리
        cleaned_prompt = prompt.strip()
        print(f"[BEFORE] 프롬프트 길이: {len(cleaned_prompt)}자")
        return func(cleaned_prompt, *args, **kwargs)
    return wrapper

def after_model(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        elapsed = time.time() - start
        # 모델 호출 후 결과 후처리
        print(f"[AFTER] 응답 시간: {elapsed:.2f}초")
        return result
    return wrapper

# 훅 적용 예시
@before_model
@after_model
def ask_llm(prompt):
    # 실제 LLM 호출 로직
    return llm.invoke(prompt)

김개발 씨는 코드를 보다가 한숨을 쉬었습니다. LLM을 호출하는 곳이 20군데가 넘는데, 매번 똑같은 로깅 코드와 시간 측정 코드를 복사해서 붙여넣고 있었습니다.

박시니어 씨가 코드를 보더니 말했습니다. "이런 반복 작업은 훅을 사용하면 깔끔하게 처리할 수 있어요." 훅이란 무엇일까요?

쉽게 비유하자면, 훅은 마치 자동 센서등과 같습니다. 사람이 지나가면 자동으로 불이 켜지고, 일정 시간이 지나면 자동으로 꺼지죠.

사람이 일일이 스위치를 켜고 끌 필요가 없습니다. 훅도 마찬가지로, 특정 이벤트가 발생하면 자동으로 지정된 코드가 실행됩니다.

@before_model은 LLM 호출 전에 실행됩니다. 프롬프트를 정제하거나, 입력 유효성을 검사하거나, 로깅을 남기는 등의 작업을 수행합니다.

위 코드에서는 프롬프트의 앞뒤 공백을 제거하고 길이를 출력합니다. @after_model은 LLM 호출 후에 실행됩니다.

응답 시간을 측정하거나, 결과를 가공하거나, 캐시에 저장하는 등의 작업을 수행합니다. 위 코드에서는 호출에 걸린 시간을 측정하여 출력합니다.

데코레이터를 스택처럼 쌓을 수 있다는 점도 중요합니다. 위 코드에서 @before_model과 @after_model을 함께 적용했습니다.

이렇게 하면 호출 전에 before_model이 실행되고, 호출 후에 after_model이 실행됩니다. 여러 개의 훅을 조합하여 복잡한 파이프라인을 구성할 수도 있습니다.

이런 패턴의 장점은 무엇일까요? 첫째, 관심사의 분리입니다.

비즈니스 로직과 부가 기능을 깔끔하게 분리할 수 있습니다. 둘째, 코드 재사용입니다.

한 번 만든 훅을 여러 함수에 적용할 수 있습니다. 셋째, 유지보수 용이성입니다.

로깅 방식을 바꾸고 싶다면 훅만 수정하면 됩니다. 실제 현업에서는 어떻게 활용될까요?

예를 들어 프롬프트 인젝션 방어, 토큰 사용량 추적, A/B 테스트 로깅, 응답 캐싱 등을 훅으로 구현할 수 있습니다. 특히 여러 팀이 같은 LLM 인프라를 공유할 때, 공통 훅을 제공하면 일관된 품질을 유지할 수 있습니다.

김개발 씨는 훅 패턴을 적용한 후 중복 코드가 사라지고 코드베이스가 훨씬 깔끔해졌습니다.

실전 팁

💡 - 훅은 가볍게 유지하고, 무거운 작업은 비동기로 처리하세요

• 훅의 실행 순서를 명확히 문서화해두세요
• 훅에서 예외가 발생해도 전체 서비스가 멈추지 않도록 에러 처리를 해두세요

---

7. 커스텀 미들웨어 작성

김개발 씨의 회사에는 독특한 요구사항이 있었습니다. 모든 LLM 응답에 회사 면책 조항을 붙이고, 특정 금지어가 포함된 응답은 필터링해야 했습니다.

기존 미들웨어로는 해결할 수 없는 상황이었습니다. 직접 만들어야 할 때가 온 것입니다.

커스텀 미들웨어는 비즈니스 요구사항에 맞게 직접 작성하는 미들웨어입니다. 마치 맞춤 양복처럼, 기성품으로는 만족할 수 없는 특별한 요구사항을 정확히 충족시킵니다.

LangChain의 콜백 시스템을 활용하면 호출의 모든 단계에 개입하는 강력한 커스텀 미들웨어를 만들 수 있습니다.

다음 코드를 살펴봅시다.

from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from typing import Any, Dict, List

class CompanyComplianceMiddleware(BaseCallbackHandler):
    def __init__(self, blocked_words: List[str], disclaimer: str):
        self.blocked_words = blocked_words
        self.disclaimer = disclaimer
        self.total_tokens = 0

    def on_llm_start(self, serialized: Dict, prompts: List[str], **kwargs):
        print(f"[시작] 요청 ID: {kwargs.get('run_id', 'N/A')}")

    def on_llm_end(self, response: LLMResult, **kwargs):
        # 토큰 사용량 집계
        if response.llm_output:
            usage = response.llm_output.get("token_usage", {})
            self.total_tokens += usage.get("total_tokens", 0)
        print(f"[완료] 누적 토큰: {self.total_tokens}")

    def on_llm_error(self, error: Exception, **kwargs):
        print(f"[에러] {type(error).__name__}: {error}")

# 사용 예시
middleware = CompanyComplianceMiddleware(
    blocked_words=["경쟁사명", "비속어"],
    disclaimer="본 응답은 참고용이며 법적 효력이 없습니다."
)

김개발 씨는 법무팀의 요청을 받고 난감해졌습니다. 모든 AI 응답에 면책 조항을 붙이고, 경쟁사 이름이 언급되면 응답을 차단해야 한다는 것이었습니다.

인터넷을 뒤져봐도 이런 기능을 제공하는 라이브러리는 없었습니다. 박시니어 씨가 말했습니다.

"이럴 때 커스텀 미들웨어를 만들면 돼요. 생각보다 어렵지 않아요." 커스텀 미들웨어는 어떻게 만들까요?

LangChain에서는 BaseCallbackHandler를 상속받아 커스텀 미들웨어를 만들 수 있습니다. 이 클래스는 LLM 호출의 각 단계에서 호출되는 여러 메서드를 제공합니다.

필요한 메서드만 오버라이드하면 됩니다. 주요 콜백 메서드를 살펴보겠습니다.

on_llm_start는 LLM 호출이 시작될 때 실행됩니다. 프롬프트 검증, 요청 로깅, 시작 시간 기록 등에 사용합니다.

on_llm_end는 LLM 호출이 성공적으로 완료되었을 때 실행됩니다. 응답 가공, 토큰 집계, 캐싱 등에 사용합니다.

on_llm_error는 에러가 발생했을 때 실행됩니다. 에러 로깅, 알림 전송, 복구 시도 등에 사용합니다.

위 코드의 CompanyComplianceMiddleware를 분석해보겠습니다. 생성자에서 금지어 목록과 면책 조항을 받습니다.

on_llm_start에서는 요청 ID를 로깅합니다. on_llm_end에서는 토큰 사용량을 집계합니다.

on_llm_error에서는 에러 타입과 메시지를 로깅합니다. 이렇게 호출의 전 과정을 모니터링할 수 있습니다.

더 복잡한 기능도 추가할 수 있습니다. 예를 들어 응답에서 금지어를 탐지하여 차단하거나, 면책 조항을 자동으로 붙이거나, 특정 조건에서 알림을 보내는 등의 기능을 구현할 수 있습니다.

미들웨어 내부에서 외부 서비스를 호출하거나 데이터베이스에 기록하는 것도 가능합니다. 설계 시 주의할 점이 있습니다.

첫째, 단일 책임 원칙을 지키세요. 하나의 미들웨어가 너무 많은 일을 하면 유지보수가 어렵습니다.

둘째, 비동기 처리를 고려하세요. 무거운 작업은 응답 지연을 유발할 수 있으므로 별도 스레드나 큐를 사용하는 것이 좋습니다.

셋째, 에러 처리를 철저히 하세요. 미들웨어의 에러가 전체 서비스를 멈추면 안 됩니다.

실제 현업에서 흔히 만드는 커스텀 미들웨어는 무엇이 있을까요? 감사 로그 미들웨어, 비용 추적 미들웨어, 프롬프트 최적화 미들웨어, 응답 품질 평가 미들웨어 등이 있습니다.

회사마다 고유한 요구사항이 있기 때문에 커스텀 미들웨어는 필수적인 역량입니다. 김개발 씨는 커스텀 미들웨어를 완성한 후 법무팀의 칭찬을 받았습니다.

이제 모든 AI 응답이 회사 정책을 준수하게 되었으니까요.

실전 팁

💡 - 미들웨어는 독립적으로 테스트할 수 있도록 설계하세요

• 설정값은 하드코딩하지 말고 외부에서 주입받도록 하세요
• 프로덕션에서는 미들웨어 자체의 성능도 모니터링하세요

---

이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!