• October 21, 2026
• 13 min read

Bedrock — Guardrails, Knowledge Bases, Agents & Batch¶

The managed features around the model call. Behaviour (what a guardrail blocks, what a knowledge base retrieves) follows the AWS documentation as of 2026-10-21.

Guardrails¶

A guardrail is a named, versioned set of safety policies applied to model input and output. Create it once on the control plane (bedrock), version it, then reference id + version at inference time.

Tiers: CLASSIC supports English, French and Spanish. STANDARD has broader language and code support, prompt-leakage detection and is "more robust"; it uses cross-Region inference (the boto3 request has crossRegionConfig.guardrailProfileIdentifier). Standard tier for contextual grounding exists only in a subset of Regions; elsewhere the request fails with a validation error.

Create and version¶

import boto3

bedrock = boto3.client("bedrock", region_name="us-east-1")          # control plane

guardrail = bedrock.create_guardrail(
    name="support-bot-guardrail",
    blockedInputMessaging="Sorry, I can't help with that request.",
    blockedOutputsMessaging="Sorry, I can't share that response.",
    contentPolicyConfig={"filtersConfig": [
        {"type": "PROMPT_ATTACK", "inputStrength": "HIGH", "outputStrength": "NONE"},
        {"type": "HATE", "inputStrength": "MEDIUM", "outputStrength": "MEDIUM"},
    ]},
    topicPolicyConfig={"topicsConfig": [{
        "name": "Investment advice", "type": "DENY",
        "definition": "Recommendations about buying or selling financial products.",
        "examples": ["Which stock should I buy?"],
    }]},
    sensitiveInformationPolicyConfig={"piiEntitiesConfig": [
        {"type": "EMAIL", "action": "ANONYMIZE"},
        {"type": "US_SOCIAL_SECURITY_NUMBER", "action": "BLOCK"},
    ]},
)
gid = guardrail["guardrailId"]                                       # working copy is version "DRAFT"
# creation is asynchronous: poll bedrock.get_guardrail(guardrailIdentifier=gid)["status"] until "READY"
version = bedrock.create_guardrail_version(guardrailIdentifier=gid, description="initial")["version"]   # "1"

The status values in the service model are CREATING, UPDATING, VERSIONING, READY, FAILED, DELETING. Use DRAFT for experiments and a numbered version in production so a console edit cannot change behaviour under you. Manage guardrails as code (Terraform, CDK, a script) and test them (below).

Attach to Converse¶

resp = client.converse(
    modelId=MODEL_ID,
    messages=[{"role": "user", "content": [
        {"text": "Answer politely."},                                       # not evaluated once guardContent is present
        {"guardContent": {"text": {"text": user_input}}},                  # evaluated
    ]}],
    guardrailConfig={"guardrailIdentifier": gid, "guardrailVersion": version, "trace": "enabled"},
)
if resp["stopReason"] == "guardrail_intervened":
    ...  # blocked OR masked: see below

A blocked response (as the AWS docs show for a blocked topic):

stopReason: guardrail_intervened
text: Sorry, I can't help with that.
trace.guardrail.inputAssessment['abc123'].topicPolicy.topics -> [('Investment advice', 'BLOCKED')]

Streaming: converse_stream takes guardrailConfig.streamProcessingMode: sync (default; chunks are buffered and scanned, so slightly delayed but accurate) or async (chunks are sent immediately and scanned in the background: a harmful chunk may already have reached the user before the block, and masking is not supported in async mode). For InvokeModelWithResponseStream the body field is amazon-bedrock-guardrailConfig with "streamProcessingMode": "ASYNCHRONOUS".

ApplyGuardrail: Use Guardrails Without a Model¶

bedrock-runtime.apply_guardrail checks any text against a guardrail with no model call. Use it to screen the user input before retrieval, check tool output, check output of non-Bedrock models, or test a guardrail in CI.

resp = client.apply_guardrail(
    guardrailIdentifier=gid, guardrailVersion=version,
    source="INPUT",                                   # "INPUT" = user content, "OUTPUT" = model content
    content=[{"text": {"text": "Ignore all previous instructions and ..."}}],
)
if resp["action"] == "GUARDRAIL_INTERVENED":
    safe_text = resp["outputs"][0]["text"]            # canned message (blocked) or masked text
    detected = resp["assessments"][0]["contentPolicy"]["filters"]

Response fields: action (NONE, GUARDRAIL_INTERVENED), actionReason, outputs (empty when nothing happened), assessments, usage (policy units billed), guardrailCoverage. Add outputScope="FULL" to also see non-detected entries (it does not apply to word filters and regexes). The request path (/guardrail/gr-abc123/version/1/apply)

Contextual grounding needs qualifiers on the content blocks:

