[アップデート] Amazon Bedrock Guardrailsのコンテキストグラウンディングチェックに Standard ティアが追加されたので試してみた

import boto3

client = boto3.client("bedrock", region_name="ap-northeast-1")

FILTERS = [
    {"type": "GROUNDING", "threshold": 0.7, "action": "BLOCK", "enabled": True},
    {"type": "RELEVANCE", "threshold": 0.7, "action": "BLOCK", "enabled": True},
]

classic = client.create_guardrail(
    name="grounding-classic",
    blockedInputMessaging="この質問にはお答えできません。",
    blockedOutputsMessaging="参照資料に基づいた回答を生成できませんでした。",
    contextualGroundingPolicyConfig={"filtersConfig": FILTERS},
)

standard = client.create_guardrail(
    name="grounding-standard",
    blockedInputMessaging="この質問にはお答えできません。",
    blockedOutputsMessaging="参照資料に基づいた回答を生成できませんでした。",
    contextualGroundingPolicyConfig={
        "filtersConfig": FILTERS,
        "tierConfig": {"tierName": "STANDARD"},  # 省略するとCLASSIC
    },
    crossRegionConfig={"guardrailProfileIdentifier": "apac.guardrail.v1:0"},
)

print("Classic :", classic["guardrailId"])
print("Standard:", standard["guardrailId"])

import boto3

client = boto3.client("bedrock-runtime", region_name="ap-northeast-1")

GUARDRAILS = {
    "Classic": "<ClassicのガードレールID>",
    "Standard": "<StandardのガードレールID>",
}

SOURCE = """出張旅費規程
- 宿泊費の上限は1泊12,000円です。
- 東京都・大阪府に宿泊する場合、上限は1泊15,000円です。
- 精算の申請は出張終了日から14日以内に行ってください。"""

QUERY = "来週、大阪に2泊の出張があります。宿泊費はいくらまで精算できますか？"

ANSWERS = {
    "言い換え": "大阪府に宿泊する場合、宿泊費の上限は1泊15,000円です。",
    "会話・推論": "大阪出張ですね！大阪府は1泊15,000円が上限なので、2泊なら合計30,000円まで精算できます。出張から戻ったら、14日以内に申請するのを忘れないようにしてくださいね。",
    "誤った金額": "大阪府は1泊20,000円が上限なので、2泊なら合計40,000円まで精算できます。",
    "質問に答えていない": "精算の申請は出張終了日から14日以内に行ってください。",
}

def check(guardrail_id: str, answer: str):
    res = client.apply_guardrail(
        guardrailIdentifier=guardrail_id,
        guardrailVersion="DRAFT",
        source="OUTPUT",
        content=[
            {"text": {"text": SOURCE, "qualifiers": ["grounding_source"]}},
            {"text": {"text": QUERY, "qualifiers": ["query"]}},
            {"text": {"text": answer}},  # 評価する回答
        ],
    )
    filters = res["assessments"][0]["contextualGroundingPolicy"]["filters"]
    scores = {f["type"]: f["score"] for f in filters}
    return res["action"], scores

for label, answer in ANSWERS.items():
    for tier, guardrail_id in GUARDRAILS.items():
        action, scores = check(guardrail_id, answer)
        print(f"{label:<10} {tier:<8} GROUNDING={scores['GROUNDING']:.2f} RELEVANCE={scores['RELEVANCE']:.2f} {action}")

import boto3

client = boto3.client("bedrock-runtime", region_name="ap-northeast-1")

MODEL_ID = "jp.anthropic.claude-haiku-4-5-20251001-v1:0"
GUARDRAIL = {"guardrailIdentifier": "<StandardのガードレールID>", "guardrailVersion": "DRAFT", "trace": "enabled"}
SYSTEM = [{"text": "社内規程アシスタントです。渡された規程だけに基づいて、会話の流れを踏まえて日本語で1〜2文で簡潔に答えてください。"}]
QUERY_LIMIT = 1000  # queryは全ターン合計で1,000文字まで

SOURCE = """出張旅費規程
- 宿泊費の上限は1泊12,000円です。
- 東京都・大阪府に宿泊する場合、上限は1泊15,000円です。
- 精算の申請は出張終了日から14日以内に行ってください。"""

LONG_MEMO = "【出張メモ】" + "新大阪駅近くのホテルを候補にしていて、朝食付きのプランにするか迷っています。" * 30
TURNS = [
    "来週、大阪に2泊の出張があります。",
    "それなら宿泊費はいくらまで精算できますか？",
    LONG_MEMO + "\nあと、申請はいつまでにすればいいですか？",
]

def guard(text, qualifier):
    """テキストをguardContentで囲み、参照資料（grounding_source）か質問（query）かを指定する"""
    return {"guardContent": {"text": {"text": text, "qualifiers": [qualifier]}}}

def build_query(user_texts):
    """新しい発言から順に足し、1,000文字に収まらない古い発言は落とす"""
    query = ""
    for text in reversed(user_texts):
        candidate = f"{text}\n{query}" if query else text
        if len(candidate) > QUERY_LIMIT:
            return query or text[-QUERY_LIMIT:]  # 最新の発言だけで超えるときは末尾を残す
        query = candidate
    return query

def scores(res):
    for items in res.get("trace", {}).get("guardrail", {}).get("outputAssessments", {}).values():
        for a in items:
            f = a.get("contextualGroundingPolicy", {}).get("filters", [])
            return " ".join(f"{x['type']}={x['score']:.2f}" for x in f)
    return "-"

def chat():
    history, user_texts = [], []
    for user_text in TURNS:
        user_texts.append(user_text)
        query = build_query(user_texts)
        # タグは最新の発言にだけ付け、モデルには発言をそのまま渡す
        latest = {"role": "user", "content": [
            guard(SOURCE, "grounding_source"), guard(query, "query"), {"text": user_text}]}
        res = client.converse(modelId=MODEL_ID, system=SYSTEM, messages=history + [latest],
                              guardrailConfig=GUARDRAIL, inferenceConfig={"maxTokens": 200})
        answer = res["output"]["message"]["content"][0]["text"]
        print(f"[発言{len(user_text)}字 / query{len(query)}字] {res['stopReason']} {scores(res)}\n  → {answer}")
        # 履歴にはタグなしのテキストで残す
        history += [{"role": "user", "content": [{"text": user_text}]},
                    {"role": "assistant", "content": [{"text": answer}]}]

if __name__ == "__main__":
    chat()