如何使用ChatGPT自动化AWS任务:与S3存储桶互动的示例

本教程将向您展示如何使用ChatGPT函数来执行与Amazon S3存储桶相关的任务的示例。这篇教程涵盖了S3存储桶的关键功能,包括运行简单的列出命令、在所有存储桶中搜索特定文件、将文件上传到存储桶以及从存储桶下载文件。OpenAI Chat API可以理解用户的指令,生成自然语言响应,并根据用户输入提取适当的函数调用。

先决条件

在开始之前,请确保您具备以下条件:

  1. 生成一个具有S3存储桶写入权限的AWS访问密钥,并将它们存储在与OpenAI密钥一起的本地环境文件中。环境文件的格式如下:
AWS_ACCESS_KEY_ID=<您的密钥>
AWS_SECRET_ACCESS_KEY=<您的密钥>
OPENAI_API_KEY=<您的密钥>
  1. 安装所需的Python模块。您可以使用以下命令来安装它们:
!pip install openai boto3 tenacity python-dotenv

代码示例

以下是使用ChatGPT和AWS S3存储桶互动的示例代码。

import openai
import json
import boto3
import os
import datetime
from urllib.request import urlretrieve
from dotenv import load_dotenv

# 从.env文件加载环境变量
load_dotenv()

# 设置OpenAI API密钥
openai.api_key = os.environ.get("OPENAI_API_KEY")
GPT_MODEL = "gpt-3.5-turbo"

# 创建S3客户端
s3_client = boto3.client('s3')

# 定义函数字典,以便为GPT模型提供S3操作的详细信息
functions = [
    # 列出所有可用的S3存储桶
    {
        "name": "list_buckets",
        "description": "列出所有可用的S3存储桶",
        "parameters": {
            "type": "object",
            "properties": {}
        }
    },
    # 列出给定S3存储桶内的对象或文件
    {
        "name": "list_objects",
        "description": "列出给定S3存储桶内的对象或文件",
        "parameters": {
            "type": "object",
            "properties": {
                "bucket": {"type": "string", "description": "S3存储桶的名称"},
                "prefix": {"type": "string", "description": "S3存储桶中的文件夹路径"},
            },
            "required": ["bucket"],
        },
    },
    # 从S3存储桶中下载特定文件到本地目标文件夹
    {
        "name": "download_file",
        "description": "从S3存储桶中下载特定文件到本地目标文件夹",
        "parameters": {
            "type": "object",
            "properties": {
                "bucket": {"type": "string", "description": "S3存储桶的名称"},
                "key": {"type": "string", "description": "S3存储桶中文件的路径"},
                "directory": {"type": "string", "description": "本地目标目录,由用户指定。"},
            },
            "required": ["bucket", "key", "directory"],
        }
    },
    # 将文件上传到S3存储桶
    {
        "name": "upload_file",
        "description": "将文件上传到S3存储桶",
        "parameters": {
            "type": "object",
            "properties": {
                "source": {"type": "string", "description": "本地源文件路径或远程URL"},
                "bucket": {"type": "string", "description": "S3存储桶的名称"},
                "key": {"type": "string", "description": "S3存储桶中文件的路径"},
                "is_remote_url": {"type": "boolean", "description": "提供的源是否为URL(True)还是本地路径(False)"},
            },
            "required": ["source", "bucket", "key", "is_remote_url"],
        }
    },
    # 在S3存储桶内搜索特定文件名
    {
        "name": "search_s3_objects",
        "description": "在S3存储桶内搜索特定文件名",
        "parameters": {
            "type": "object",
            "properties": {
                "search_name": {"type": "string", "description": "要搜索的文件名"},
                "bucket": {"type": "string", "description": "S3存储桶的名称"},
                "prefix": {"type": "string", "description": "S3存储桶中的文件夹路径"},
                "exact_match": {"type": "boolean", "description": "如果搜索应匹配完整文件名,请将exact_match设置为True。将exact_match设置为False以比较文件名字符串的一部分(文件包含)"}
            },
            "required": ["search_name"],
        },
    }
]

# 创建用于将S3操作与函数连接的辅助函数

def datetime_converter(obj):
    if isinstance(obj, datetime.datetime):
        return obj.isoformat()
    raise TypeError(f"Object of type {obj.__class__.__name__} is not JSON serializable")

def list_buckets():
    response = s3_client.list_buckets()
    return json.dumps(response['Buckets'], default=datetime_converter)

