가드레일은 LLM 기반 평가자의 점수를 바탕으로 LLM 애플리케이션의 동작에 적극적으로 개입합니다. 출력이 사용자에게 전달되기 전에 실시간으로 작동하며, 점수가 임계값을 초과하면 응답을 차단하거나 수정할 수 있습니다. 가드레일을 사용하면 유해한 콘텐츠를 차단하고, 응답에서 개인 식별 정보(PII)을 필터링하거나, 사용자의 악의적 입력을 차단할 수 있습니다.
Weave 가드레일은 인라인 Weave Scorers를 사용해 사용자 입력이나 LLM 출력을 평가하고, LLM의 응답을 실시간으로 조정합니다. 다양한 목적에 맞춰 콘텐츠를 평가하도록 맞춤형 scorer를 설정하거나 기본 제공 scorer를 사용할 수 있습니다. 이 가이드에서는 두 유형의 scorer를 모두 가드레일로 사용하는 방법을 설명합니다.
애플리케이션의 제어 흐름을 수정하지 않고 프로덕션 트래픽을 수동적으로 평가하려면 모니터를 대신 사용하세요.
모니터와 달리 가드레일은 애플리케이션의 제어 흐름에 영향을 미치므로 코드 변경이 필요합니다. 하지만 가드레일의 모든 scorer 결과는 Weave 데이터베이스에 자동으로 저장되므로, 추가 설정 없이도 가드레일은 모니터 역할도 합니다. 원래 어떤 방식으로 사용했는지와 관계없이 과거 scorer 결과를 분석할 수 있습니다.
Weave TypeScript SDK는 가드레일을 설정하는 데 필요한 도구를 지원하지 않습니다.
- 가드레일 로직을 단순하고 빠르게 유지하기
- 자주 사용하는 결과 캐싱하기
- 비용이 큰 외부 API 호출 피하기
- 반복적인 초기화 비용을 피할 수 있도록 메인 함수 밖에서 가드레일 초기화하기
특히 다음과 같은 경우에는 메인 함수 밖에서 가드레일을 초기화하는 것이 중요합니다.
- scorer가 ML 모델을 로드하는 경우
- 지연 시간이 중요한 로컬 LLM을 사용하는 경우
- scorer가 네트워크 연결을 유지하는 경우
- 트래픽이 많은 애플리케이션을 운영하는 경우
예시: 내장 moderation scorer를 사용해 가드레일 생성하기
OpenAIModerationScorer를 사용해 LLM의 원래 응답을 확인하는 가드레일 함수(generate_safe_response())로 전달됩니다. 이후 함수 로직은 OpenAI의 평가 응답에서 passed 필드의 불리언 값을 확인하고, 이 값에 따라 애플리케이션이 어떻게 응답할지가 결정됩니다.
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}")
{ground_truth}를 기준으로 {output}이 정확한지 평가하세요.” 자세한 내용은 프롬프트 변수를 참조하세요.
예시: 맞춤형 scorer를 사용해 가드레일 만들기
generate_safe_response 함수는 맞춤형 PIIDetectionScorer를 적용합니다.
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}")
Weave를 AWS Bedrock Guardrails와 통합하기
BedrockGuardrailScorer는 AWS Bedrock Guardrails를 사용해 구성된 정책에 따라 콘텐츠를 감지하고 필터링합니다.
Bedrock Guardrails 인테그레이션을 설정하기 전에 다음이 필요합니다.
별도의 Bedrock 클라이언트를 직접 만들 필요는 없습니다. Weave가 대신 생성해 줍니다. 리전을 지정하려면 scorer의 bedrock_runtime_kwargs 매개변수에 리전 값을 전달하세요.
AWS Bedrock에서 가드레일을 만드는 예시는 Bedrock guardrails notebook을 참조하세요.
다음 예제에서는 사용자에게 결과를 반환하기 전에 텍스트 생성 결과가 AWS Bedrock Guardrails 정책에 부합하는지 확인합니다:
import weave
from weave.scorers.bedrock_guardrails import BedrockGuardrailScorer
weave.init("your-team-name/your-project-name")
guardrail_scorer = BedrockGuardrailScorer(
guardrail_id="your-guardrail-id",
guardrail_version="DRAFT",
source="INPUT",
bedrock_runtime_kwargs={"region_name": "us-east-1"}
)
@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