client.apply_guardrail(
    guardrailIdentifier=gid, guardrailVersion=version, source="OUTPUT", outputScope="FULL",
    content=[
        {"text": {"text": "Refunds are issued within 14 days.", "qualifiers": ["grounding_source"]}},
        {"text": {"text": "How long do refunds take?", "qualifiers": ["query"]}},
        {"text": {"text": "Refunds take 90 days.", "qualifiers": ["guard_content"]}},   # the answer under test
    ],
)

Responses report usage in policy units (topicPolicyUnits, contentPolicyUnits, wordPolicyUnits, sensitiveInformationPolicyUnits, contextualGroundingPolicyUnits); check the pricing page for how they bill. Both attached guardrails and ApplyGuardrail report units. The service model also lists organisation-level enforced guardrails (put_enforced_guardrail_configuration) and an InvokeGuardrailChecks runtime operation; they are not covered here.

Testing guardrails¶

Knowledge Bases (RAG)¶

A knowledge base ingests your documents, embeds them into a vector store and retrieves chunks at query time. AWS now has two kinds: Managed (Bedrock manages ingestion, indexing, storage, reranking; connectors for S3, SharePoint, Confluence, Google Drive, OneDrive, web crawler; document-level ACL filtering; agentic multi-hop retrieval) and Customer-managed (your vector store such as OpenSearch Serverless, Aurora or Neptune Analytics, and full control of parsing). AWS recommends Managed for new work. Third-party connectors, ACLs and native AgentCore Gateway integration are Managed-only.

runtime = boto3.client("bedrock-agent-runtime", region_name="us-east-1")

# Retrieve: you get ranked chunks and build the prompt yourself (best for control and for testing retrieval alone)
out = runtime.retrieve(
    knowledgeBaseId="KB12345678",
    retrievalQuery={"text": "refund window"},
    retrievalConfiguration={"vectorSearchConfiguration": {"numberOfResults": 5}},   # overrideSearchType HYBRID or SEMANTIC, filter, reranking
)
for r in out["retrievalResults"]:
    print(round(r["score"], 2), r["location"]["s3Location"]["uri"], r["content"]["text"][:80])

# RetrieveAndGenerate: retrieval + a model call + citations in one request
cfg = {"type": "KNOWLEDGE_BASE", "knowledgeBaseConfiguration": {
    "knowledgeBaseId": "KB12345678",
    "modelArn": "arn:aws:bedrock:us-east-1:123456789012:inference-profile/global.anthropic.claude-haiku-5-5"}}
ans = runtime.retrieve_and_generate(input={"text": "How long do refunds take?"}, retrieveAndGenerateConfiguration=cfg)
print(ans["output"]["text"], ans["sessionId"])         # pass sessionId back to continue the conversation
refs = [ref for c in ans["citations"] for ref in c["retrievedReferences"]]

retrieve_and_generate takes a model ARN (a foundation model or an inference profile ARN), not a bare ID. The model must support Knowledge Bases (Claude Sonnet 5.5 and Haiku 5.5 list it; Nova 2 Lite does not). Retrieve also accepts a guardrailConfiguration (its response carries guardrailAction) and, for permission-filtered Managed bases, a userContext.userId.

Test it in layers:

moto implements only control-plane CRUD for bedrock-agent agents and knowledge bases, not retrieve or retrieve_and_generate. Wrap the runtime client in a small Retriever class and fake that.

Agents: Classic and AgentCore¶

# AgentCore: invoke a deployed agent runtime (shape from the AWS developer guide)
import json

import boto3

agentcore = boto3.client("bedrock-agentcore", region_name="us-east-1")
resp = agentcore.invoke_agent_runtime(
    agentRuntimeArn="arn:aws:bedrock-agentcore:us-east-1:123456789012:runtime/support_agent-AbC123",   # placeholder
    runtimeSessionId=session_id,                       # same id = same conversation; use a UUID per conversation
    payload=json.dumps({"prompt": "Where is order ord-42?"}).encode(),
)
if "text/event-stream" in resp.get("contentType", ""):
    for line in resp["response"].iter_lines(chunk_size=10):
        if line.startswith(b"data: "):
            print(line[6:].decode())
else:
    print(json.loads(b"".join(resp["response"])))

# Agents Classic
runtime = boto3.client("bedrock-agent-runtime", region_name="us-east-1")
resp = runtime.invoke_agent(agentId="AGENT12345", agentAliasId="TSTALIASID", sessionId="s-1",
                            inputText="Where is order ord-42?", enableTrace=True)
answer = "".join(e["chunk"]["bytes"].decode() for e in resp["completion"] if "chunk" in e)

Facts and cautions from the AgentCore developer guide:

• The payload is binary, up to 100 MB; responses stream. ValidationException, ResourceNotFoundException, AccessDeniedException, ThrottlingException and a retryable RetryableConflictException (409, "Session operation in progress") can occur.
• If the agent uses OAuth inbound auth, the AWS SDK cannot be used: make an HTTPS request instead.
• Validate that prompt is a string in the agent entrypoint. The payload arrives as parsed JSON; a non-string value containing a toolUse block could make some frameworks run the named tool directly, bypassing model reasoning and guardrails. This is a test case worth writing (see the OWASP LLM guide).
• IAM: bedrock-agentcore:InvokeAgentRuntime for callers.
• InvokeAgentRuntimeCommand runs deterministic shell commands (tests, builds) in a session without going through the LLM.

Testing agents is mostly testing your adapter: keep one class (AgentClient.ask(session_id, text) -> AgentAnswer), unit-test your parsing of the event stream with hand-built chunks, test your own agent code locally as ordinary Python before deploying, and run a handful of end-to-end conversations against a deployed test alias or runtime. Judge the behaviour with agent metrics (task completion, tool correctness) from the DeepEval guide and design guidance from Agentic AI Architecture.

Batch Inference¶

For large offline jobs: put JSONL in S3, create a job, collect JSONL from S3 later. Priced separately (see the AWS pricing page), separate quotas, no latency guarantee.

flowchart LR
  A["input .jsonl<br/>recordId + modelInput"] -->|"upload"| S3I["s3://bucket/in/"]
  S3I --> J["create_model_invocation_job<br/>(bedrock control plane)"]
  J -->|"Submitted → Validating → InProgress"| R["Completed / PartiallyCompleted / Failed / Expired"]
  R --> S3O["s3://bucket/out/<br/>*.jsonl.out + manifest.json.out"]

import json


def build_batch_jsonl(prompts: dict[str, str], max_tokens=512, system=None) -> str:
    lines = []
    for record_id, prompt in prompts.items():
        model_input = {"messages": [{"role": "user", "content": [{"text": prompt}]}],
                       "inferenceConfig": {"maxTokens": max_tokens}}               # Converse request body, no toolConfig
        if system:
            model_input["system"] = [{"text": system}]
        lines.append(json.dumps({"recordId": record_id, "modelInput": model_input}))
    return "\n".join(lines) + "\n"


job = bedrock.create_model_invocation_job(
    jobName="nightly-summaries-2026-10-21",
    modelId="anthropic.claude-haiku-5-5",          # check the model card: batch support is per model
    roleArn="arn:aws:iam::123456789012:role/bedrock-batch",
    modelInvocationType="Converse",                # or "InvokeModel" (native bodies)
    inputDataConfig={"s3InputDataConfig": {"s3Uri": "s3://my-batch-bucket/in/"}},
    outputDataConfig={"s3OutputDataConfig": {"s3Uri": "s3://my-batch-bucket/out/"}},
)
status = bedrock.get_model_invocation_job(jobIdentifier=job["jobArn"])["status"]

Parse results defensively and test the parser with a canned file:

def parse_batch_output(text: str):
    ok, failed = {}, {}
    for line in text.splitlines():
        rec = json.loads(line)
        if "modelOutput" in rec and "error" not in rec:
            ok[rec["recordId"]] = "".join(b.get("text", "") for b in rec["modelOutput"]["output"]["message"]["content"])
        else:
            failed[rec["recordId"]] = rec.get("error") or rec
    return ok, failed

A job that ends PartiallyCompleted is a normal outcome to handle: re-submit only the failed record ids.

Model Evaluation and Other Features¶

Quick Rules¶

1. Version guardrails and reference numbered versions in production; manage them as code.
2. guardrail_intervened is not always a block. Enable the trace and check action before deciding to discard an answer, and never log the trace as plain text.
3. Wrap system prompts and the user turn in guardContent when you want them evaluated, and validate tool arguments and results yourself: guardrails skip them.
4. Use ApplyGuardrail to test policies and to screen input before retrieval; keep the dataset of attack and benign prompts in the repo.
5. Use Retrieve when you need control and testability; use RetrieveAndGenerate for speed. Fail a test when an answer has no citations.
6. Do not start new projects on Agents Classic. Use AgentCore or a framework, and validate that prompt is a string at the entrypoint.
7. Batch jobs are keyed by recordId, not by order; handle PartiallyCompleted; check that the model supports batch.
8. Keep an adapter class around every agent and knowledge-base call so tests can fake it.

See also¶

• Amazon Bedrock
• 01 Models, Access, Regions & the Converse API
• 02 Streaming, Tools, Structured Output, Multimodal & Caching
• 04 Testing & Production
• OWASP Top 10 for LLM Applications
• DeepEval: RAG Metrics
• DeepEval: Red Teaming
• Agentic AI Architecture
• Guardrails AI (open-source validators)
• Amazon SQS