def list_objects(bucket, prefix=''):
    response = s3_client.list_objects_v2(Bucket=bucket, Prefix=prefix)
    return json.dumps(response.get('Contents', []), default=datetime_converter)

def download_file(bucket, key, directory):

    filename = os.path.basename(key)

    # 解析目标,得到正确的文件路径
    destination = os.path.join(directory, filename)

    s3_client.download_file(bucket, key, destination)
    return json.dumps({"status": "success", "bucket": bucket, "key": key, "destination": destination})

def upload_file(source, bucket, key, is_remote_url=False):
    if is_remote_url:
        file_name = os.path.basename(source)
        urlretrieve(source, file_name)
        source = file_name

    s3

_client.upload_file(source, bucket, key)
    return json.dumps({"status": "success", "source": source, "bucket": bucket, "key": key})

def search_s3_objects(search_name, bucket=None, prefix='', exact_match=True):
    search_name = search_name.lower()

    if bucket is None:
        buckets_response = json.loads(list_buckets())
        buckets = [bucket_info["Name"] for bucket_info in buckets_response]
    else:
        buckets = [bucket]

    results = []

    for bucket_name in buckets:
        objects_response = json.loads(list_objects(bucket_name, prefix))
        if exact_match:
            bucket_results = [obj for obj in objects_response if search_name == obj['Key'].lower()]
        else:
            bucket_results = [obj for obj in objects_response if search_name in obj['Key'].lower()]

        if bucket_results:
            results.extend([{"Bucket": bucket_name, "Object": obj} for obj in bucket_results])

    return json.dumps(results)

# 连接名称和函数以供执行
available_functions = {
    "list_buckets": list_buckets,
    "list_objects": list_objects,
    "download_file": download_file,
    "upload_file": upload_file,
    "search_s3_objects": search_s3_objects
}

# ChatGPT函数调用
def chat_completion_request(messages, functions=None, function_call='auto', 
                            model_name=GPT_MODEL):

    if functions is not None:
        return openai.ChatCompletion.create(
            model=model_name,
            messages=messages,
            functions=functions,
            function_call=function_call)
    else:
        return openai.ChatCompletion.create(
            model=model_name,
            messages=messages)

# 主要对话函数
def run_conversation(user_input, topic="S3 bucket functions.", is_log=False):

    system_message=f"不要假设要插入函数的值。如果用户请求不明确,请请求澄清。如果用户要求与{topic}无关的问题,请告诉他您的范围是{topic}。"

    messages = [{"role": "system", "content": system_message},
                {"role": "user", "content": user_input}]

    # 调用模型以获取响应
    response = chat_completion_request(messages, functions=functions)
    response_message = response['choices'][0]['message']

    if is_log:
        print(response['choices'])

    # 检查GPT是否想要调用函数
    if response_message.get("function_call"):
        function_name = response_message['function_call']['name']
        function_args = json.loads(response_message['function_call']['arguments'])

        # 调用函数
        function_response = available_functions[function_name](**function_args)

        # 将响应添加到对话中
        messages.append(response_message)
        messages.append({
            "role": "function",
            "name": function_name,
            "content": function_response,
        })

        # 再次调用模型以总结结果
        second_response = chat_completion_request(messages)
        final_message = second_response['choices'][0]['message']['content']
    else:
        final_message = response_message['content']

    return final_message

# 测试S3存储桶助手
# 列出和搜索
print(run_conversation('列出我的S3存储桶'))

search_file = '<文件名>'
print(run_conversation(f'在所有存储桶中搜索文件{search_file}'))

search_word = '<文件名部分>'
bucket_name = '<存储桶名称>'
print(run_conversation(f'在{bucket_name}中搜索包含{search_word}的文件'))

# 检查模型是否会拒绝无关的任务
print(run_conversation('今天天气如何'))

# 下载文件
search_file = '<文件名>'
bucket_name = '<存储桶名称>'
local_directory = '<本地目录路径>'
print(run_conversation(f'从{bucket_name}存储桶中下载{search_file}到{local_directory}目录'))

# 上传文件
local_file = '<文件名>'
bucket_name = '<存储桶名称>'
print(run_conversation(f'将{local_file}上传到{bucket_name}存储桶'))

结论

