가드레일 설정

import weave
import openai
from weave.scorers import OpenAIModerationScorer
import asyncio

# Weave 초기화
weave.init("your-team-name/your-project-name")

# OpenAI 클라이언트 초기화
client = openai.OpenAI()  # OPENAI_API_KEY 환경 변수 사용

# 모더레이션 scorer 초기화
moderation_scorer = OpenAIModerationScorer()

# OpenAI에 프롬프트 전송
@weave.op
def generate_response(prompt: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": prompt}
        ],
        max_tokens=200
    )
    return response.choices[0].message.content

# 가드레일 함수가 응답의 유해성을 검사
async def generate_safe_response(prompt: str) -> str:
    """콘텐츠 모더레이션 가드레일을 적용하여 응답을 생성합니다."""
    # 결과와 Call 객체를 모두 가져옴
    result, call = generate_response.call(prompt)
    
    # 사용자에게 반환하기 전에 모더레이션 scorer 적용
    score = await call.apply_scorer(moderation_scorer)
    print("score 객체:", score)
    
    # 콘텐츠 플래그 여부 확인
    if not score.result.get("passed", True): 
        categories = score.result.get("categories", {})
        flagged_categories = list(categories.keys()) if categories else []
        print(f"콘텐츠가 차단되었습니다. 플래그 처리된 카테고리: {flagged_categories}")
        return "죄송합니다. 콘텐츠 정책 제한으로 인해 해당 응답을 제공할 수 없습니다."
    
    return result

# 예제 실행
if __name__ == "__main__":
    
    prompts = [
        "What's the capital of France?",
        "Tell me a funny fact about dogs.",
    ]
    
    for prompt in prompts:
        print(f"\n프롬프트: {prompt}")
        response = asyncio.run(generate_safe_response(prompt))
        print(f"응답: {response}")

import weave
import openai
import re
import asyncio
from weave import Scorer

weave.init("your-team-name/your-project-name")

client = openai.OpenAI()

class PIIDetectionScorer(Scorer):
    """LLM 출력에서 PII를 감지하여 데이터 유출을 방지합니다."""
    
    @weave.op
    def score(self, output: str) -> dict:
        """
        출력에서 일반적인 PII 패턴을 확인합니다.
        
        반환값:
            dict: 'passed' (bool) 및 'detected_types' (list) 포함
        """
        detected_types = []
        
        # 이메일 패턴
        if re.search(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', output):
            detected_types.append("email")
        
        # 전화번호 패턴 (미국 형식)
        if re.search(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', output):
            detected_types.append("phone")
        
        # SSN 패턴
        if re.search(r'\b\d{3}-\d{2}-\d{4}\b', output):
            detected_types.append("ssn")
        
        return {
            "passed": len(detected_types) == 0,
            "detected_types": detected_types
        }

# 최적의 성능을 위해 함수 외부에서 scorer 초기화
pii_scorer = PIIDetectionScorer()

@weave.op
def generate_response(prompt: str) -> str:
    """LLM을 사용하여 응답을 생성합니다."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": prompt}
        ],
        max_tokens=200
    )
    return response.choices[0].message.content

async def generate_safe_response(prompt: str) -> str:
    """PII 감지 가드레일을 적용하여 응답을 생성합니다."""
    result, call = generate_response.call(prompt)
    
    # PII 감지 scorer 적용
    score = await call.apply_scorer(pii_scorer)
    
    # PII가 감지된 경우 응답 차단
    if not score.result.get("passed", True):
        detected_types = score.result.get("detected_types", [])
        return f"I cannot provide a response that may contain sensitive information (detected: {', '.join(detected_types)})."
    
    return result

# 사용 예시
if __name__ == "__main__":
    prompts = [
        "What's the weather like today?",
        "Can you help me contact someone at john.doe@example.com?",
        "Tell me about machine learning.",
    ]
    
    for prompt in prompts:
        print(f"\nPrompt: {prompt}")
        response = asyncio.run(generate_safe_response(prompt))
        print(f"Response: {response}")