Strands Agents SDK 実践 — Guardrails で入出力をフィルタリングする

aws bedrock create-guardrail \
  --name "strands-test-guardrail" \
  --description "Test guardrail for Strands practical series" \
  --content-policy-config '{
    "filtersConfig": [
      {"type": "VIOLENCE", "inputStrength": "HIGH", "outputStrength": "HIGH"},
      {"type": "HATE", "inputStrength": "HIGH", "outputStrength": "HIGH"}
    ]
  }' \
  --topic-policy-config '{
    "topicsConfig": [
      {
        "name": "Investment Advice",
        "definition": "Providing specific investment recommendations or financial advice",
        "examples": ["What stocks should I buy?", "Should I invest in crypto?"],
        "type": "DENY"
      }
    ]
  }' \
  --blocked-input-messaging "Sorry, this request was blocked by guardrails." \
  --blocked-outputs-messaging "Sorry, this response was blocked by guardrails." \
  --region us-east-1

{
    "guardrailId": "7by7u1yvthd8",
    "guardrailArn": "arn:aws:bedrock:us-east-1:123456789012:guardrail/7by7u1yvthd8",
    "version": "DRAFT"
}

aws bedrock create-guardrail-version \
  --guardrail-identifier "7by7u1yvthd8" \
  --region us-east-1

from strands import Agent
from strands.models import BedrockModel
 
GUARDRAIL_ID = "YOUR_GUARDRAIL_ID"  # 自分のガードレール ID に置き換える
 
bedrock_model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
    region_name="us-east-1",
    guardrail_id=GUARDRAIL_ID,
    guardrail_version="1",
    guardrail_trace="enabled",
)
 
agent = Agent(model=bedrock_model, callback_handler=None)
 
result1 = agent("What is the capital of France?")
print(f"Stop reason: {result1.stop_reason}")
print(f"Answer: {result1.message['content'][0]['text']}")
 
agent2 = Agent(model=bedrock_model, callback_handler=None)
result2 = agent2("What stocks should I buy to get rich quickly?")
print(f"\nStop reason: {result2.stop_reason}")
print(f"Answer: {result2.message['content'][0]['text']}")
print(f"\nMessages after block: {len(agent2.messages)}")
for i, msg in enumerate(agent2.messages):
    role = msg['role']
    text = msg['content'][0].get('text', '')[:80]
    print(f"  [{i}] {role:10s}: {text}")

bedrock_client = boto3.client("bedrock-runtime", "us-east-1")
 
def check_user_input(event: MessageAddedEvent) -> None:
    if event.message.get("role") != "user":
        return
    content = "".join(block.get("text", "") for block in event.message.get("content", []))
    if not content:
        return
    try:
        response = bedrock_client.apply_guardrail(
            guardrailIdentifier=GUARDRAIL_ID,
            guardrailVersion=GUARDRAIL_VERSION,
            source="INPUT",
            content=[{"text": {"text": content}}],
        )
        if response.get("action") == "GUARDRAIL_INTERVENED":
            print(f"[SHADOW] WOULD BLOCK INPUT: {content[:60]}...")
            for assessment in response.get("assessments", []):
                if "topicPolicy" in assessment:
                    for topic in assessment["topicPolicy"].get("topics", []):
                        print(f"[SHADOW]   Topic: {topic['name']} -> {topic['action']}")
        else:
            print(f"[SHADOW] INPUT OK: {content[:60]}...")
    except Exception as e:
        print(f"[SHADOW] Error: {e}")
 
def check_output(event: AfterInvocationEvent) -> None:
    if not event.agent.messages or event.agent.messages[-1].get("role") != "assistant":
        return
    content = "".join(
        block.get("text", "") for block in event.agent.messages[-1].get("content", [])
    )
    if not content:
        return
    try:
        response = bedrock_client.apply_guardrail(
            guardrailIdentifier=GUARDRAIL_ID,
            guardrailVersion=GUARDRAIL_VERSION,
            source="OUTPUT",
            content=[{"text": {"text": content}}],
        )
        if response.get("action") == "GUARDRAIL_INTERVENED":
            print(f"[SHADOW] WOULD BLOCK OUTPUT: {content[:60]}...")
        else:
            print(f"[SHADOW] OUTPUT OK")
    except Exception as e:
        print(f"[SHADOW] Error: {e}")