通过本教程,您已经学会了如何使用ChatGPT自动化AWS S3存储桶任务。这个互动的示例展示了如何连接ChatGPT的自然语言处理能力和AWS S3存储桶的功能,以执行列出、搜索、下载和上传文件等操作。这种自动化可以节省大量时间,提高工作效率。

本文由作者 王大神 原创发布于 大神网的AI博客。

转载请注明作者:王大神

原文出处:如何使用ChatGPT自动化AWS任务:与S3存储桶互动的示例

(0)
打赏 微信扫一扫 微信扫一扫
上一篇 2023年10月14日
下一篇 2023年10月14日

相关推荐

  • 前苹果高管创立Humane公司,将发布无屏幕AI可穿戴设备Ai Pin

    Humane公司,由两位前苹果公司的设计和工程团队高管创立,即将发布一款引人注目的无屏幕AI可穿戴设备——Ai Pin。这个备受期待的新产品将在11月9日正式亮相,让我们一起来了解一下这个新兴品牌和他们的创新之处。 Hum…

    2023年11月7日
    00
  • 聊天机器人与我——新手如何更好地与ChatGPT交流?

    在一个寒冷的夜晚,我坐在书桌前,面对屏幕,不断地试图和我的新助理——ChatGPT进行交流。每当我认为我掌握了和它聊天的技巧,结果往往让我失望。这就好像我突然进入了鲁迅笔下的小杂货铺,和店主尝试用我不太熟悉的…

    2023年10月5日
    00
  • 利用ChatGPT助力联盟营销:创新工具提升收益效能

    利用ChatGPT助力联盟营销:创新工具提升收益效能 前言 在数字营销领域,联盟营销一直是一种受欢迎的盈利模式。而现在,借助人工智能工具如Chat Generative Pre-trained Transformer(ChatGPT),我们可以进一步提升…

    2023年9月3日
    00
  • 大神网自用chatgpt prompt分享和不算教程的教程

    大家有了chatgpt但都不怎么会用,单纯的跟chatgpt聊天,然后还需要自己整理内容。今天分享一下自己的心得和prompt。 首先如果你要写一篇通用的热点新闻 prompt: —————————————— 根据我…

    2023年11月13日
    00
  • 创建自动化试卷生成工具:使用Python和OpenPyXL

    你是否曾经需要为学生创建大量的试卷,但又不想手动编写每一份试卷?在这个教程中,我们将使用Python编写一个自动化试卷生成工具,可以根据题库随机生成试卷,并保存为Word文档。无需编程经验,我们将一步步引导你…

    2023年9月13日
    00
  • ChatGPT最新版本插件大揭秘

    近年来,人工智能技术取得了巨大的突破,而ChatGPT作为其中一员,一直在不断进化。最新发布的GPT-4版本不仅在核心功能上有所提升,还引入了一系列令人兴奋的插件,以满足不同用户的需求。在本文中,我们将深入介绍…

    2023年11月7日 指数词
    00
  • 操纵ChatGPT之艺:高质量回答的获取指南与提示工程全解

    摘要:在本文中,我将与你一起探索如何通过精心设计的提示(Prompt Engineering)向ChatGPT提出问题,以获得相关、准确和高质量的回答。我们将深入了解各种提示技术的应用,如何结合创作需求设计有效的Prompt,并通…

    2024年3月20日
    00
  • 谷歌的 Gemini 流量挑战 ChatGPT,未来 AI 对话界的霸主?

    本文将探讨谷歌 Gemini AI的市场表现,与其他领先的对话AI,如ChatGPT和Claude的市场表现进行比较。通过分析流量数据和用户行为,我们将了解当前AI对话平台的竞争格局,并预测未来可能的发展趋势。 引言 在AI对话领…

    2024年4月18日
    00
  • ChatGPT注册教程

    ChatGPT是一款强大的自然语言处理工具,它可以回答问题、参与对话、提供解决方案,甚至作为学习工具和自动化助手使用。在本教程中,我们将介绍如何注册ChatGPT并开始使用它,以便充分利用其功能。 注册流程 1. 进入…

    2023年2月14日
    00
  • WeiboBot:基于微博H5 API的机器人框架

    一天,王大神正在研究如何自动化他的社交媒体活动,特别是在微博上。他发现了一款神奇的工具 – WeiboBot,这是一个基于微博H5 API开发的机器人框架,能够让他的机器人更加智能地参与微博互动。今天,我们将深入了解…

    2023年10月10日
    00