ガードレールは、LLM ジャッジによるスコアに基づいて、LLM アプリケーションの挙動に積極的に介入します。ユーザーに出力が届く前にリアルタイムで動作し、スコアがしきい値を超えた場合にはレスポンスをブロックしたり変更したりできます。ガードレールを使用して、有害なコンテンツをブロックしたり、個人を特定できる情報 (PII) を含むレスポンスをフィルタリングしたり、ユーザーからの攻撃的な入力をブロックしたりできます。
Weave ガードレールはインラインの Weave Scorers を使って、ユーザーからの入力または LLM からの出力を評価し、LLM の応答をリアルタイムで調整します。カスタムスコアラーを設定することも、組み込みスコアラー を使用して、さまざまな目的でコンテンツを評価することもできます。このガイドでは、両方の種類のスコアラーをガードレールとして使用する方法を説明します。
アプリケーションの制御フローを変更せずに本番トラフィックを受動的にスコアリングしたい場合は、代わりに monitors を使用してください。
monitors と異なり、ガードレールはアプリケーションの制御フローに影響するため、コードの変更が必要です。ただし、ガードレールからのすべてのスコアラーの結果は自動的に Weave のデータベースに保存されるため、追加の設定なしでガードレールを monitors としても機能させることができます。スコアラーがもともとどのように使用されていたかに関係なく、過去のスコアラー結果を分析できます。
Weave TypeScript SDK は、ガードレールを設定するために必要なツールをサポートしていません。
- ガードレールのロジックをシンプルかつ高速に保つ
- 頻出する結果をキャッシュする
- 高コストな外部 API 呼び出しを避ける
- 繰り返しの初期化コストを避けるために、メイン関数の外でガードレールを初期化する
メイン関数の外でガードレールを初期化することは、次のような場合に特に重要です。
- スコアラーが ML モデルをロードする場合
- レイテンシが重要なローカル LLM を使用している場合
- スコアラーがネットワーク接続を維持する場合
- 高トラフィックなアプリケーションの場合
例: 組み込みのモデレーションスコアラーを使ってガードレールを作成する
generate_safe_response()) に渡され、この関数内で OpenAIModerationScorer を使用して LLM の元の応答をチェックします。関数のロジックは 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 環境変数を使用する
# モデレーションスコアラーを初期化する
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)
# ユーザーに返す前にモデレーションスコアラーを適用する
score = await call.apply_scorer(moderation_scorer)
print("This is the score object:", score)
# コンテンツがフラグ付きかどうかを確認する
if not score.result.get("passed", True):
categories = score.result.get("categories", {})
flagged_categories = list(categories.keys()) if categories else []
print(f"Content blocked. Flagged categories: {flagged_categories}")
return "I'm sorry, I can't provide that response due to content policy restrictions."
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"\nPrompt: {prompt}")
response = asyncio.run(generate_safe_response(prompt))
print(f"Response: {response}")
{ground_truth} に基づいて {output} が正確かどうかを評価してください。」のように記述できます。詳細については、prompt variables を参照してください。
例：カスタムスコアラーを使用してガードレールを作成する
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パターンを確認します。
Returns:
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")
# 電話番号パターン（US形式）
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
}
# パフォーマンス最適化のため、スコアラーを関数の外で初期化する
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検出スコアラーを適用する
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 がクライアントを作成します。リージョンを指定するには、スコアラーの 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
プラットフォーム
W&B Models
W&B Weave
W&B 推論
W&B 学習
リリースノート