AI 에이전트란?

사용자: "VIP 고객 중 최근 3개월 미구매자에게 쿠폰을 발송해 주세요"

에이전트:
  1. [생각] VIP 고객 목록이 필요합니다 → get_customer_list(tier='VIP') 호출
  2. [관찰] VIP 고객 150명 확인
  3. [생각] 3개월 미구매자를 필터링해야 합니다 → query_orders(days=90) 호출
  4. [관찰] 42명이 3개월 내 구매 없음
  5. [생각] 쿠폰을 발송해야 합니다 → send_coupon(customers=42명, type='10%할인') 호출
  6. [관찰] 42명에게 쿠폰 발송 완료
  7. [응답] "VIP 고객 중 최근 3개월 미구매자 42명에게 10% 할인 쿠폰을 발송했습니다."

# Databricks에서 Single Agent 구현 예시
from databricks.agents import ChatAgent, Tool

class HRAgent(ChatAgent):
    def __init__(self):
        self.tools = [
            Tool(func=lookup_employee, description="직원 정보 조회"),
            Tool(func=check_leave_balance, description="연차 잔여일 조회"),
            Tool(func=submit_leave_request, description="연차 신청"),
        ]

# 라우터 기반 Multi-Agent 패턴
agents = {
    "hr": HRAgent(),       # HR 관련 질문 전담
    "it": ITAgent(),       # IT 지원 전담
    "finance": FinanceAgent(),  # 재무 관련 전담
}

def route(user_query: str) -> str:
    # LLM이 사용자 질문을 분류하여 적절한 에이전트로 라우팅
    category = classify_intent(user_query)
    return agents[category].chat(user_query)

# 슬라이딩 윈도우 + 요약 하이브리드 전략
def manage_memory(messages: list, max_recent: int = 10, max_tokens: int = 4000):
    if len(messages) <= max_recent:
        return messages

    # 오래된 메시지를 요약
    old_messages = messages[:-max_recent]
    summary = llm.summarize(old_messages)

    # 요약 + 최근 메시지를 결합
    return [{"role": "system", "content": f"이전 대화 요약: {summary}"}] + messages[-max_recent:]

# Lakebase를 활용한 장기 메모리 저장 예시
import psycopg2

def save_user_preference(user_id: str, key: str, value: str):
    """사용자 선호도를 Lakebase에 저장"""
    conn = psycopg2.connect(host="lakebase-endpoint", dbname="agent_memory")
    with conn.cursor() as cur:
        cur.execute("""
            INSERT INTO user_preferences (user_id, pref_key, pref_value, updated_at)
            VALUES (%s, %s, %s, NOW())
            ON CONFLICT (user_id, pref_key) DO UPDATE SET pref_value = %s, updated_at = NOW()
        """, (user_id, key, value, value))
    conn.commit()

# Human-in-the-loop 패턴: 위험한 도구 실행 전 승인 요청
DANGEROUS_TOOLS = {"delete_record", "process_payment", "update_user_data"}

def execute_tool(tool_name: str, params: dict, user_session):
    if tool_name in DANGEROUS_TOOLS:
        # 사용자에게 승인 요청
        approval = request_human_approval(
            user_session,
            action=f"{tool_name}({params})",
            message=f"다음 작업을 실행하시겠습니까?"
        )
        if not approval:
            return "사용자가 작업을 취소했습니다."

    return tools[tool_name].execute(**params)

# Guardrails를 활용한 입출력 필터링
import re

PII_PATTERNS = {
    "주민등록번호": r"\d{6}-[1-4]\d{6}",
    "신용카드": r"\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}",
    "전화번호": r"01[0-9]-\d{3,4}-\d{4}",
}

def sanitize_output(response: str) -> str:
    """에이전트 응답에서 PII를 마스킹"""
    for pii_type, pattern in PII_PATTERNS.items():
        response = re.sub(pattern, f"[{pii_type} 마스킹됨]", response)
    return response

# 모든 도구 호출을 MLflow로 추적하여 보안 감사 로그 생성
import mlflow

@mlflow.trace
def agent_chat(user_input: str):
    with mlflow.start_span("input_validation"):
        validated_input = validate_and_sanitize(user_input)

    with mlflow.start_span("llm_reasoning"):
        response = llm.generate(validated_input)

    with mlflow.start_span("tool_execution"):
        result = execute_tools(response.tool_calls)

    with mlflow.start_span("output_sanitization"):
        safe_output = sanitize_output(result)

    return safe_output

# 월 비용 추정 예시 (일 1,000 대화, 평균 5턴/대화 기준)
- LLM 토큰: 1,000 x 5 x 4,000 토큰 = 20M 토큰/일 ≈ $300~600/월
- Model Serving: Provisioned Throughput 2 인스턴스 ≈ $2,000~4,000/월
- Vector Search: Endpoint 1개 ≈ $200~500/월
- 총 예상: $2,500~5,100/월

# MLflow Evaluate를 활용한 에이전트 품질 평가
import mlflow

eval_data = [
    {"input": "지난 달 매출 알려줘", "expected_tool": "get_sales_data", "expected_contains": "매출"},
    {"input": "서울 날씨 어때?", "expected_tool": "get_weather", "expected_contains": "온도"},
]

results = mlflow.evaluate(
    model=agent_model_uri,
    data=eval_data,
    model_type="databricks-agent",
    evaluator_config={
        "databricks-agent": {
            "metrics": ["toxicity", "groundedness", "relevance", "tool_accuracy"]
        }
    }
)

# 결과 확인
print(f"Tool 정확도: {results.metrics['tool_accuracy/mean']:.2%}")
print(f"Groundedness: {results.metrics['groundedness/mean']:.2%}")
print(f"Relevance: {results.metrics['relevance/mean']:.2%}")