# コードライブラリ
このセクションでは、Converse API または InvokeModel API を使用した一般的な Amazon Nova オペレーションのコード例を示します。
## Converse API の例
### 基本リクエスト
Converse API を使用して Amazon Nova モデルに基本的なテキストリクエストを送信します。
------
#### [ 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)
```
------
### 埋め込みアセットを使用したマルチモーダル入力
マルチモーダルコンテンツを処理するには、ドキュメント、画像、動画や音声データをリクエストに直接埋め込みます。この例では、画像データを使用します。他のモダリティのコンテンツ構造の詳細については、[Amazon Bedrock API ドキュメントの ContentBlock の詳細](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)
```
------
### S3 URI を使用したマルチモーダル入力
S3 に保存されているドキュメント、画像、動画、または音声ファイルを参照して、マルチモーダルコンテンツを処理します。この例では、イメージリファレンスを使用します。他のモダリティのコンテンツ構造の詳細については、[Amazon Bedrock API ドキュメントの ContentBlock の詳細](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)
```
------
### 拡張思考 (推論)
複雑な問題解決タスクに対して拡張思考を有効にします。
------
#### [ 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
```
------
### 組み込みツール: 引用を含む Nova グラウンディング
Nova グラウンディングを使用して、引用を含むリアルタイム情報をウェブから取得します。
------
#### [ 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)
```
------
### 組み込みツール: コードインタープリタ
コードインタープリタツールを使用して、計算とデータ分析のために Python コードを実行します。
------
#### [ 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)
```
------
### ツール使用
会話中に使用するモデルのカスタムツールを定義します。
------
#### [ 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 API の例
以下の例では、Invoke API のリクエストとレスポンスの構造が Converse API の構造とわずかに異なる主要領域に焦点を当てています。他のほとんどの方法では、2 つの API に互換性があるため、上記の Converse API の例を簡単に適応させて InvokeModel API と連携させることができます。
### 基本リクエスト
InvokeModel API を使用して、基本的なテキストリクエストを Amazon Nova 2 モデルに送信します。
------
#### [ 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
複雑な問題解決のために推論を有効にした InvokeModel API を使用します。
------
#### [ 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)
```
------