Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.
# Libreria di codici
Questa sezione fornisce esempi di codice per le operazioni più comuni di Amazon Nova che utilizzano l'API Converse o l' InvokeModel API.
## Esempi di API Converse
### Richiesta di base
Invia una richiesta di testo di base ai modelli Amazon Nova utilizzando l'API Converse.
------
#### [ Non-streaming ]
```
import boto3
from botocore.config import Config
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(read_timeout=3600),
)
# Invoke the model
response = bedrock.converse(
modelId="us.amazon.nova-2-lite-v1:0",
messages=[
{
"role": "user",
"content": [{"text": "Write a short story. End the story with 'THE END'."}],
}
],
system=[{"text": "You are a children's book author."}], # Optional
inferenceConfig={ # These parameters are optional
"maxTokens": 1500,
"temperature": 0.7,
"topP": 0.9,
"stopSequences": ["THE END"],
},
additionalModelRequestFields={ # These parameters are optional
"inferenceConfig": {
"topK": 50,
}
},
)
# Extract the text response
content_list = response["output"]["message"]["content"]
for content in content_list:
if "text" in content:
print(content["text"])
```
------
#### [ Streaming ]
```
import boto3
from botocore.config import Config
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(connect_timeout=3600, read_timeout=3600),
)
# Invoke the model
response = bedrock.converse_stream(
modelId="us.amazon.nova-2-lite-v1:0",
messages=[
{
"role": "user",
"content": [{"text": "Write a short story. End the story with 'THE END'."}],
}
],
system=[{"text": "You are a children's book author."}], # Optional
inferenceConfig={ # These parameters are optional
"maxTokens": 1500,
"temperature": 0.7,
"topP": 0.9,
"stopSequences": ["THE END"],
},
additionalModelRequestFields={ # These parameters are optional
"inferenceConfig": {
"topK": 50,
}
},
)
# Handle streaming events
for event in response["stream"]:
if "contentBlockDelta" in event:
delta = event["contentBlockDelta"]["delta"]
if "text" in delta:
print(delta["text"], end="", flush=True)
```
------
### Input multimodale tramite asset incorporato
Elabora i contenuti multimodali incorporando dati di documenti, immagini, video o audio direttamente nella richiesta. Questo esempio utilizza dati di immagine. Per dettagli sulla struttura dei contenuti per altre modalità, consulta [ContentBlock i dettagli nella documentazione dell'API Amazon Bedrock](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ContentBlock.html).
------
#### [ Non-streaming ]
```
import boto3
from botocore.config import Config
# Read a document, image, video, or audio file
with open("sample_image.png", "rb") as image_file:
binary_data = image_file.read()
data_format = "png"
# Define message with image
messages = [
{
"role": "user",
"content": [
{
"image": {
"format": data_format,
"source": {
"bytes": binary_data # For Invoke API, encode as Base64 string
},
},
},
{"text": "Provide a brief caption for this asset."},
],
}
]
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(read_timeout=3600),
)
# Invoke model
response = bedrock.converse(
modelId="us.amazon.nova-2-lite-v1:0",
messages=messages,
)
# Extract the text response
content_list = response["output"]["message"]["content"]
for content in content_list:
if "text" in content:
print(content["text"])
```
------
#### [ Streaming ]
```
import boto3
from botocore.config import Config
# Read a document, image, video, or audio file
with open("sample_image.png", "rb") as image_file:
binary_data = image_file.read()
data_format = "png"
# Define message with image
messages = [
{
"role": "user",
"content": [
{
"image": {
"format": data_format,
"source": {
"bytes": binary_data # For Invoke API, encode as Base64 string
},
},
},
{"text": "Provide a brief caption for this asset."},
],
}
]
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(connect_timeout=3600, read_timeout=3600),
)
# Invoke model with streaming
response = bedrock.converse_stream(
modelId="us.amazon.nova-2-lite-v1:0",
messages=messages,
)
# Handle streaming events
for event in response["stream"]:
if "contentBlockDelta" in event:
delta = event["contentBlockDelta"]["delta"]
if "text" in delta:
print(delta["text"], end="", flush=True)
```
------
### Ingresso multimodale tramite URI S3
Elabora contenuti multimodali facendo riferimento a documenti, immagini, video o file audio archiviati in S3. Questo esempio utilizza un riferimento a un'immagine. Per dettagli sulla struttura dei contenuti per altre modalità, consulta [ContentBlock i dettagli nella documentazione dell'API Amazon Bedrock](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ContentBlock.html).
------
#### [ Non-streaming ]
```
import boto3
from botocore.config import Config
# Define message with image
messages = [
{
"role": "user",
"content": [
{
"image": {
"format": "png",
"source": {
"s3Location": {
"uri": "s3://path/to/your/asset",
# "bucketOwner": "" # Optional
}
},
},
},
{"text": "Provide a brief caption for this asset."},
],
}
]
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(read_timeout=3600),
)
# Invoke model
response = bedrock.converse(
modelId="us.amazon.nova-2-lite-v1:0",
messages=messages,
)
# Extract the text response
content_list = response["output"]["message"]["content"]
for content in content_list:
if "text" in content:
print(content["text"])
```
------
#### [ Streaming ]
```
import boto3
from botocore.config import Config
# Define message with image
messages = [
{
"role": "user",
"content": [
{
"image": {
"format": "png",
"source": {
"s3Location": {
"uri": "s3://path/to/your/asset",
# "bucketOwner": "" # Optional
}
},
},
},
{"text": "Provide a brief caption for this asset."},
],
}
]
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(connect_timeout=3600, read_timeout=3600),
)
# Invoke model with streaming
response = bedrock.converse_stream(
modelId="us.amazon.nova-2-lite-v1:0",
messages=messages,
)
# Handle streaming events
for event in response["stream"]:
if "contentBlockDelta" in event:
delta = event["contentBlockDelta"]["delta"]
if "text" in delta:
print(delta["text"], end="", flush=True)
```
------
### Pensiero esteso (ragionamento)
Abilita il pensiero esteso per attività complesse di risoluzione di problemi.
------
#### [ Non-streaming ]
```
import boto3
from botocore.config import Config
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(read_timeout=3600),
)
# Invoke the model
response = bedrock.converse(
modelId="us.amazon.nova-2-lite-v1:0",
messages=[
{
"role": "user",
"content": [
{
"text": 'How many capital letters appear in the following passage. Your response must include only the number: "Wilfred ordered an anvil from ACME. Shipping was expensive."'
}
],
}
],
additionalModelRequestFields={
"reasoningConfig": {
"type": "enabled",
"maxReasoningEffort": "low", # "low" | "medium" | "high"
}
},
)
# Extract response content
content_list = response["output"]["message"]["content"]
for content in content_list:
# Extract the reasoning response
if "reasoningContent" in content:
print("\n== Reasoning ==")
print(content["reasoningContent"]["reasoningText"]["text"])
# Extract the text response
if "text" in content:
print("\n== Text ==")
print(content["text"])
```
------
#### [ Streaming ]
```
import boto3
from botocore.config import Config
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(connect_timeout=3600, read_timeout=3600),
)
# Invoke the model
response = bedrock.converse_stream(
modelId="us.amazon.nova-2-lite-v1:0",
messages=[
{
"role": "user",
"content": [
{
"text": 'How many capital letters appear in the following passage. Your response must include only the number: "Wilfred ordered an anvil from ACME. Shipping was expensive."'
}
],
}
],
additionalModelRequestFields={
"reasoningConfig": {
"type": "enabled",
"maxReasoningEffort": "low", # "low" | "medium" | "high"
},
},
)
# Process the streaming response
reasoning_output = ""
text_output = ""
for event in response["stream"]:
if "contentBlockDelta" in event:
delta = event["contentBlockDelta"]["delta"]
if "reasoningContent" in delta:
if len(reasoning_output) == 0:
print("\n\n== Reasoning ==")
reasoning_text_chunk = delta["reasoningContent"]["text"]
print(reasoning_text_chunk, end="", flush=True)
reasoning_output += reasoning_text_chunk
elif "text" in delta:
if len(text_output) == 0:
print("\n\n== Text ==")
text_chunk = delta["text"]
print(text_chunk, end="", flush=True)
text_output += text_chunk
```
------
### Strumento integrato: Nova Grounding con citazioni
Usa Nova Grounding per recuperare informazioni in tempo reale dal Web con citazioni.
------
#### [ Non-streaming ]
```
import boto3
from botocore.config import Config
# Define the list of tools the model may use
tool_config = {"tools": [{"systemTool": {"name": "nova_grounding"}}]}
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(read_timeout=3600),
)
messages = [
{
"role": "user",
"content": [
{"text": "What is the latest news about renewable energy sources?"}
],
}
]
# Invoke the model
response = bedrock.converse(
modelId="us.amazon.nova-2-lite-v1:0", messages=messages, toolConfig=tool_config
)
# Extract the text with interleaved citations
output_with_citations = ""
content_list = response["output"]["message"]["content"]
for content in content_list:
if "text" in content:
output_with_citations += content["text"]
elif "citationsContent" in content:
citations = content["citationsContent"]["citations"]
for citation in citations:
url = citation["location"]["web"]["url"]
output_with_citations += f"[{url}]"
print(output_with_citations)
```
------
#### [ Streaming ]
```
import boto3
from botocore.config import Config
# Define the list of tools the model may use
tool_config = {"tools": [{"systemTool": {"name": "nova_grounding"}}]}
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(read_timeout=3600),
)
messages = [
{
"role": "user",
"content": [
{"text": "What is the latest news about renewable energy sources?"}
],
}
]
# Invoke the model with streaming
response = bedrock.converse_stream(
modelId="us.amazon.nova-2-lite-v1:0", messages=messages, toolConfig=tool_config
)
# Process the streaming response with interleaved citations
for event in response["stream"]:
if "contentBlockDelta" in event:
delta = event["contentBlockDelta"]["delta"]
if "text" in delta:
print(delta["text"], end="", flush=True)
elif "citation" in delta:
url = delta["citation"]["location"]["web"]["url"]
print(f"[{url}]", end="", flush=True)
```
------
### Strumento integrato: Code Interpreter
Usa lo strumento Code Interpreter per eseguire codice Python per calcoli e analisi dei dati.
------
#### [ Non-streaming ]
```
import boto3
from botocore.config import Config
# Define the list of tools the model may use
tool_config = {"tools": [{"systemTool": {"name": "nova_code_interpreter"}}]}
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(read_timeout=3600),
)
messages = [
{
"role": "user",
"content": [
{
"text": "What is the average of 10, 24, 2, 3, 43, 52, 13, 68, 6, 7, 902, 82?"
}
],
}
]
# Invoke the model
response = bedrock.converse(
modelId="us.amazon.nova-2-lite-v1:0", messages=messages, toolConfig=tool_config
)
# Extract the text and the code the was executed
content_list = response["output"]["message"]["content"]
for content in content_list:
if "text" in content:
print("\n== Text ==")
print(content["text"])
elif "toolUse" in content and content["toolUse"]["name"] == "nova_code_interpreter":
print("\n== Code Interpreter: input.snippet ==")
print(content["toolUse"]["input"]["snippet"])
```
------
#### [ Streaming ]
```
import boto3
from botocore.config import Config
import json
# Define the list of tools the model may use
tool_config = {"tools": [{"systemTool": {"name": "nova_code_interpreter"}}]}
messages = [
{
"role": "user",
"content": [
{
"text": "What is the average of 10, 24, 2, 3, 43, 52, 13, 68, 6, 7, 902, 82?"
}
],
}
]
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(connect_timeout=3600, read_timeout=3600),
)
# Invoke the model with streaming
response = bedrock.converse_stream(
modelId="us.amazon.nova-2-lite-v1:0", messages=messages, toolConfig=tool_config
)
# Process the streaming response
current_block_start = None
response_text = ""
for event in response["stream"]:
if "contentBlockStart" in event:
current_block_start = event["contentBlockStart"]["start"]
elif "contentBlockStop" in event:
current_block_start = None
elif "contentBlockDelta" in event:
delta = event["contentBlockDelta"]["delta"]
if (
current_block_start
and "toolUse" in current_block_start
and current_block_start["toolUse"]["name"] == "nova_code_interpreter"
):
# This is code interpreter content
tool_input = json.loads(delta["toolUse"]["input"])
print("\n== Executed Code Snippet ==")
print(tool_input["snippet"], end="", flush=True)
elif "text" in delta:
# This is text response content
if len(response_text) == 0:
print("\n== Text ==")
text = delta["text"]
response_text += text
print(text, end="", flush=True)
```
------
### Utilizzo degli strumenti
Definisci strumenti personalizzati per il modello da utilizzare durante la conversazione.
------
#### [ Non-streaming ]
```
import boto3
from botocore.config import Config
def get_weather(city):
# Mock function to simulate weather API
return {"temperatureF": 48, "conditions": "light rain"}
# Define the toolSpec for the weather tool
weather_tool = {
"toolSpec": {
"name": "get_weather",
"description": "Get the current weather conditions in a given location",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
}
},
"required": ["city"],
}
},
}
}
# Define the list of tools the model may use
tool_config = {"tools": [weather_tool]}
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(read_timeout=3600),
)
# Start tracking message history
messages = []
messages.append(
{
"role": "user",
"content": [
{
"text": "Suggest some activities to do in Seattle based on the current weather."
}
],
}
)
# Invoke the model
response = bedrock.converse(
modelId="us.amazon.nova-2-lite-v1:0", messages=messages, toolConfig=tool_config
)
assistant_message = response["output"]["message"]
# Add the assistant response to the message history
messages.append(assistant_message)
content_list = assistant_message["content"]
stop_reason = response["stopReason"]
if stop_reason == "tool_use":
# Extract the toolUse details
tool_use = next(
content["toolUse"] for content in content_list if "toolUse" in content
)
tool_name = tool_use["name"]
tool_use_id = tool_use["toolUseId"]
if tool_name == "get_weather":
# Call the tool
weather = get_weather(tool_use["input"]["city"])
# Send the result back to the model
messages.append(
{
"role": "user",
"content": [
{
"toolResult": {
"toolUseId": tool_use_id,
"content": [{"json": weather}],
}
}
],
}
)
# Submit the tool result back to the model
response = bedrock.converse(
modelId="us.amazon.nova-2-lite-v1:0",
messages=messages,
toolConfig=tool_config,
)
content_list = response["output"]["message"]["content"]
for content in content_list:
# Extract the text response
if "text" in content:
print("\n== Text ==")
print(content["text"])
else:
# A tool call was not needed
for content in content_list:
# Extract the text response
if "text" in content:
print("\n== Text ==")
print(content["text"])
```
------
#### [ Streaming ]
```
import boto3
from botocore.config import Config
import json
def get_weather(city):
# Mock function to simulate weather API
return {"temperatureF": 48, "conditions": "light rain"}
# Define the toolSpec for the weather tool
weather_tool = {
"toolSpec": {
"name": "get_weather",
"description": "Get the current weather conditions in a given location",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
}
},
"required": ["city"],
}
},
}
}
# Define the list of tools the model may use
tool_config = {"tools": [weather_tool]}
# Create the Bedrock Runtime client, using an extended timeout configuration
# to support long-running requests.
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(read_timeout=3600),
)
# Start tracking message history
messages = []
messages.append(
{
"role": "user",
"content": [
{
"text": "Suggest some activities to do in Seattle based on the current weather."
}
],
}
)
# Invoke the model with streaming
response = bedrock.converse_stream(
modelId="us.amazon.nova-2-lite-v1:0", messages=messages, toolConfig=tool_config
)
# Process the streaming response
assistant_message = {"role": "assistant", "content": []}
current_tool_use = None
stop_reason = None
for event in response["stream"]:
if "contentBlockStart" in event:
start = event["contentBlockStart"]["start"]
if "toolUse" in start:
current_tool_use = start["toolUse"]
current_tool_use["input"] = ""
elif "contentBlockDelta" in event:
delta = event["contentBlockDelta"]["delta"]
if "toolUse" in delta:
current_tool_use["input"] += delta["toolUse"]["input"]
elif "text" in delta:
print(delta["text"], end="", flush=True)
elif "contentBlockStop" in event:
if current_tool_use:
# Parse the accumulated tool input
current_tool_use["input"] = json.loads(current_tool_use["input"])
assistant_message["content"].append({"toolUse": current_tool_use})
current_tool_use = None
elif "messageStop" in event:
stop_reason = event["messageStop"]["stopReason"]
if stop_reason == "end_turn":
exit
# Add the assistant response to the message history
messages.append(assistant_message)
if stop_reason == "tool_use":
# Extract the toolUse details
tool_use = next(
content["toolUse"]
for content in assistant_message["content"]
if "toolUse" in content
)
tool_name = tool_use["name"]
tool_use_id = tool_use["toolUseId"]
if tool_name == "get_weather":
# Call the tool
weather = get_weather(tool_use["input"]["city"])
# Send the result back to the model
messages.append(
{
"role": "user",
"content": [
{
"toolResult": {
"toolUseId": tool_use_id,
"content": [{"json": weather}],
}
}
],
}
)
# Submit the tool result back to the model with streaming
response = bedrock.converse_stream(
modelId="us.amazon.nova-2-lite-v1:0",
messages=messages,
toolConfig=tool_config,
)
# Handle the final streaming response
print("\n== Text ==")
for event in response["stream"]:
if "contentBlockDelta" in event:
delta = event["contentBlockDelta"]["delta"]
if "text" in delta:
print(delta["text"], end="", flush=True)
```
------
## InvokeModel Esempi di API
Gli esempi seguenti si concentrano sulle poche aree chiave in cui le strutture di richiesta e risposta dell'API Invoke differiscono leggermente da quelle dell'API Converse. Nella maggior parte degli altri modi, le due APIs sono compatibili, quindi dovresti essere in grado di adattare facilmente gli esempi di API Converse di cui sopra per farli funzionare con l'API. InvokeModel
### Richiesta di base
Invia una richiesta di testo di base ai modelli Amazon Nova 2 utilizzando l' InvokeModel API.
------
#### [ Non-streaming ]
```
import json
import boto3
from botocore.config import Config
# Configure the request
request_body = {
"messages": [
{
"role": "user",
"content": [{"text": "Write a short story. End the story with 'THE END'."}],
}
],
"system": [{"text": "You are a children's book author."}], # Optional
"inferenceConfig": { # These parameters are optional
"maxTokens": 1500,
"temperature": 0.7,
"topP": 0.9,
"topK": 50,
"stopSequences": ["THE END"],
},
}
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(read_timeout=3600),
)
# Invoke the model
response = bedrock.invoke_model(
modelId="us.amazon.nova-2-lite-v1:0", body=json.dumps(request_body)
)
response_body = json.loads(response["body"].read())
# Extract the text response
content_list = response_body["output"]["message"]["content"]
for content in content_list:
if "text" in content:
print(content["text"])
```
------
#### [ Streaming ]
```
import json
import boto3
from botocore.config import Config
# Configure the request
request_body = {
"messages": [
{
"role": "user",
"content": [{"text": "Write a short story. End the story with 'THE END'."}],
}
],
"system": [{"text": "You are a children's book author."}], # Optional
"inferenceConfig": { # These parameters are optional
"maxTokens": 1500,
"temperature": 0.7,
"topP": 0.9,
"topK": 50,
"stopSequences": ["THE END"],
},
}
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(connect_timeout=3600, read_timeout=3600),
)
# Invoke the model with streaming
response = bedrock.invoke_model_with_response_stream(
modelId="us.amazon.nova-2-lite-v1:0", body=json.dumps(request_body)
)
# Process the streaming response
for event in response["body"]:
chunk = json.loads(event["chunk"]["bytes"])
if "contentBlockDelta" in chunk:
delta = chunk["contentBlockDelta"]["delta"]
if "text" in delta:
print(delta["text"], end="", flush=True)
```
------
### InvokeModel API con ragionamento
Utilizza l' InvokeModel API con il ragionamento abilitato per la risoluzione di problemi complessi.
------
#### [ Non-streaming ]
```
import json
import boto3
from botocore.config import Config
# Configure the request
request_body = {
"messages": [
{
"role": "user",
"content": [
{
"text": 'How many capital letters appear in the following passage. Your response must include only the number: "Wilfred ordered an anvil from ACME. Shipping was expensive."'
}
],
}
],
"reasoningConfig": {
"type": "enabled",
"maxReasoningEffort": "low", # "low" | "medium" | "high"
},
}
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(read_timeout=3600),
)
# Invoke the model
response = bedrock.invoke_model(
modelId="us.amazon.nova-2-lite-v1:0", body=json.dumps(request_body)
)
response_body = json.loads(response["body"].read())
# Extract response content
content_list = response_body["output"]["message"]["content"]
for content in content_list:
# Extract the reasoning response
if "reasoningContent" in content:
print("\n== Reasoning ==")
print(content["reasoningContent"]["reasoningText"]["text"])
# Extract the text response
if "text" in content:
print("\n== Text ==")
print(content["text"])
```
------
#### [ Streaming ]
```
import json
import boto3
from botocore.config import Config
# Configure the request
request_body = {
"messages": [
{
"role": "user",
"content": [
{
"text": 'How many capital letters appear in the following passage. Your response must include only the number: "Wilfred ordered an anvil from ACME. Shipping was expensive."'
}
],
}
],
"reasoningConfig": {
"type": "enabled",
"maxReasoningEffort": "low", # "low" | "medium" | "high"
},
}
bedrock = boto3.client(
"bedrock-runtime",
region_name="us-east-1",
config=Config(connect_timeout=3600, read_timeout=3600),
)
# Invoke the model with streaming
response = bedrock.invoke_model_with_response_stream(
modelId="us.amazon.nova-2-lite-v1:0", body=json.dumps(request_body)
)
# Process the streaming response
for event in response["body"]:
chunk = json.loads(event["chunk"]["bytes"])
if "contentBlockDelta" in chunk:
delta = chunk["contentBlockDelta"]["delta"]
# Extract the reasoning response
if "reasoningContent" in delta:
print("\n== Reasoning ==")
print(delta["reasoningContent"]["reasoningText"]["text"], end="", flush=True)
# Extract the text response
if "text" in delta:
print("\n== Text ==")
print(delta["text"], end="", flush=True)
```
------