View a markdown version of this page

Bibliothèque de codes - Amazon Nova

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Bibliothèque de codes

Cette section fournit des exemples de code pour les opérations Amazon Nova courantes à l'aide de l'API Converse ou de l' InvokeModel API.

Exemples d'API Converse

Demande de base

Envoyez une demande de texte de base aux modèles Amazon Nova à l'aide de 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)

Entrée multimodale à l'aide d'un actif intégré

Traitez le contenu multimodal en incorporant des documents, des images, des vidéos ou des données audio directement dans la demande. Cet exemple utilise des données d'image. Pour en savoir plus sur la structure du contenu et les autres modalités, consultez la ContentBlock documentation de l'API Amazon Bedrock.

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)

Entrée multimodale à l'aide de l'URI S3

Traitez le contenu multimodal en référençant des documents, des images, des vidéos ou des fichiers audio stockés dans S3. Cet exemple utilise une référence d'image. Pour en savoir plus sur la structure du contenu et les autres modalités, consultez la ContentBlock documentation de l'API Amazon Bedrock.

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": "<account_id>" # 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": "<account_id>" # 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)

Pensée étendue (raisonnement)

Favorisez une réflexion approfondie pour les tâches complexes de résolution de problèmes.

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

Outil intégré : Nova Grounding avec citations

Utilisez Nova Grounding pour récupérer des informations en temps réel sur le Web avec des citations.

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)

Outil intégré : Interpréteur de code

Utilisez l'outil Code Interpreter pour exécuter du code Python pour les calculs et l'analyse des données.

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)

Utilisation d’outil

Définissez des outils personnalisés pour le modèle à utiliser pendant la conversation.

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 Exemples d'API

Les exemples ci-dessous se concentrent sur les quelques domaines clés dans lesquels les structures de demande et de réponse de l'API Invoke diffèrent légèrement de celles de l'API Converse. Dans la plupart des autres cas, les deux APIs sont compatibles. Vous devriez donc être en mesure d'adapter facilement les exemples d'API Converse ci-dessus pour qu'ils fonctionnent avec l' InvokeModel API.

Demande de base

Envoyez une demande de texte de base aux modèles Amazon Nova 2 à l'aide de 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 avec raisonnement

Utilisez l' InvokeModel API avec le raisonnement activé pour la résolution de problèmes complexes.

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)