[AI 에이전트 엔지니어링] Ch10. 운영 환경 모니터링

목표

• 기존 소프트웨어와 달리 확률적 작동하므로 모든 것을 test case 해볼 수 없음
• 모니터링의 역할: 단순 장애 탐지 → 프로덕션 피드백을 통한 개선 루프 구축
• 학습 경로: 지표 정의 → 스택 선택 → 계측 → 시각화 → 운영 패턴 

---

에이전트가 실패하는 경우?

• 미묘한 실패
  • 도구는 성공했으나, 체인 오류로 실패
  • LLM 출력이 유창하지만 사용자가 이해하지 못하기도 함
  • 계획이 부분적으로 작동하여 목표 미달성
• 인프라 신호: CPU/메모리/지연시간만으로는 불충분
• 의미 파악 여부, 할루시네이션, 작업 포기 추적 필수
• 프로덕션 가시성은 선택이 아니라 필수: 문제가 커지기 전에 실패 포착 가능
• 실패를 테스트 케이스로 전환하여 회귀 테스트 강화

---

 

오픈텔레메트리(opentelemetry)

• 각 단계를 추적하도록 span을 추적함
  • 그러나, 너무 많이 하면 데이터가 너무 많이 쌓여 정리가 안 돼...

---

시각화와 실시간 알림(그라파나)

• 할루시네이션 비율이 30분간 5% 초과 → Slack/PageDuty 알림
• 재시도 루프 3회 초과  → 온콜 엔지니어 호출

---

사용자 피드백

• 암시적 피드백: 작업 포기, 상호작용 머뭇거림
• 명시적 피드백: 👍/ 👎, 별점(⭐), 자유 텍스트

---

분포 변화 탐지

• KS 검정
• KL 발산
• PSI
• 대응: 임베딩 코사인 유사도 < 0.8 => 재학습이나 가드레일 적용

---

예제 1번: KS검정

import numpy as np
from scipy import stats
from sklearn.metrics.pairwise import cosine_similarity
from sklearn.feature_extraction.text import TfidfVectorizer

# 예제 1: 수치형 기능 편향(drift)에 대한 Kolmogorov-Smirnov (KS) 검정 (예: 쿼리 길이)
def detect_ks_drift(historical_data: np.ndarray, current_data: np.ndarray, threshold: float = 0.1) -> bool:
    """
    KS 검정을 사용하여 분포 변화를 감지합니다.
    유의미한 편향(통계량 > 임계값)이 있으면 True를 반환합니다.
    """
    ks_stat, p_value = stats.ks_2samp(historical_data, current_data)
    print(f"KS 통계량: {ks_stat}, p-값: {p_value}")
    return ks_stat > threshold

# 사용 예시
historical_lengths = np.array([len(q) for q in ["What is the weather?", "Book a flight to Paris", "Recommend a book"] * 100])
current_lengths = np.array([len(q) for q in ["Query about latest AI news", "Longer user input with details"] * 150])
if detect_ks_drift(historical_lengths, current_lengths):
    print("편향 감지됨: 입력 변경 사항을 검토하세요.")

 

 

1. historical_lengths: 과거 사용자 입력 길이

• What is the weather? → 20자
• Book a flight to Paris → 22자
• Recommend a book → 16자

2. current_lengths: 최근 사용자 입력 길이

• Query about latest AI news → 26자
• Longer user input with details → 30자

3. ks_2samp: 두 그룹의 데이터가 근본적으로 같은 분포에서 추출된 것인지 확인하는 함수

• KS 통계량: 1.0, p-value: 1.48e-179 → < 0.1이면, 동일한 분포를 가진다는 귀무가설 기각 → 두 분포가 매우 다름
• 편향 감지됨: 입력 변경 사항을 검토하세요.

---

예제 2번: KL Divergence

# 예제 2: 확률 분포 변화에 대한 Kullback-Leibler (KL) 발산 (예: 토큰 분포)
def kl_divergence(p: np.ndarray, q: np.ndarray, epsilon: float = 1e-10) -> float:
    """
    두 확률 분포 간의 KL 발산을 계산합니다.
    0으로 나누는 것을 방지하기 위해 엡실론을 추가합니다.
    """
    # 두 배열의 길이를 맞춥니다 (작은 쪽을 기준으로 자르거나 0으로 채움)
    min_len = min(len(p), len(q))
    p = p[:min_len]
    q = q[:min_len]

    p = p + epsilon
    q = q + epsilon
    p = p / np.sum(p)
    q = q / np.sum(q)
    return np.sum(p * np.log(p / q))

