5장: 콘텐츠 안전성과 유해 출력 방지

from azure.ai.contentsafety import ContentSafetyClient
from azure.core.credentials import AzureKeyCredential
 
client = ContentSafetyClient(
    endpoint="https://your-resource.cognitiveservices.azure.com",
    credential=AzureKeyCredential("your-key"),
)
 
def analyze_content(text: str) -> dict:
    """텍스트의 안전성 분석"""
    from azure.ai.contentsafety.models import AnalyzeTextOptions
 
    response = client.analyze_text(
        AnalyzeTextOptions(text=text)
    )
 
    results = {}
    for category in response.categories_analysis:
        results[category.category] = {
            "severity": category.severity,  # 0, 2, 4, 6
            "is_safe": category.severity <= 2,
        }
 
    return results

async def detect_hallucination(
    client,
    question: str,
    answer: str,
    sources: list[str],
) -> dict:
    """답변이 소스에 근거하는지 검증"""
    source_text = "\n---\n".join(sources)
 
    response = await client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""다음 답변이 제공된 소스 자료에 충실한지 검증해주세요.
 
소스 자료:
{source_text}
 
질문: {question}
답변: {answer}
 
검증 결과를 JSON으로:
{{
  "is_faithful": true/false,
  "unsupported_claims": ["소스에서 확인할 수 없는 주장 목록"],
  "confidence": 0.0-1.0
}}""",
        }],
    )
    return parse_json(response.content[0].text)

async def check_self_consistency(
    client,
    question: str,
    num_samples: int = 3,
) -> dict:
    """여러 번 생성하여 일관성 확인"""
    answers = []
    for _ in range(num_samples):
        response = await client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=500,
            temperature=0.7,
            messages=[{"role": "user", "content": question}],
        )
        answers.append(response.content[0].text)
 
    # 답변 간 일관성 평가
    consistency_check = await client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": f"""다음 {num_samples}개의 답변이 서로 일관되는지 평가하세요.
 
{chr(10).join(f'답변 {i+1}: {a}' for i, a in enumerate(answers))}
 
JSON 응답:
{{"is_consistent": true/false, "inconsistencies": ["불일치 내용"], "confidence": 0.0-1.0}}""",
        }],
    )
    return parse_json(consistency_check.content[0].text)

bias_detection_prompt = """다음 텍스트에서 잠재적 편향을 분석해주세요.
 
텍스트: {text}
 
다음 편향 유형을 검사:
1. 성별 편향: 특정 성별에 대한 고정관념이나 불균형
2. 인종/민족 편향: 특정 인종이나 민족에 대한 편견
3. 연령 편향: 특정 연령대에 대한 선입견
4. 문화 편향: 특정 문화적 관점의 과대 대표
5. 확증 편향: 한쪽 관점만 제시
 
JSON 응답:
{{
  "has_bias": true/false,
  "detected_biases": [
    {{"type": "성별", "description": "설명", "severity": "low/medium/high"}}
  ],
  "suggestion": "개선 제안"
}}"""

DOMAIN_DISCLAIMERS = {
    "medical": "이 정보는 의학적 조언을 대체하지 않습니다. 반드시 의료 전문가와 상담하세요.",
    "legal": "이 정보는 법률 자문을 대체하지 않습니다. 구체적인 법률 문제는 변호사와 상담하세요.",
    "financial": "이 정보는 투자 조언이 아닙니다. 투자 결정 전 전문 금융 자문을 받으세요.",
}
 
class DomainSafetyGuardrail(BaseGuardrail):
    def __init__(self, domain: str):
        self.domain = domain
        self.disclaimer = DOMAIN_DISCLAIMERS.get(domain, "")
 
    async def check(self, content: str, context: dict) -> GuardrailResult:
        # 면책 조항이 포함되어 있는지 확인
        if self.domain in ["medical", "legal", "financial"]:
            if self.disclaimer and self.disclaimer not in content:
                modified = f"{content}\n\n---\n{self.disclaimer}"
                return GuardrailResult(
                    passed=False,
                    action="modify",
                    reason="면책 조항 추가",
                    modified_content=modified,
                )
        return GuardrailResult(passed=True, action="allow")