Amazon Bedrock Agents

Instrument LLM calls to AWS Bedrock Agents via the boto3 client using OpenInference.

boto3 provides Python bindings to AWS services, including Bedrock Agents. Calls to these agents can be instrumented using OpenInference, enabling OpenTelemetry-compliant observability. Traces collected using OpenInference can be viewed in Arize.

OpenInference Traces collect telemetry data about the execution of your LLM application. Consider using this instrumentation to understand how Bedrock Agents are calling their own internal action groups, knowledgebases, and models.

Python

To get started instrumenting Bedrock Agents via boto3, we need to install two components: the OpenInference instrumentation for AWS Bedrock, and an OpenTelemetry exporter used to send these traces to Phoenix.

pip install openinference-instrumentation-bedrock
pip install opentelemetry-exporter-otlp
pip install arize-otel

Instrument boto3 prior to initializing a bedrock-runtime client. All clients created after instrumentation will send traces on all calls to invoke_agent.

# Import open-telemetry dependencies
from arize.otel import register

# Setup OTel via our convenience function
tracer_provider = register(
    space_id = "your-space-id", # in app space settings page
    api_key = "your-api-key", # in app space settings page
    project_name = "your-project-name", # name this to whatever you would like
)

# Import the automatic instrumentor from OpenInference
from openinference.instrumentation.bedrock import BedrockInstrumentor

# Start the instrumentor for Bedrock
BedrockInstrumentor().instrument(tracer_provider=tracer_provider)

You can use the following code to test whether your tracing is working.

Note: we are showing examples using both the Converse API as well as the Invoke Model API. The Converse API was introduced in botocore v1.34.116. Please use v1.34.116 or above to utilize converse.

import boto3

# make sure these are set
session = boto3.session.Session(
    aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
    aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
    aws_session_token=os.environ["AWS_SESSION_TOKEN"],
    region_name=os.environ["AWS_REGION_NAME"]
)

AGENT_ID = "your Bedrock Agent id"
AGENT_ALIAS_ID = "your Bedrock Agent alias id"

client = session.client("bedrock-agent-runtime")

session_id = f"default-session1_{int(time.time())}"

attributes = dict(
    inputText="Tell me a joke",
    agentId=AGENT_ID,
    agentAliasId=AGENT_ALIAS_ID,
    sessionId=session_id,
    enableTrace=True,
)
response = client.invoke_agent(**attributes)

# Stream the response
for _, event in enumerate(response["completion"]):
    if "chunk" in event:
        print(event)
        chunk_data = event["chunk"]
        if "bytes" in chunk_data:
            output_text = chunk_data["bytes"].decode("utf8")
            print(output_text)
    elif "trace" in event:
        print(event["trace"])