# 사용 예시 (토큰 빈도 히스토그램)
historical_tokens = np.bincount([ord(c) for q in ["hello world"] * 100 for c in q], minlength=256)  # 단순화된 토큰 카운트
current_tokens = np.bincount([ord(c) for q in ["hola mundo"] * 100 for c in q], minlength=256)
kl_score = kl_divergence(historical_tokens, current_tokens)
print(f"KL 발산: {kl_score}")
if kl_score > 0.5:
    print("개념 편향 감지됨: 언어 변화 가능성.")

 

1. historical_tokens: 과거 토큰 분포: hello world → [h, e, l, o, w, r, d]

2. current_tokens: 최근 토큰 분포: hola mundo  → [h, o, l, a, m, u, n, d]

3. kl_divergence()

• KL 발산: 7.74 > 0.5
• 개념 편향 감지됨: 언어 변화 가능성
• 즉, 영어 중심  → 스페인어 중심

---

예제 3번

# 예제 3: 범주형 지표에 대한 모집단 안정성 지수 (PSI) (예: 도구 사용)
def calculate_psi(expected: np.ndarray, actual: np.ndarray, buckets: int = 10) -> float:
    """
    범주형 또는 구간화된 연속형 데이터에 대한 PSI를 계산합니다.
    """
    expected_percents = expected / np.sum(expected)
    actual_percents = actual / np.sum(actual)
    psi_values = (actual_percents - expected_percents) * np.log(actual_percents / expected_percents)
    return np.sum(psi_values)

# 사용 예시 (도구 호출 횟수)
historical_tools = np.array([50, 30, 20])  # 예: 'refund', 'cancel', 'modify'에 대한 카운트
current_tools = np.array([40, 40, 20])
psi = calculate_psi(historical_tools, current_tools)
print(f"PSI: {psi}")
if psi > 0.25:
    print("주요 편향 발생: 개입 필요.")
elif psi > 0.1:
    print("도구 사용에 경미한 편향 발생.")
else:
    print("정상 범위")

 

1. historical_tools

• refund : 50회
  cancel : 30회
  modify : 20회

2. current_tools

• refund : 40회
  cancel : 40회
  modify : 20회

3. calculate_psi()

• PSI: 0.051
• 정상 범위

---

예제 4번

# 예제 4: 쿼리 편향에 대한 임베딩 기반 유사도
def detect_embedding_drift(historical_queries: list, current_queries: list, threshold: float = 0.8):
    """
    쿼리 임베딩 간의 평균 코사인 유사도를 계산합니다.
    단순함을 위해 TF-IDF를 사용합니다. 더 나은 의미론적 분석을 위해 sentence transformers로 대체하세요.
    """
    vectorizer = TfidfVectorizer()
    all_queries = historical_queries + current_queries
    embeddings = vectorizer.fit_transform(all_queries)
    hist_emb = embeddings[:len(historical_queries)]
    curr_emb = embeddings[len(historical_queries):]
    similarities = cosine_similarity(curr_emb, hist_emb)
    mean_sim = np.mean(similarities)
    print(f"평균 코사인 유사도: {mean_sim}")
    return mean_sim < threshold

# 사용 예시
historical = ["Refund my order", "Cancel shipment", "Change address"] * 50
current = ["Return damaged item", "Stop delivery now", "Update shipping info"] * 50
if detect_embedding_drift(historical, current):
    print("쿼리 편향 감지됨: 프롬프트를 재학습하거나 조정하세요.")

 

 

 

1. historical: 과거에 많이 들어오던 사용자 요청들 (150개 데이터 가라로 하기 위해 *50)

• Refund my order: 주문 환불
• Cancel shipment: 배송 취소
• Change address: 주소 변경

2. current: 최근 들어오는 새로운 요청들 (150개 데이터 가라로 하기 위해 *50)

• Return damaged item: 파손된 상품 환
• Stop delivery now : 배송 지금 중단
• Update shipping info: 배송 정보 변경

3. detect_embedding_drift

• TfidfVectorizer(): TF - IDF ( Term Frequency (단어 빈도) - Inverse Document Frequency 역 문서 빈도 )
  • ex) Refund my order → 숫자로 바꿈
• fit_transform(): 숫자 →  벡터
• cosine_similarity(): 코사인 유사도 계산
  • 0 → < 0.8  → 쿼리 편향 감지됨: 프롬프트를 재학습하거나 조정하세요.