가드레일 및 모니터 설정하기

# op를 직접 호출하는 경우:
result = generate_text(input)  # op를 호출하는 기본 방법이지만 Call 오브젝트에 접근할 수 없습니다.

# .call() 메소드를 사용하여 결과와 Call 오브젝트를 모두 가져오기:
result, call = generate_text.call(input)  # 이제 Scorer와 함께 call 오브젝트를 사용할 수 있습니다.

import weave
from weave import Scorer

class LengthScorer(Scorer):
    @weave.op
    def score(self, output: str) -> dict:
        """출력 길이를 확인하는 간단한 Scorer입니다."""
        return {
            "length": len(output),
            "is_short": len(output) < 100
        }

@weave.op
def generate_text(prompt: str) -> str:
    return "Hello, world!"

# 결과와 Call 오브젝트를 모두 가져옵니다.
result, call = generate_text.call("Say hello")

# 이제 Scorer를 적용할 수 있습니다.
await call.apply_scorer(LengthScorer())

import weave
from weave import Scorer

@weave.op
def generate_text(prompt: str) -> str:
    """LLM을 사용하여 텍스트를 생성합니다."""
    # 여기에 LLM 생성 로직 추가
    return "Generated response..."

class ToxicityScorer(Scorer):
    @weave.op
    def score(self, output: str) -> dict:
        """
        콘텐츠의 독성 언어를 평가합니다.
        """
        # 여기에 독성 감지 로직 추가
        return {
            "flagged": False,  # 콘텐츠가 독성인 경우 True
            "reason": None     # 플래그가 지정된 경우 선택적 설명
        }

async def generate_safe_response(prompt: str) -> str:
    # 결과와 Call 오브젝트 가져오기
    result, call = generate_text.call(prompt)
    
    # 안전성 확인
    safety = await call.apply_scorer(ToxicityScorer())
    if safety.result["flagged"]:
        return f"해당 콘텐츠를 생성할 수 없습니다: {safety.result['reason']}"
    
    return result

@weave.op
def my_function(foo: str, bar: str) -> str:
    return f"{foo} and {bar}"

import weave
import boto3
from weave.scorers.bedrock_guardrails import BedrockGuardrailScorer

# Weave 초기화
weave.init("my_app")

# 가드레일 Scorer 생성
guardrail_scorer = BedrockGuardrailScorer(
    guardrail_id="your-guardrail-id",  # "your-guardrail-id"를 귀하의 가드레일 ID로 바꿉니다.
    guardrail_version="DRAFT",          # 특정 가드레일 버전을 사용하려면 guardrail_version을 사용하세요.
    source="INPUT",                             # "INPUT" 또는 "OUTPUT"이 될 수 있습니다.
    bedrock_runtime_kwargs={"region_name": "us-east-1"}  # AWS 리전
)

@weave.op
def generate_text(prompt: str) -> str:
    # 여기에 텍스트 생성 로직 추가
    return "Generated text..."

# 안전성 확인으로 가드레일 사용
async def generate_safe_text(prompt: str) -> str:
    result, call = generate_text.call(prompt)
    
    # 가드레일 적용
    score = await call.apply_scorer(guardrail_scorer)
    
    # 콘텐츠가 가드레일을 통과했는지 확인
    if not score.result.passed:
        # 수정된 출력이 있는 경우 사용
        if score.result.metadata.get("modified_output"):
            return score.result.metadata["modified_output"]
        return "콘텐츠 정책 제한으로 인해 해당 콘텐츠를 생성할 수 없습니다."
    
    return result

@weave.op
def generate_styled_text(prompt: str, style: str, temperature: float) -> str:
    """특정 스타일로 텍스트를 생성합니다."""
    return "요청된 스타일로 생성된 텍스트입니다..."

class StyleScorer(Scorer):
    @weave.op
    def score(self, output: str, prompt: str, style: str) -> dict:
        """
        출력이 요청된 스타일과 일치하는지 평가합니다.
        
        Args:
            output: 생성된 텍스트 (자동 제공)
            prompt: 원본 프롬프트 (함수 입력과 일치)
            style: 요청된 스타일 (함수 입력과 일치)
        """
        return {
            "style_match": 0.9,  # 요청된 스타일과 얼마나 잘 일치하는지
            "prompt_relevance": 0.8  # 프롬프트와 얼마나 관련이 있는지
        }

# 사용 예시
async def generate_and_score():
    # 스타일과 함께 텍스트 생성
    result, call = generate_styled_text.call(
        prompt="이야기를 하나 써줘",
        style="noir",
        temperature=0.7
    )
    
    # 결과 스코어링
    score = await call.apply_scorer(StyleScorer())
    print(f"스타일 일치 점수: {score.result['style_match']}")

