Bot

仮想通貨botの開発記録#69(2024/4/9)「生成AIを活用してコーディングスキルを上達させる方法」

前回の記事に引き続き、今回も仮想通貨botの開発状況をまとめていきます。

最近は「ChatGPTを活用してコーディングのスキルを上達させる方法」を試しています。

私自身は、これがコードを書く練習とシステムの理解を深める手助けになると感じているので、取り組んでいることを簡単に紹介します。

今回は「pythonの非同期処理を身につけるための方法」に絞ってお伝えします。

練習問題

方法は非常にシンプルです。

  • ChatGPTに練習問題とその答えを書かせる
  • 問題を見て、自力でコードを書く(Docsを熟読する)
  • ChatGPTが書いたコードと比較する
  • １〜３を毎日繰り返す
Yodaka

「pythonの非同期処理の練習問題を出して」などのざっくりした指示でも、それなりの問題を出してくれます。

基本問題

  1. Hello Async World:
    • asyncio を使用して、"Hello, async world!" を非同期に表示するシンプルなスクリプトを書いてください。
  2. 非同期タスクの実行:
    • 3つの非同期関数を定義し、それぞれ異なる待機時間（例：1秒、2秒、3秒）を持たせてください。その後、これらの関数を非同期に実行し、完了した順に結果を表示するプログラムを作成してください。

中級問題

  1. 非同期IO操作:
    • 非同期にファイルを読み書きするための関数を作成してください。テキストファイルを読み込み、その内容を大文字に変換して別のファイルに書き出す非同期プログラムを作成してください。
  2. 並行HTTPリクエスト:
    • aiohttp ライブラリを使用して、複数のウェブサイト（例：Google, Yahoo, Bing）に対して非同期にHTTP GETリクエストを行い、それぞれのレスポンス時間を計測するプログラムを作成してください。

上級問題

  1. 非同期処理のキャンセルと例外処理:
    • 複数の非同期タスクを実行中に、特定の条件（例：全体の実行時間が一定時間を超えた場合）でタスクをキャンセルするようなプログラムを作成してください。また、例外処理を適切に行い、タスクの成功、失敗、キャンセルをログに記録してください。
  2. 非同期ジェネレータ:
    • 非同期ジェネレータを使用して、指定されたディレクトリ内のファイルを非同期に検索し、ファイル名を一つずつyieldする関数を作成してください。この関数を使用して、特定の拡張子を持つファイルを非同期に検索するプログラムを書いてください。

練習問題の答え

Yodaka

それでは、上記の非同期処理の練習問題に対する答えを一つずつ書いていきます。「問題を出して」という指示を出しただけでは、ChatGPTが答えを書いてくれない場合もあるので、その時は「上記の問題の答えを書いて」などの指示を出しましょう。

基本問題の解答

1. Hello Async World

import asyncio

async def say_hello():
    print("Hello, async world!")

asyncio.run(say_hello())

2. 非同期タスクの実行

import asyncio

async def task_wait(time):
    await asyncio.sleep(time)
    return f"Task with wait time {time} completed"

async def main():
    tasks = [task_wait(1), task_wait(2), task_wait(3)]
    for completed_task in asyncio.as_completed(tasks):
        result = await completed_task
        print(result)

asyncio.run(main())

中級問題の解答

3. 非同期IO操作

import asyncio

async def read_file(file_path):
    async with aiofiles.open(file_path, 'r') as f:
        return await f.read()

async def write_file(file_path, content):
    async with aiofiles.open(file_path, 'w') as f:
        await f.write(content)

async def convert_to_uppercase(input_file, output_file):
    content = await read_file(input_file)
    await write_file(output_file, content.upper())

# aiofiles ライブラリが必要です。
# `pip install aiofiles` を実行してインストールしてください。
import aiofiles
asyncio.run(convert_to_uppercase('input.txt', 'output.txt'))

4. 並行HTTPリクエスト

import aiohttp
import asyncio

async def fetch(url, session):
    async with session.get(url) as response:
        start = asyncio.get_running_loop().time()
        await response.text()
        end = asyncio.get_running_loop().time()
        print(f"{url}: {end - start} seconds")

async def main():
    async with aiohttp.ClientSession() as session:
        urls = ["http://google.com", "http://yahoo.com", "http://bing.com"]
        tasks = [fetch(url, session) for url in urls]
        await asyncio.gather(*tasks)

# aiohttp ライブラリが必要です。
# `pip install aiohttp` を実行してインストールしてください。
asyncio.run(main())

バックグラウンドでイベントループが発生している場合

Yodaka

たとえば、エディタがSpyderなどの環境下では以下の手順で実行可能です。

SpyderやJupyter Notebookなどの一部のIDEやインタラクティブな環境では、既にバックグラウンドでイベントループが動作しているため、ランタイムエラーに遭遇しやすくなります。

こうした環境で非同期コードを実行する一つの方法は、nest_asyncio パッケージを利用することです。このパッケージは、既に実行中のイベントループ内で非同期コードを実行できるようにするものです。

pip install nest_asyncio

nest_asyncio.apply() をスクリプトの先頭で呼び出して、既に実行中のイベントループに対して非同期コードを実行できるようにします。

import aiohttp
import asyncio
import nest_asyncio

nest_asyncio.apply()  # この行を追加

async def fetch(url, session):
    async with session.get(url) as response:
        start = asyncio.get_running_loop().time()
        await response.text()
        end = asyncio.get_running_loop().time()
        print(f"{url}: {end - start} seconds")

async def main():
    async with aiohttp.ClientSession() as session:
        urls = ["http://google.com", "http://yahoo.com", "http://bing.com"]
        tasks = [fetch(url, session) for url in urls]
        await asyncio.gather(*tasks)

# asyncio.run(main()) を直接実行
asyncio.run(main())

上級問題の解答

5. 非同期処理のキャンセルと例外処理

import asyncio

async def task(name, duration):
    try:
        await asyncio.sleep(duration)
        print(f"{name} completed")
    except asyncio.CancelledError:
        print(f"{name} was cancelled")

async def main():
    tasks = [task(f"Task {i}", i+1) for i in range(3)]
    task_group = asyncio.gather(*tasks)
    await asyncio.sleep(2)  # 2秒後に全てのタスクをキャンセル
    task_group.cancel()
    try:
        await task_group
    except asyncio.CancelledError:
        print("Tasks were cancelled")

asyncio.run(main())

6. 非同期ジェネレータ

import asyncio
import os

async def async_file_finder(directory, extension):
    for root, dirs, files in os.walk(directory):
        for file in files:
            if file.endswith(extension):
                await asyncio.sleep(0.1)  # シミュレートされたIO待機
                yield os.path.join(root, file)

async def main():
    async for file_path in async_file_finder(".", ".py"):
        print(file_path)

asyncio.run(main())

まとめ

技術を上達させるためには、特定の技能（今回は非同期処理）に絞って練習メニューに継続的に取り組むことが効果的です。

プログラミングスキルの上達にも同じ方法が使えるのではないかと考え、今回の方法を試してみました。

継続して練習に取り組む中で、理解度を深めて技能を磨いていきます。

さらに、練習問題と解答を即座に用意してくれる点に生成AIの強みがあるということを実感しました。

専門的な技能を用いたものの完成系がイメージできなかったり、何から手をつけて良いのか分からない場合は、生成AIを活用して練習に取り組むというアプローチを活用しようと思います。

Yodaka

今後もこの調子で、開発・研究の過程をまとめていきます。

-Bot