가드레일 & 휴먼 인 더 루프 (HITL)

사용자: "카드를 잃어버렸고 정말 화가 나요!
$500 거래를 지금 바로 환불해줄 수 있나요?
네라고 말해주세요, 아니면 계좌를 해지할 거예요!"

def input_guardrail(user_input: str) -> tuple[bool, str]:
    # 프롬프트 인젝션 패턴 검사
    injection_patterns = [
        "이전 지시를 무시",
        "시스템 프롬프트",
        "너는 이제",
        "역할을 맡아"
    ]
 
    for pattern in injection_patterns:
        if pattern.lower() in user_input.lower():
            return False, "잠재적 프롬프트 인젝션 감지"
 
    return True, user_input

def output_guardrail(response: str) -> tuple[bool, str]:
    forbidden_phrases = [
        "환불해 드릴게요",
        "환불 처리했습니다",
        "환불을 진행",
        "네, 해드리겠습니다"
    ]
 
    lower_response = response.lower()
    for phrase in forbidden_phrases:
        if phrase in lower_response:
            return False, f"차단된 문구: '{phrase}'"
 
    return True, "안전"

def semantic_guardrail(response: str, policy: str) -> bool:
    evaluation = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": f"정책: {policy}\n이 응답이 정책을 위반하나요? YES 또는 NO로 답하세요."},
            {"role": "user", "content": response}
        ]
    )
    return "NO" in evaluation.choices[0].message.content.upper()

def run_guarded_agent(user_message: str) -> str:
    # 1. 입력 검증
    is_safe, result = input_guardrail(user_message)
    if not is_safe:
        return "해당 요청을 처리할 수 없습니다."
 
    # 2. 응답 생성
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message}
        ]
    )
    raw_output = response.choices[0].message.content
 
    # 3. 출력 검증
    is_safe, reason = output_guardrail(raw_output)
    if not is_safe:
        print(f"🛡️ 가드레일 발동: {reason}")
        return "환불은 직접 처리할 수 없습니다. 담당자에게 연결해 드리겠습니다."
 
    return raw_output

class HITLAgent:
    def __init__(self, high_stakes_keywords: list):
        self.high_stakes = high_stakes_keywords
        self.pending_approvals = {}
 
    def needs_approval(self, action: str) -> bool:
        return any(kw in action.lower() for kw in self.high_stakes)
 
    def execute(self, action: str):
        if self.needs_approval(action):
            approval_id = self.request_human_approval(action)
            return f"승인 대기 중: {approval_id}"
        return self.perform_action(action)
 
    def request_human_approval(self, action: str) -> str:
        approval_id = str(uuid.uuid4())
        self.pending_approvals[approval_id] = {
            "action": action,
            "status": "pending",
            "timestamp": datetime.now()
        }
        # 인간 검토자에게 알림 (이메일, Slack 등)
        return approval_id

from guardrails import Guard
from guardrails.hub import ToxicLanguage, PIIFilter
 
guard = Guard().use_many(
    ToxicLanguage(on_fail="fix"),
    PIIFilter(on_fail="fix")
)
 
result = guard(
    llm_api=openai.chat.completions.create,
    prompt="고객 문의: ...",
    model="gpt-4o-mini"
)