View a markdown version of this page

將檔案寫入工作階段 - Amazon Bedrock AgentCore

將檔案寫入工作階段

您可以使用 Code Interpreter 在沙盒環境中讀取和寫入檔案。這可讓您上傳資料檔案、使用程式碼處理它們,以及擷取結果。

安裝相依項目

執行下列命令來安裝所需的套件:

pip install bedrock-agentcore

使用 檔案工具上傳程式碼和資料

下列 Python 程式碼說明如何將檔案上傳至 Code Interpreter 工作階段,並執行處理這些檔案的程式碼。所需的檔案為 data.csvstats.py ,可在此套件中使用

# file_mgmt_ci_agent.py from bedrock_agentcore.tools.code_interpreter_client import CodeInterpreter import json from typing import Dict, Any, List #Configure and Start the code interpreter session code_client = CodeInterpreter('<Region>') code_client.start() #read the content of the sample data file data_file = "data.csv" try: with open(data_file, 'r', encoding='utf-8') as data_file_content: data_file_content = data_file_content.read() #print(data_file_content) except FileNotFoundError: print(f"Error: The file '{data_file}' was not found.") except Exception as e: print(f"An error occurred: {e}") #read the content of the python script to analyze the sample file code_file = "stats.py" try: with open(code_file, 'r', encoding='utf-8') as code_file_content: code_file_content = code_file_content.read() #print(code_file_content) except FileNotFoundError: print(f"Error: The file '{code_file}' was not found.") except Exception as e: print(f"An error occurred: {e}") files_to_create = [ { "path": "data.csv", "text": data_file_content }, { "path": "stats.py", "text": code_file_content }] #define the method to call the invoke API def call_tool(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: response = code_client.invoke(tool_name, arguments) for event in response["stream"]: return json.dumps(event["result"], indent=2) #write the sample data and analysis script into the code interpreter session writing_files = call_tool("writeFiles", {"content": files_to_create}) print(f"writing files: {writing_files}") #List and validate that the files were written successfully listing_files = call_tool("listFiles", {"path": ""}) print(f"listing files: {listing_files}") #Run the python script to analyze the sample data file execute_code = call_tool("executeCode", { "code": files_to_create[1]['text'], "language": "python", "clearContext": True}) print(f"code execution result: {execute_code}") #Clean up and stop the code interpreter session code_client.stop()