@weave.op
def generate_text(user_input: str):  # 'user_input' 사용
    return process(user_input)

class QualityScorer(Scorer):
    @weave.op
    def score(self, output: str, prompt: str):  # 'prompt' 기대
        """응답 품질을 평가합니다."""
        return {"quality_score": evaluate_quality(prompt, output)}

result, call = generate_text.call(user_input="안녕")

# 'prompt' 파라미터를 'user_input'에 매핑
scorer = QualityScorer(column_map={"prompt": "user_input"})
await call.apply_scorer(scorer)

class ReferenceScorer(Scorer):
    @weave.op
    def score(self, output: str, reference_answer: str):
        """출력을 기준 답안과 비교합니다."""
        similarity = compute_similarity(output, reference_answer)
        return {"matches_reference": similarity > 0.8}

# 기준 답안을 추가 파라미터로 제공
await call.apply_scorer(
    ReferenceScorer(),
    additional_scorer_kwargs={
        "reference_answer": "지구는 태양 주위를 궤도에 따라 돕니다."
    }
)

result, call = generate_text.call(input)
score = await call.apply_scorer(MyScorer())

scorer = MyScorer()
score = scorer.score(output="some text")

@weave.op
def generate_text(prompt: str) -> str:
    return generate_response(prompt)

async def generate_with_sampling(prompt: str) -> str:
    result, call = generate_text.call(prompt)
    
    # 호출의 10%만 모니터링
    if random.random() < 0.1:
        await call.apply_scorer(ToxicityScorer())
        await call.apply_scorer(QualityScorer())
    
    return result

async def evaluate_comprehensively(call):
    await call.apply_scorer(ToxicityScorer())
    await call.apply_scorer(QualityScorer())
    await call.apply_scorer(LatencyScorer())

import weave
from weave import Scorer
import asyncio
import random
from typing import Optional

class ToxicityScorer(Scorer):
    def __init__(self):
        # 여기에 비용이 많이 드는 리소스 초기화
        self.model = load_toxicity_model()
    
    @weave.op
    async def score(self, output: str) -> dict:
        """콘텐츠의 독성 언어를 확인합니다."""
        try:
            result = await self.model.evaluate(output)
            return {
                "flagged": result.is_toxic,
                "reason": result.explanation if result.is_toxic else None
            }
        except Exception as e:
            # 에러 로그를 남기고 보수적인 행동을 기본값으로 함
            print(f"Toxicity check failed: {e}")
            return {"flagged": True, "reason": "안전 확인 서비스를 사용할 수 없습니다."}

class QualityScorer(Scorer):
    @weave.op
    async def score(self, output: str, prompt: str) -> dict:
        """응답 품질과 관련성을 평가합니다."""
        return {
            "coherence": evaluate_coherence(output),
            "relevance": evaluate_relevance(output, prompt),
            "grammar": evaluate_grammar(output)
        }

# 모듈 수준에서 Scorer 초기화 (선택적 최적화)
toxicity_guard = ToxicityScorer()
quality_monitor = QualityScorer()
relevance_monitor = RelevanceScorer()

@weave.op
def generate_text(
    prompt: str,
    style: Optional[str] = None,
    temperature: float = 0.7
) -> str:
    """LLM 응답을 생성합니다."""
    # 여기에 LLM 생성 로직 추가
    return "Generated response..."

async def generate_safe_response(
    prompt: str,
    style: Optional[str] = None,
    temperature: float = 0.7
) -> str:
    """안전 점검 및 품질 모니터링이 포함된 응답을 생성합니다."""
    try:
        # 초기 응답 생성
        result, call = generate_text.call(
            prompt=prompt,
            style=style,
            temperature=temperature
        )

        # 안전 점검 적용 (가드레일)
        safety = await call.apply_scorer(toxicity_guard)
        if safety.result["flagged"]:
            return f"해당 콘텐츠를 생성할 수 없습니다: {safety.result['reason']}"

        # 샘플 품질 모니터링 (요청의 10%)
        if random.random() < 0.1:
            # 품질 점검을 병렬로 실행
            await asyncio.gather(
                call.apply_scorer(quality_monitor),
                call.apply_scorer(relevance_monitor)
            )
        
        return result

    except Exception as e:
        # 에러 로그 및 사용자 친화적 메시지 반환
        print(f"Generation failed: {e}")
        return "죄송합니다. 오류가 발생했습니다. 다시 시도해 주세요."

# 사용 예시
async def main():
    # 기본 사용법
    response = await generate_safe_response("이야기를 하나 들려줘")
    print(f"Basic response: {response}")
    
    # 모든 파라미터를 사용한 고급 사용법
    response = await generate_safe_response(
        prompt="이야기를 하나 들려줘",
        style="noir",
        temperature=0.8
    )
    print(f"Styled response: {response}")