BioErrorLog Tech Blog

試行錯誤の記録

LangChain 入門

LLMによる開発パターンを容易に実装できると噂のLangChainに入門します。

はじめに

LangChainは、LLMを活用した開発を容易に実現するフレームワークです。

We believe that the most powerful and differentiated applications will not only call out to a language model via an API, but will also:
- Be data-aware: connect a language model to other sources of data
- Be agentic: allow a language model to interact with its environment

(ドキュメントより)

LamgChainは単にLLMのAPIをラップするのではなく、

  • 外部データと繋げる
  • エージェントとして相互作用する

など、LLMの活用パターンを容易に実現できます。

今回は下記のLangChain Python公式ガイドを参考にしながら、このLangChainの基本的な使い方を見ていきます。

Quickstart Guide — 🦜🔗 LangChain 0.0.150

LangChainに入門する

事前準備

今回はLLMモデルとしてOpenAI APIを使うので、langchainopenaiをインストールします。

pip install langchain openai

OpenAIのAPIキーも環境変数に設定しておきます。

export OPENAI_API_KEY="..."


以降、OpenAI APIを使ったことがある方向けの内容として書いていきます。 まだOpenAI API、特にChat completionを触ったことのない方は、先にOpenAI APIのドキュメントに軽く目を通しておくことをおすすめします:

OpenAI API

Chat Model

まずは、LangChainを用いてgpt-3.5-turboのようなチャット形式でやり取りできるモデルを使ってみます。

from langchain.chat_models import ChatOpenAI
from langchain.schema import (
    AIMessage,
    HumanMessage,
    SystemMessage,
)

chat = ChatOpenAI(temperature=0)

response = chat([HumanMessage(content="Translate this sentence from English to French. I love programming.")])
# -> AIMessage(content="J'aime programmer.", additional_kwargs={})

print(response)  # content="J'aime programmer." additional_kwargs={}
print(type(response))  # <class 'langchain.schema.AIMessage'>

OpenAI APIのgpt-3.5-turboではメッセージをsystem/user/assistantの3つのroleとして与えますが、LangChainのChatOpenAI (こちらもデフォルトでgpt-3.5-turboが使われます)ではそれぞれ、

  • HumanMessage (user role)
  • SystemMessage (system role)
  • AIMessage (assistant role)

として与えます。

シンプルに一問一答の形で良ければ、上記のようにHumanMessageとしてテキストを与えます。


Chat ModelにHumanMessage以外の文脈を与えるには、下記のようにします。

from langchain.chat_models import ChatOpenAI
from langchain.schema import (
    AIMessage,
    HumanMessage,
    SystemMessage,
)

chat = ChatOpenAI(temperature=0)

messages = [
    SystemMessage(content="You are a helpful assistant that translates English to French."),
    HumanMessage(content="Translate this sentence from English to French. I love programming.")
]
response = chat(messages)
# -> AIMessage(content="J'aime programmer.", additional_kwargs={})

print(response)  # content="J'aime programmer." additional_kwargs={}
print(type(response))  # <class 'langchain.schema.AIMessage'>

SystemプロンプトはSystemMessageとして格納し、Chat Modelからの返答はAIMessageとして返されます。 もしそのままやり取りを続ける場合は、Chat Modelからの返答AIMessageをそのままメッセージに追記する、という使い方ができます。


ここまでは単純にOpenAI APIをラップしただけの機能と言えます。 以下、LangChainならではの便利な機能を見ていきます。

Prompt Templates

API経由でのLLMモデル利用パターンの一つに、Prompt Templatesという使い方があります。

Prompt Templatesは、あらかじめ用意されたテンプレートにユーザーからの入力を挿入してプロンプトを構築する方法です。 複雑なプロンプトが必要になるような用途でも、ユーザーからの入力をシンプルにしつつ、LLMに想定の振る舞いをさせることができます。

from langchain.chat_models import ChatOpenAI
from langchain.prompts.chat import (
    ChatPromptTemplate,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
)

chat = ChatOpenAI(temperature=0)

template = "You are a helpful assistant that translates {input_language} to {output_language}."
system_message_prompt = SystemMessagePromptTemplate.from_template(template)
human_template = "{text}"
human_message_prompt = HumanMessagePromptTemplate.from_template(human_template)

chat_prompt = ChatPromptTemplate.from_messages([system_message_prompt, human_message_prompt])

format_prompt = chat_prompt.format_prompt(input_language="English",
                                          output_language="French", text="I love programming.").to_messages()
print(format_prompt)
# -> [SystemMessage(content='You are a helpful assistant that translates English to French.', additional_kwargs={}), HumanMessage(content='I love programming.', additional_kwargs={})]

SystemMessageのプロンプトテンプレートはSystemMessagePromptTemplateで、HumanMessageのプロンプトテンプレートはHumanMessagePromptTemplateで用意します。

後から挿入する部分を鉤括弧{}で用意し、format_prompt関数で挿入箇所を補完することで、最終的なプロンプトが生成されています。

Chains

ここまで挙げたようなModelやPrompt Templatesは、Chainsによってシンプルに繋げることができます。

from langchain.chat_models import ChatOpenAI
from langchain import LLMChain
from langchain.prompts.chat import (
    ChatPromptTemplate,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
)

chat = ChatOpenAI(temperature=0)

template = "You are a helpful assistant that translates {input_language} to {output_language}."
system_message_prompt = SystemMessagePromptTemplate.from_template(template)
human_template = "{text}"
human_message_prompt = HumanMessagePromptTemplate.from_template(human_template)
chat_prompt = ChatPromptTemplate.from_messages([system_message_prompt, human_message_prompt])

chain = LLMChain(llm=chat, prompt=chat_prompt)
response = chain.run(input_language="English", output_language="French", text="I love programming.")

print(response)
# -> "J'aime programmer."

LLMChainによって、LLMモデルとPrompt Templateを連結しています。

Chainsに連結したら、あとはrun関数の引数にPrompt Templateの補完を渡して、LLMへのリクエストを投げることができます。


ここまでのおさらい:

  • Chat Model: LLMモデルのAPIラッパー
  • Prompt Templates: プロンプトを後から補完できるテンプレート
  • Chains: これらを連結してシンプルなインターフェースを提供

以下、より応用的な使い方を見ていきます。

Agents

ここまでは、LLMに行わせる振る舞いをあくまで人間が指定してきました。 Agentsでは、何のアクションを行うべきなのか、をLLMに決めさせることができます。

Agentsは大きく3つの概念から成り立っています。

  • Tool: 特定タスクを実行するための外部機能
  • LLM: Agentの振る舞いを決定するLLM
  • Agent: Agentのタイプ。今回はChat Modelを使う際の最も標準的なタイプAgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTIONを使います。
from langchain.agents import load_tools
from langchain.agents import initialize_agent
from langchain.agents import AgentType
from langchain.chat_models import ChatOpenAI


chat = ChatOpenAI(temperature=0)
tools = load_tools(["terminal"])

agent = initialize_agent(
    tools,
    chat,
    agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

agent.run("Who is the current terminal user?I would like to know the list of directories under that user's home directory.")

使い方はシンプルです。 initialize_agentに先ほど挙げた3つの概念Tool/LLM/Agentを渡して初期化し、run関数で実行します。

上の例では、Toolとしてterminalを、LLMとしてChatOpenAIを、AgentとしてAgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTIONを渡しています。

サポートされているToolの一覧はこちら、Agentの一覧はこちらを参照ください。


この例では、Toolとしてterminalを利用し、現在のユーザーとそのホームディレクトリ配下のディレクトリ構成を質問しています。

Who is the current terminal user? I would like to know the list of directories under that user's home directory.

実行してみましょう。

> Entering new AgentExecutor chain...
Thought: We can use the Terminal tool to execute the `whoami` command to get the current user and then use the `ls` command to list the directories under the user's home directory.

Action:
{
  "action": "Terminal",
  "action_input": "whoami"
}


Observation: bioerrorlog

Thought:Now that we have the current user, we can use the `ls` command to list the directories under the user's home directory.

Action:
{
  "action": "Terminal",
  "action_input": "ls /home/bioerrorlog"
}


Observation: Command 'ls /home/bioerrorlog' returned non-zero exit status 1.
Thought:The `ls` command returned a non-zero exit status, which means that there might be an issue with the path we provided. We can try using the `~` symbol to represent the home directory of the current user.

Action:
{
  "action": "Terminal",
  "action_input": "ls ~"
}


Observation: AWSCLIV2.pkg
Applications
Desktop
Documents
Downloads
Dropbox
Library
Movies
Music
Pictures
Public
dev
dotfiles

Thought:We have successfully listed the directories under the current user's home directory using the `ls` command with the `~` symbol. 

Final Answer: The list of directories under the current user's home directory are: AWSCLIV2.pkg, Applications, Desktop, Documents, Downloads, Dropbox, Library, Movies, Music, Pictures, Public, dev, and dotfiles.

> Finished chain.

与えられた質問に回答するためのアクションが何なのかを推察し、最終回答が得られるまで繰り返されます。

Thought: We can use the Terminal tool to execute the whoami command to get the current user and then use the ls command to list the directories under the user's home directory.


あるアクションがエラーになったとしても、またすぐ別の方法を試行してくれるのも面白いところです。

Observation: Command 'ls /home/bioerrorlog' returned non-zero exit status 1.

このようにホームディレクトリをlsしようとしてエラーになった場合、また別の方法で試行してくれます:

Thought:The ls command returned a non-zero exit status, which means that there might be an issue with the path we provided. We can try using the ~ symbol to represent the home directory of the current user.

Toolを利用したアクションが繰り返された後、最終回答が得られます:

Final Answer: The list of directories under the current user's home directory are: AWSCLIV2.pkg, Applications, Desktop, Documents, Downloads, Dropbox, Library, Movies, Music, Pictures, Public, dev, and dotfiles.

Memory

LLMは基本的にステートレスなAPIのため、過去に行ったやりとりの記憶は何か工夫しない限り保持されません。

そこで最後に、ChainやAgentに記憶を保持させることができる、Memoryを紹介します。

from langchain.prompts import (
    ChatPromptTemplate,
    MessagesPlaceholder,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate
)
from langchain.chains import ConversationChain
from langchain.chat_models import ChatOpenAI
from langchain.memory import ConversationBufferMemory

prompt = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template("The following is a friendly conversation between a human and an AI. The AI is talkative and provides lots of specific details from its context. If the AI does not know the answer to a question, it truthfully says it does not know."),
    MessagesPlaceholder(variable_name="history"),
    HumanMessagePromptTemplate.from_template("{input}")
])

llm = ChatOpenAI(temperature=0)
memory = ConversationBufferMemory(return_messages=True)
conversation = ConversationChain(memory=memory, prompt=prompt, llm=llm)

response = conversation.predict(input="Hi I'm BioErrorLog, living in Tokyo.")
print(response)
# -> "Hello BioErrorLog, it's nice to meet you! I'm an AI language model. How can I assist you today?"

response = conversation.predict(input="Today is a holiday and the weather is nice.")
print(response)
# -> "That sounds lovely! Do you have any plans for the day?"

response = conversation.predict(input="Summarize our conversation so far.")
print(response)
# -> "Sure! You introduced yourself as BioErrorLog and mentioned that you're currently living in Tokyo. You also mentioned that today is a holiday and the weather is nice."

ConversationBufferMemoryをChainに渡すことで、そのChainで行ったやりとりの記憶を簡単に保持させることができます。

上の例では3回のやり取りを行なっていますが、3回目の質問の時点でも1, 2回目のやり取りの記憶が保持されていることがわかります。

Q:
Summarize our conversation so far.
A:
Sure! You introduced yourself as BioErrorLog and mentioned that you're currently living in Tokyo. You also mentioned that today is a holiday and the weather is nice.

この他にも、Vector Storeを使ったMemoryやナレッジグラフをベースにしたMemoryなど、様々なタイプが提供されています。 色々遊んでみると面白そうですね: How-To Guides — 🦜🔗 LangChain 0.0.154

おわりに

以上、LangChainの基本的な使い方をざっとまとめました。

LLMのフレームワークは、LangChainの他にもMicrosoftのSemantic Kernelなど色々と選択肢があると思います。

現時点ではその中でもLangChainが一番広く使われているように見受けられます。

ぜひ遊び倒していきたいところです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

www.bioerrorlog.work

参考

GitHub - hwchase17/langchain: ⚡ Building applications with LLMs through composability ⚡

Welcome to LangChain — 🦜🔗 LangChain 0.0.152

Quickstart Guide — 🦜🔗 LangChain 0.0.150

OpenAI API

GitHub - bioerrorlog/langchain-sandbox

GitHub - microsoft/semantic-kernel: Integrate cutting-edge LLM technology quickly and easily into your apps

flake8で特定ルールのみ指定してチェックする | Python

--selectオプションで指定できます。

はじめに

flake8を使っていて、特定ルール/エラーコードのみ指定してチェックを実行したくなりました。

--helpマニュアルをぱっと眺めて見逃してしまったので、備忘録を残します。

flake8で特定ルールのみ指定する

やり方

# 例
flake8 --select=F811,E202

--selectオプションに対して、コンマ区切りでエラーコードを渡すことで指定できます。

エラーコード指定は前方一致で適用されるので、例えば--select=Fとした場合はFから始まるエラー全てが、--select=F8として場合はF8から始まるエラー全てが指定されます。

# マニュアル抜粋
$ flake8 --help                                    
usage: flake8 [options] file file ...

positional arguments:
  filename

options:
<中略>
  --select errors       Comma-separated list of error codes to enable. For example, ``--select=E4,E51,W234``. (Default: E,F,W,C90)

実行例

# エラー単体指定
$ flake8 --select=F811     
./app/backend/approaches/readretrieveread.py:9:1: F811 redefinition of unused 'AzureOpenAI' from line 5

# エラー複数指定
$ flake8 --select=F811,E202
./app/backend/approaches/readdecomposeask.py:39:126: E202 whitespace before ']'
./app/backend/approaches/readretrieveread.py:9:1: F811 redefinition of unused 'AzureOpenAI' from line 5
./scripts/prepdocs.py:289:70: E202 whitespace before '}'

# エラー前方一致指定
$ flake8 --select=F5 
./app/backend/langchainadapters.py:21:22: F541 f-string is missing placeholders
./app/backend/langchainadapters.py:39:22: F541 f-string is missing placeholders
./scripts/prepdocs.py:301:11: F541 f-string is missing placeholders

おわりに

以上、ちょっとした備忘録でした。

前方一致で指定できる、というのは少し面白いですね。

どなたかの参考になれば幸いです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

参考

flake8 · PyPI

Full Listing of Options and Their Descriptions — flake8 6.0.0 documentation

Error / Violation Codes — flake8 6.0.0 documentation

ChatGPTをロボットの頭脳にする その1: カメラ/サーボモーターとChatGPTを組み合わせる

Raspberry Pi上で、カメラ/サーボモーターとChatGPTを組み合わせて簡単な実験を行います。

はじめに

ChatGPTを初めて触った時、これはロボットの頭脳として使えるのでは、とピンと来ました。

最小限の世界観を作ってみたので、備忘録を残します。

ChatGPTとRaspberry Pi/カメラ/サーボモーターを組み合わせる

プロトタイプコンセプト

今回の実験コンセプトはこちら:

  1. ChatGPTが現実世界の情報を認識する
  2. ChatGPTが現実世界の次のアクションを自ら決定する
  3. 1-2を繰り返し、自分が置かれている現実世界の状況をChatGPTが把握する
  4. 「ねえ今どんな気持ち?」ってChatGPTに聞く

現実世界の次のアクションを"ChatGPT自身が"決定する、というところがこだわりです。

将来的には、ChatGPTを頭脳としてロボットが自由に動き回り、環境を自律的に学習していく、みたいなことが出来たら面白いですね。

構成

ハードウェア

ハードウェア構成

サーボモーター2つとカメラ1つをRaspberry Piに接続しています。

サーボモーターの制御方法は、以前書いたこちらの記事と同様です。

www.bioerrorlog.work

サーボモーターとカメラは、両面テープと輪ゴムでくっつける、という小学生の工作みたいなことをしています。

サーボモーターとカメラを両面テープと輪ゴムでくくり付けただけの"首振りロボット"

これで水平方向および垂直方向の首振りと、カメラによる画像取得が可能です。

ソフトウェア

ソースコードはこちら:

github.com

ざっくり下記の処理を行っています。

  1. カメラから画像を取得
  2. 取得画像から物体認識
  3. 現状のサーボモーターの角度と、画像に映った物体の名前をChatGPT APIに送信
  4. 次のサーボモーターの角度と、今の気持ち(フリートーク)がChatGPT APIから返される
  5. ChatGPTが指定した角度に従い、サーボモーターを制御
  6. 1-5を複数回繰り返す
  7. 置かれた環境についての説明を求める


一つのポイントは、"次のサーボモーターの角度"をChatGPTに指定させることです。 返答に含まれる値をPython側で解釈し、実際にサーボモーターを制御しています。 画面のあちら側だけの存在だったChatGPTに、現実世界に作用できる手段を与えたみたいで興奮しました。

これは、ChatGPTからの返答をJSON形式に固定させることで実現しています。 (JSONならそのままPythonで容易に解釈できる)

ChatGPTからの返答をJSON形式に固定させるためには、ChatGPT APIのtemperature(回答のランダム性パラメータのようなもの)を低めに設定する必要があります。 今回は0.2に設定して上手く機能しましたが、高め(例えばデフォルトの1.0)に設定すると、指定したJSON形式とは異なる返答がくる場合が出てきます。


もう一つの注意点は、ChatGPT APIからのレスポンスは現状かなり遅い(10-20sec, ひどい時はもっと遅い)、という点です。 これはForumで調べてみても皆同じように言っているのですが、現状は改善を待つことしかできなそうです。

Search results for 'API response too slow' - OpenAI API Community Forum

スムーズにこの首振りロボットを動かすには、なるべくAPIを呼び出す回数を減らす工夫が必要になります。 今回のケースでは、一回の応答の中に"次のサーボモーター角度"の指定を複数回分入れ込ませることで、API応答の待ち時間が毎回発生するのを回避しました。

最終的には、こちらのsystemプロンプトに落ち着きました:

You are a robot with a camera, composed of 2 servo motors: horizontal & vertical.
Horizontal: min -90 right, max 90 left.
Vertical: min -90 down, max 90 up.
Your behavior principles: [curiosity, inquisitiveness, playfulness].
Your answer MUST be in this JSON format: {"NextServoMotor": [{"Horizontal": int(-90~90), "Vertical": int(-90~90)}], "FreeTalk": string}
Constraint: len(your_answer["NextServoMotor"]) == 5
Answer example: {"NextServoMotor": [{"Horizontal": -60, "Vertical": -30},{"Horizontal": 0, "Vertical": 0},{"Horizontal": 90, "Vertical": -45},{"Horizontal": 0, "Vertical": 60},{"Horizontal": -30, "Vertical": -60}],"FreeTalk": "Based on what I've seen, I'm curious about the PC and mouse. I wonder what you use them for and what kind of work or play they are involved in?"}

robot-gpt/robot.py at 85c256e3366f57532e74ee5c1294b69717647df9 · bioerrorlog/robot-gpt · GitHub

動作結果

実際にこの"首振りロボット"が動作している様子はこちらです:

周囲の探索を複数回行った後、置かれている状況を尋ねると下記のような返答が返ってきます。

Based on what I have seen, there are a few objects in the room, including a bottle and a laptop. However, I have not seen much else yet. I am curious to explore more and see what other interesting things I can find!

「これまでに見たものに基づくと、部屋にはボトルとラップトップ等の物がいくつかあります。 しかし、まだそれ以外の物はあまり見てません。 もっと探索して、他にどんな面白いものが見つけられるか楽しみです!」

想定した最小限の世界観を実現できました。

夢が広がります。

課題

最小限の世界観は実現できましたが、もちろん課題がたくさんあります。

物体認識

現状の実装では、物体認識の精度は高くありません。

今回はカメラで取得した画像をYOLOv3で物体認識させています。 実際は周囲の物体を認識できることは稀で、たまに何かを認識する(ほとんど何も認識しない)くらいの挙動です。

画像認識関連に詳しくなかったので、ChatGPTに聞いたりググって出てきたやり方で簡単に実装しましたが、改めて調べてみるとYOLOv3はかなり古いモデルのようです。 新しいモデルを使えば、精度も処理速度も向上するかもしれません。

そもそも、GPTでマルチモーダルが利用可能になれば、取得画像ごとGPTに送って認識させる、という形にできるでしょう。 マルチモーダルの課金体系が良心的であることを祈っています。

Token消費量

現在の実装では、ChatGPT APIから帰ってきたレスポンスは全てassistantプロンプトとして追加しています。 動作を続ければ続けるほど、蓄積されたレスポンス全てをプロンプトに入れ込むので、tokenの消費量は増加していく仕様です。

ちょっとした実験くらいなら無視できるような課金量ですが、今後長く起動し続けるような使い方をしてくならば、今のやり方は現実的ではありません。

レスポンス全てをプロンプトに入れ込むのではなく、重要情報を要約してcontextに入れ込むようにすれば、情報量の喪失を抑えたままtoken消費を軽減できるでしょう。

記憶保持

現在は一連の処理が終われば記憶は残りません。 起動する度にまたゼロから周囲を探索する形になっています。

例えばデータベースを使って外部記憶の仕組みを導入すれば、いわゆる長期記憶的な感じで情報を保持できるようになるでしょう。 長期記憶があれば、遊びの幅が広がります。

外界の認識

現在の実装では、

  • サーボモーターの角度 (水平方向および垂直方向)
  • 認識された物体名のリスト

の組み合わせを蓄積することで、外界の状況を認識させています。

このやり方でも、結構上手く外界を認識できています。 例えば「周囲にある物を左側から順に説明して」みたいに聞いても、ある程度正しい位置関係で説明が返ってきます。

ただ、もっとChatGPT/LLMならではのやり方が、もっと機械っぽくない良いやり方があると思っています。 物体リストと角度の組み合わせではなく、例えば「周囲の状況を詳細に描写した文章」を記憶の拠り所にする、とかも面白そうですね。 あるいはマルチモーダルが使えるようになったら、もっと自由な手段が取れそうです。

「機械/プログラムを扱っている」という人間側の思い入れが、遊びの幅を狭めてしまうことのないよう気をつけたいところです。

おわりに

以上、ChatGPTをロボットの頭脳にする取り組みの第一弾として、試作したものをまとめました。

昨今のLLMの盛り上がりから、何か面白い物が作れるんじゃないか、という興奮が止みません。

色々実験して遊び倒していきたいところです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

www.bioerrorlog.work

参考

OpenAI API

Search results for 'API response too slow' - OpenAI API Community Forum

GitHub - bioerrorlog/robot-gpt at 85c256e3366f57532e74ee5c1294b69717647df9

Parquetファイルのダミーデータを生成する | Python

Pythonでparquetファイルのダミーデータを生成する方法の備忘録です。

はじめに

ちょっとした検証のために、ダミーデータのparquetファイルを用意する機会がありました。

Pythonスクリプトを書いたので、備忘録にメモします。

※ソースコードはこちらに置いています: github.com

Parquetファイルのダミーデータを生成する

Parquetファイルのダミーデータを生成するPythonスクリプト例です。 圧縮形式はここではgzipとしています。

from typing import Dict, List
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import boto3


def create_dummy_data(rows: int = 1000) -> pd.DataFrame:
    data: Dict[str, List] = {
        'id': range(1, rows + 1),
        'name': [f'name_{i}' for i in range(1, rows + 1)],
        'value': [i * 100 for i in range(1, rows + 1)]
    }
    return pd.DataFrame(data)


def save_gz_parquet(df: pd.DataFrame, file_path: str) -> None:
    table: pa.Table = pa.Table.from_pandas(df)
    pq.write_table(table, file_path, compression='gzip')


def main() -> None:
    dummy_data: pd.DataFrame = create_dummy_data()

    file_path: str = 'dummy_data.parquet.gz'
    save_gz_parquet(dummy_data, file_path)


if __name__ == '__main__':
    main()

pyarrow.parquetを使うことで、pandas DataFrameから簡単にparquetファイルを書き出すことができます。

上記のコード例では、create_dummy_data関数でダミーのpandas DataFrameを作成し、save_gz_parquet関数でそのデータをparquetファイルで保存しています。

補足:S3に配置する

作成したparquetファイルをそのままS3に配置するには、下記のような形になります。

from typing import Dict, List
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import boto3


def create_dummy_data(rows: int = 1000) -> pd.DataFrame:
    data: Dict[str, List] = {
        'id': range(1, rows + 1),
        'name': [f'name_{i}' for i in range(1, rows + 1)],
        'value': [i * 100 for i in range(1, rows + 1)]
    }
    return pd.DataFrame(data)


def save_gz_parquet(df: pd.DataFrame, file_path: str) -> None:
    table: pa.Table = pa.Table.from_pandas(df)
    pq.write_table(table, file_path, compression='gzip')


def upload_to_s3(bucket: str, s3_key: str, file_path: str) -> None:
    s3 = boto3.client('s3')
    s3.upload_file(file_path, bucket, s3_key)


def main() -> None:
    dummy_data: pd.DataFrame = create_dummy_data()

    file_path: str = 'dummy_data.parquet.gz'
    save_gz_parquet(dummy_data, file_path)

    bucket_name: str = 'your-bucket-name'
    # ".parquet.gz" doesn't work in S3 select.
    s3_key: str = 'path/to/dummy_data.gz.parquet'
    upload_to_s3(bucket_name, s3_key, file_path)


if __name__ == '__main__':
    main()

boto3のupload_fileでシンプルにファイルをS3にアップロードしています。

ちなみに、ファイルの拡張子を.parquet.gzとしてしまうと、S3 Selectがうまく機能しないので注意が必要です。 (ファイル全体をgzipされたものとしてS3に判定されてしまい、S3 Selectできない)

おわりに

以上、Pythonでparquetファイルのダミーデータを生成する方法の簡単な備忘録でした。

拡張子を.parquet.gzにするとS3 Selectdできない、というのは盲点でした。

参考になれば幸いです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

参考

raspberry-pi-examples/put_parquet_gz_to_s3 at main · bioerrorlog/raspberry-pi-examples · GitHub

GPTのToken分割を確認する

GPTにて、テキストのToken数を確認する方法をまとめます。

はじめに

ChatGPTでも使われているGPTシリーズは、その入力テキストを'Token'という単位で処理しています。

OpenAI APIの課金はこのToken単位で行われるので、いかに情報をToken効率良く記載するか、も重要な観点になります。

例:GPT-4の課金体系

今回は、このGPTシリーズの実際のToken分割を確認する方法をまとめます。

GPTのToken分割を確認する

GUIで確認する | Tokenizer

手軽かつ分かりやすくToken分割を確認できる方法に、OpenAIが提供しているTokenizerというサイトがあります。

Tokenizer - OpenAI API

テキストを入力するだけで、どの単語がどのようにToken分割されているのかが可視化できます。

TokenizerでToken分割を可視化する例

余談ですが、日本語が英語の倍近いToken数を消費していることが分かります。

Pythonで確認する | tiktoken

tiktokenを使うことで、PythonでToken分割を確認することができます。
(先述のTokenizerも内部ではこのtiktokenが使われているようです)

# インストール
pip install tiktoken


テキストのToken数は、例えば下記のようにして取得できます。

import tiktoken


def num_tokens_from_string(string: str, model_name: str) -> int:
    """Returns the number of tokens in a text string."""
    encoding = tiktoken.encoding_for_model(model_name)
    num_tokens = len(encoding.encode(string))
    return num_tokens

(コードはOpenAI Cookbookを元に改変)

実行例:

text = "This is an example of using a Tokenizer."

# GPT-3でToken数を確認
num_tokens_from_string(text, "gpt-3.5-turbo")
# 10

# GPT-4でToken数を確認
num_tokens_from_string(text, "gpt-4")
# 10

選択できるモデルと、そこで使われるエンコーディングは以下の通りです。

OpenAI Cookbookより

(gpt-3.5-turbogpt-4は同じエンコーディングcl100k_baseが利用されているので、先の例でも両者の結果に差は出ていません)

終わりに

以上、GPTのToken分割を確認する方法をまとめました。

いかに少ない文字数(Token数)でこちらの意図を明示するか、というのは、シンプルで綺麗なコードを書く、という行為に似ていて面白いですね。

どなたかの参考になれば幸いです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

www.bioerrorlog.work

参考

OpenAI API

Pricing

GitHub - openai/tiktoken: tiktoken is a fast BPE tokeniser for use with OpenAI's models.

openai-cookbook/How_to_count_tokens_with_tiktoken.ipynb at main · openai/openai-cookbook · GitHub

Raspberry PiでPyTorchがIllegal instructionエラーを起こす事象の対処法

Raspberry Pi 4でPyTorch2.0が下記エラーを出す事象の解決策をメモします。

Illegal instruction

はじめに

最近、Raspberry PiとChatGPTを組み合わせて何か面白いことができないだろうか、とあれこれ実験しています。

そんな中、Raspberry PiでPyTorch2.0.0(記事執筆時の最新バージョン)をインストールしたところ、import torchしただけでIllegal instructionのエラーが出る事象に遭遇しました。

状況と取り急ぎの対処をメモします。

Raspberry PiでPyTorchがIllegal instruction

起きた事象

PyTorch2.0.0をpip installしてimport torchすると、Illegal instructionのエラーが出る。

pip install torch==2.0.0 torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu
python -c "import torch;print(torch.__version__)"
# Illegal instruction


環境: Raspberry Pi 4 Model B 4GB RAM / 64bit Raspberry Pi OS(bullseye)

$ hostnamectl
   Static hostname: raspberrypi
         Icon name: computer
        Machine ID: 4f8b19cb3280458c99d89xxxxxx
           Boot ID: 275f81g1c1fd4249a49bbxxxxxx
  Operating System: Debian GNU/Linux 11 (bullseye)
            Kernel: Linux 6.1.19-v8+
      Architecture: arm64

対処法

根本解決ではありませんが、PyTorchバージョンを1.13.1に戻せば解消します。

pip install torch==1.13.1 torchvision==0.14.1 torchaudio==0.13.1 --index-url https://download.pytorch.org/whl/cpu
python -c "import torch;print(torch.__version__)"
# 1.13.1

原因はPyTorch最新バージョン2.0.0とRaspberry PiのCPUアーキテクチャの相性の問題だと思いますが、下記のissueでも未だ議論は決着していません。

Illegal instruction (core dumped) : PyTorch 2.0 on Raspberry Pi 4.0 8gb · Issue #97226 · pytorch/pytorch · GitHub

バージョンを戻せば解消する、という元も子もない話ですが、逆にその他もろもろの足掻きでは解消できなかった事例、と受け取ってもらえれば幸いです。

おわりに

バージョンを戻さないと解決しなかった、というただの懺悔のような話になってしまいました。

今後のversionで解決されるのを祈っています。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

www.bioerrorlog.work

参考

Illegal instruction (core dumped) : PyTorch 2.0 on Raspberry Pi 4.0 8gb · Issue #97226 · pytorch/pytorch · GitHub

Illegal instruction (core dumped) on Raspberry Pi 4B · Issue #8 · KumaTea/pytorch-aarch64 · GitHub

SageMakerでlocal training jobが実行できない時の対処法 | Unable to locate credentials

Unable to locate credentials

のエラーで、SageMaker instanceのlocal training jobが実行できないときの対処法をまとめます。

はじめに

SageMakerでは、SageMaker Python SDKを利用して簡単にtraining job/学習ジョブを実行することができます。

SageMaker SDKのtraining jobには、

  • オンデマンドに別インスタンスが立ち上がってジョブを実行する通常ジョブ
  • 作業中のNotebookインスタンス上でジョブを実行するlocalジョブ

の2種類があります。

この後者、local training jobを実行する際に、Unable to locate credentialsエラーが発生してジョブが実行できない現象に遭遇しました。

対処法を見つけたので、備忘録としてまとめておきます。

SageMakerでlocal training jobが実行できない時の対処法

事象

  • SageMaker instanceでlocal training jobを実行した際に、ジョブコンテナでUnable to locate credentialsエラーが発生してジョブが異常終了する

ただし、私はまだ確実なエラー再現手順を特定できてません。
同様の現象は下記のようにいくつか報告があがっています:

原因

  • ジョブ実行コンテナからインスタンスメタデータエンドポイント(169.254.169.254)にアクセスできていないこと

インスタンスメタデータエンドポイント(169.254.169.254)は、インスタンスのメタデータを取得するためのリンクローカルアドレスです。

Localのジョブ実行コンテナがインスタンスに付与されたIAM権限を参照/Token取得する際に、このエンドポイントにアクセスする必要があります。

しかし、このエンドポイントにアクセスできていないことで、ジョブ実行コンテナがIAMのTokenを取得できない = 権限を取得できない = Unable to locate credentials、という状況です。

対処法

  • 下記のようなスクリプトを実行し、ジョブ実行コンテナから169.254.169.254へアクセスできるように設定変更する
#!/bin/bash

# check if we need to configure our docker interface
SAGEMAKER_NETWORK=`docker network ls | grep -c sagemaker-local`
if [ $SAGEMAKER_NETWORK -eq 0 ]; then
  docker network create --driver bridge sagemaker-local
fi

# Get the Docker Network CIDR and IP for the sagemaker-local docker interface.
SAGEMAKER_INTERFACE=br-`docker network ls | grep sagemaker-local | cut -d' ' -f1`
DOCKER_NET=`ip route | grep $SAGEMAKER_INTERFACE | cut -d" " -f1`
DOCKER_IP=`ip route | grep $SAGEMAKER_INTERFACE | cut -d" " -f9`

# check if both IPTables and the Route Table are OK.
IPTABLES_PATCHED=`sudo iptables -S PREROUTING -t nat | grep -c $SAGEMAKER_INTERFACE`
ROUTE_TABLE_PATCHED=`sudo ip route show table agent | grep -c $SAGEMAKER_INTERFACE`

if [ $ROUTE_TABLE_PATCHED -eq 0 ]; then
  # fix routing
  sudo ip route add $DOCKER_NET via $DOCKER_IP dev $SAGEMAKER_INTERFACE table agent
  echo "route tables for Docker setup done"
else
  echo "SageMaker instance route table setup is ok. We are good to go."
fi

if [ $IPTABLES_PATCHED -eq 0 ]; then
  # fix ip table
  sudo iptables -t nat -A PREROUTING  -i $SAGEMAKER_INTERFACE -d 169.254.169.254/32 -p tcp -m tcp --dport 80 -j DNAT --to-destination 169.254.0.2:9081
  echo "iptables for Docker setup done"
else
  echo "SageMaker instance routing for Docker is ok. We are good to go!"
fi

重要なのは下記2つの設定変更部分です。

# fix routing
sudo ip route add $DOCKER_NET via $DOCKER_IP dev $SAGEMAKER_INTERFACE table agent

# fix ip table
sudo iptables -t nat -A PREROUTING  -i $SAGEMAKER_INTERFACE -d 169.254.169.254/32 -p tcp -m tcp --dport 80 -j DNAT --to-destination 169.254.0.2:9081

SageMaker SDKのlocal training jobでは、sagemaker-localという名前のdocker networkが利用されます (存在しない場合は作成する処理をスクリプト冒頭で行っています)。

このsagemaker-localを変更(ip routeの追加とiptablesのdestination変更)することによって、ジョブ実行コンテナから169.254.169.254へアクセスできるようになる -> インスタンスメタデータ取得できるようになる -> IAM権限Tokenを取得できるようになる、で、エラーを解消できました。

※上記スクリプトは、AWS公式が提供しているSageMaker examples内で見つけたセットアップスクリプトから一部を抽出/改変したものです。

おわりに

以上、Unable to locate credentialsエラーでSageMakerのlocal training jobが実行できないときの対処法をまとめました。

再現方法が特定できていないなどモヤモヤが残る部分はありますが、取り急ぎ解決にたどり着くことができました。

思わぬハマりポイントだったので、この記事が同じ目に遭ったどなたかの参考になれば幸いです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

www.bioerrorlog.work

参考

amazon web services - Assume Sagemaker Notebook instance role from Docker container with default network mode - Stack Overflow

Local mode: role chaining/assumed role on notebook instances does not forward correct credentials · Issue #3464 · aws/sagemaker-python-sdk · GitHub

amazon-sagemaker-examples/setup.sh at main · aws/amazon-sagemaker-examples · GitHub

GitHub - aws/sagemaker-python-sdk: A library for training and deploying machine learning models on Amazon SageMaker

Retrieve instance metadata - Amazon Elastic Compute Cloud

Link-local address - Wikipedia

boto3とbotoの違い | AWS SDK for Pythonの歴史を調べる

boto3とbotoの違いや、そもそもの"boto"の由来など、AWS SDK for Pythonの歴史を調べてまとめます。

はじめに

AWS SDK for Pythonとしてboto3をいつも使っています。

ただ、私がAWSを触り始めたときには既に"boto3"になっていたので、"3"というからには無印botoやboto2もあるんだよね..?というモヤっとした疑問を抱えていました。 以前からAWSを知る人にとっては自明かもしれませんか、自分はまだ明確な答えを知りません。

そこで今回はboto3の歴史的背景を調べ、

  • boto3とbotoの違い
  • boto2はどこに行ったのか
  • そもそも"boto"の由来は?

をまとめます。

AWS SDK for Pythonの歴史

boto3とbotoの違い

少し調べると、boto3とは別にbotoがライブラリとして存在していたことが分かります。

このbotoとboto3の違いを調べてみると、stackoverflowにbotoの作者自身が回答を寄せていました:
python - What is the difference between the AWS boto and boto3 - Stack Overflow

ざっくり意訳抜粋すると、

  • botoは2006年から使われていたAWS公式のAWS Python SDK
  • しかしその設計から、AWSサービスが増えるにしたがってメンテが困難になった
  • そこで設計を刷新したboto3が開発された
    • botocoreをベースとする設計
    • AWSとの低レベルなインターフェースは自動生成される
    • これによりクライアント層の設計を綺麗に保てるようになった

ということのようです。
なるほど私が知らなかっただけで、botoは2006年から(boto3がGAする)2015年まで10年近くも活躍していたんですね。 時間の経過とともに設計刷新の必要性が出たのも納得できます。

各GitHub starsの推移からも、この背景が透けて見えてきます:

  • botoは元々人気のあったライブラリ
  • 2015頃からboto3/botocoreが登場
  • 以後boto3の人気は直線状に上昇/botoは役目を終える

boto, boto3, botocoreのGitHub star数推移

boto2はどこに行ったのか

botoとboto3登場の歴史は納得しました。
が、"boto2はどこに行ったのか"という疑問が残ります。

これは"boto2"という言葉がどの文脈で使われているのか、を過去の情報から調べることで推し量ろうと思います。

いろいろ調べて得た結論はこちらです:

「boto2 はbotoのversion2のことを指すらしい」


ドキュメント類の記述からいくつか例示します:

botoのドキュメントには、"これはbotoの古いバージョン(boto2)のドキュメントだから、boto3を見ることをお勧めします"という記述があります:

You are viewing the documentation for an older version of boto (boto2).
boto: A Python interface to Amazon Web Services — boto v2.49.0


boto3のドキュメントにも、"boto2からの移行"という文脈でbotoのversion2のことを"boto2"と呼んでいる様子が見られます:

The rest of this document will describe specific common usage scenarios of Boto 2 code and how to accomplish the same tasks with Boto3.
Migrating from Boto 2.x - Boto3 1.26.127 documentation


以上、歴史的推移をまとめると、

  • 2006年よりbotoが利用されてきた
  • botoのversion2のことをboto2と呼んでいた
  • 2015年、設計刷新したboto3がリリースされた

ということになりそうです。 スッキリしました。

"boto"の由来

ちなみに"boto"という言葉の由来については、boto3のREADMEに記載があります:

Boto (pronounced boh-toh) was named after the fresh water dolphin native to the Amazon river.
GitHub - boto/boto3: AWS SDK for Python

アマゾンカワイルカ/Amazon river dolphin/Inia geoffrensis/通称boto、が由来のようです。

また下記Issueでは、botoの作者自身がその由来をコメントされています:

It was named after the fresh water dolphin native to the Amazon river. I wanted something short, unusual, and with at least some kind of connection to Amazon. Boto seemed to fit the bill 8^)
Why is the project named boto? · Issue #1023 · boto/boto3 · GitHub

(意訳)
アマゾン川に生息する川イルカから名前を付けました。 短くて、特徴的で、何かしらAmazonに関係する名を付けたかったので、Botoはぴったりの名前でした。

Amazon river dolphin (boto) の写真 | 画像はWikipediaより

おわりに

以上、boto3とbotoの違いから、AWS SDK for Pythonの歴史を調べてまとめました。

ふと思い立って発作的に始めた調べものでしたが、なかなか面白いことが知れて満足です。

同じように不思議に思ったどなたかの参考になれば幸いです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

www.bioerrorlog.work

参考

python - What is the difference between the AWS boto and boto3 - Stack Overflow

Elastician

What is the difference between the AWS boto and boto3? - Quora

boto: A Python interface to Amazon Web Services — boto v2.49.0

Migrating from Boto 2.x - Boto3 1.26.127 documentation

GitHub - boto/boto3: AWS SDK for Python

GitHub - boto/boto: For the latest version of boto, see https://github.com/boto/boto3 -- Python interface to Amazon Web Services

GitHub - boto/botocore: The low-level, core functionality of boto3 and the AWS CLI.

Why is the project named boto? · Issue #1023 · boto/boto3 · GitHub

Amazon river dolphin - Wikipedia

Raspberry Piで複数のサーボモーターを制御する

PWM/サーボ制御モジュールを使わずに、複数のサーボモーターを制御する方法の備忘録です。

はじめに

最近またRaspberry Piを触って遊んでいます。

複数のサーボモーターを制御したい、となったとき、世の事例を検索してみると、多くの場合はPWM/サーボ制御モジュールを取り付けることが多そうです。 ただ今回は外部モジュールを使わずに、とりあえず素のGPIOで複数サーボを制御してみました (パーツの用意やはんだ付けが面倒だったので...)。

やり方の備忘録を残します。


[関連記事] www.bioerrorlog.work

Raspberry Piで複数のサーボモータを制御する

OS環境の準備

モデルはRaspberry Pi 4 Model B (4GB RAM)を使いました。

Raspberry PiにはUbuntu 22.04.1 LTS (64-bit)を入れています (Raspberry Pi OSなど他の環境でも特に動作に違いはないはず)。

$ lsb_release -a
No LSB modules are available.
Distributor ID: Ubuntu
Description:    Ubuntu 22.04.1 LTS
Release:    22.04
Codename:   jammy

Raspberry PiにUbuntuを入れる方法は別途記事にしているので、こちらを参照ください: www.bioerrorlog.work

回路を組む

回路は↓のように組みました。

  • サーボモーターにはSG90を使用
  • サーボモーターに供給する外部電源は単4電池4本パック/6Vを使用
    (データシートによるとギリギリSG90の動作範囲)
  • 外部電源とサーボモーターのマイナス端子はRaspberry PiのGNDに接続
  • サーボモーターのPWM制御線とRaspberry PiのGPIOを1kΩ抵抗を挟んで接続
    (今回は17, 18, 27のpinを使用)

サーボモータを動かすスクリプトの用意

では、サーボモーターの動作チェックのための簡単なスクリプトを書きます。 今回はPythonで書きました。

ソースコード: github.com

import math
import time
from gpiozero import AngularServo


def main():
    servo_17 = AngularServo(17)
    servo_18 = AngularServo(18)
    servo_27 = AngularServo(27)

    t = 0
    max_angle = 60
    gap_angle = 30

    try:
        while True:
            servo_17.angle = math.sin(math.radians(t % 360)) * max_angle
            servo_18.angle = math.sin(math.radians((t % 360) + gap_angle)) * max_angle
            servo_27.angle = math.sin(math.radians((t % 360) + (gap_angle * 2))) * max_angle

            t += 2
            time.sleep(0.01)
    except KeyboardInterrupt:
        servo_17.angle = 0
        servo_18.angle = 0
        servo_27.angle = 0


if __name__ == '__main__':
    main()

gpiozeroAngularServoを使って3つのサーボを動かす、動作確認用のちょっとしたスクリプトです。 回転を滑らかにするため、角度の計算にはsin関数を挟んでいます。

サーボモータを動かす

では用意したスクリプトをRaspberry Piで実行します。

スクリプトをcloneして、必要なライブラリをRaspberry Piにインストールします:

# Clone
git clone https://github.com/bioerrorlog/raspberry-pi-examples.git
cd raspberry-pi-examples/multi_servo_gpio/

# Install required packages
sudo apt update
sudo apt upgrade
sudo apt install python3-pip python3-dev gcc git
sudo pip install -r requirements.txt

そしたら、Pythonスクリプトを実行します:

sudo python3 main.py

# sudoで実行するのは、下記エラー回避のため:
# RuntimeError: No access to /dev/mem. Try running as root!

上手く実行できれば、↓みたいな感じでサーボモーターが動きます。

(サーボモーター同士をダンボールで固定しています)

おわりに

Raspberry Piで複数のサーボモーターを制御する方法の備忘録でした。

PWM制御モジュールも使ってないし、ソフト面での工夫もしていないので結構ジッターが入ってしまいますが、とりあえず動く、というところは見ることができます。

以上、備忘録でした。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

www.bioerrorlog.work

参考

O'Reilly Japan - Raspberry Piクックブック 第3版

マイクロサーボ9g SG-90 | 秋月電子通商

GitHub - gpiozero/gpiozero: A simple interface to GPIO devices with Raspberry Pi

GitHub - bioerrorlog/raspberry-pi-examples: Example projects for Raspberry Pi.

Athena経由でpandas DataFrameを作成する

Amazon Athena経由でpandas DataFrameを作成するやり方をまとめます。

はじめに

こんにちは、@bioerrorlogです。

Amazon Athenaクエリ経由でpandas DataFrameを作成したい、としたらどのようなやり方があるでしょうか。 

例えばいちばん愚直にやろうとすれば、boto3でAthenaクエリを発行し、クエリ結果からデータを取得して、上手いことpandasに読み込ませる...とかでしょうか。 そんな複雑なことやりたくないなぁと思って調べてみても、stackoverflowの上位answerでも似たような解決策が示されています。

しかしもう少し調べると、awswranglerを使えばかなりスッキリ書けることを知りました。

今回は、awswranglerを使ってAthena経由でpandas DataFrameを作成するやり方をメモします。


※追記:awswranglerはAWS SDK for pandasに改名されました(使い方は特に変わってない模様です)。

awswrangler (AWS SDK for pandas)とは

awswranglerは、AWSサービスとpandasを連携するツールです。

AWS公式(awslabs)が開発しており、Athenaだけでなく各AWSのDB系サービスとの連携もサポートしています。

Pandas on AWS. Easy integration with Athena, Glue, Redshift, Timestream, OpenSearch, Neptune, QuickSight, Chime, CloudWatchLogs, DynamoDB, EMR, SecretManager, PostgreSQL, MySQL, SQLServer and S3 (Parquet, CSV, JSON and EXCEL).

github.com

Athena経由でpandas DataFrameを作成する

このawswranglerを使えば、次のように簡単にAthenaのクエリ結果をそのままpandas DataFrameに読み込ませることができます。

import awswrangler as wr

df = wr.athena.read_sql_query(sql='SELECT * FROM "<table_name>"', database='<database_name>')

ちなみに上記のコードでは、Athena query resultの出力先を指定していません。 その場合はデフォルトでs3://aws-athena-query-results-ACCOUNT-REGION/にquery resultが出力されますが、任意のS3を指定したければs3_outputから指定することができます。

その他athena.read_sql_queryの各オプションはドキュメントをご覧ください

おわりに

今回は、awswranglerを使ってAthena経由でpandas DataFrameを作成するやり方をメモしました。

素のboto3で書くよりもずっとシンプルに書くことができるので、なかなか使い勝手が良いのではないでしょうか。 awslabs公式のツールであることも嬉しいポイントです。

以上、どなたかの参考になれば幸いです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

参考

GitHub - aws/aws-sdk-pandas: Pandas on AWS - Easy integration with Athena, Glue, Redshift, Timestream, Neptune, OpenSearch, QuickSight, Chime, CloudWatchLogs, DynamoDB, EMR, SecretManager, PostgreSQL, MySQL, SQLServer and S3 (Parquet, CSV, JSON and EXCEL).

Quick Start — AWS SDK for pandas 2.17.0 documentation

awswrangler.athena.read_sql_query — AWS SDK for pandas 2.17.0 documentation

python - How to Create Dataframe from AWS Athena using Boto3 get_query_results method - Stack Overflow

boto3でエラーハンドリングする方法をまとめる | AWS SDK for Python

boto3でエラーハンドリングする方法を整理します。

はじめに

こんにちは、@bioerrorlogです。

boto3で発生するエラーをcatchしようとしたときに、少しハマったことがありました。

例えば以下のコードを実行すると、examplebucketという名前のS3バケットが既に存在しているため、BucketAlreadyExistsエラーが吐かれます。

import boto3


client = boto3.client('s3')
client.create_bucket(
    Bucket='examplebucket',
    CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1',},
)
botocore.errorfactory.BucketAlreadyExists: An error occurred (BucketAlreadyExists) when calling the CreateBucket operation: The requested bucket name is not available.
The bucket namespace is shared by all users of the system. Please select a different name and try again.

では、このエラーをcatchするにはどうすればいいでしょうか。

エラーメッセージに従って愚直にbotocore.errorfactory.BucketAlreadyExistsをcatchする例外処理を書いてしまうと、上手く動きません。

import boto3
import botocore


client = boto3.client('s3')
try:
    client.create_bucket(
        Bucket='examplebucket',
        CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1',},
    )
except botocore.errorfactory.BucketAlreadyExists as e:
    print(e)
AttributeError: module 'botocore.errorfactory' has no attribute 'BucketAlreadyExists'

今回は、このようなケースでboto3エラーをハンドリングする方法をまとめます。

boto3でエラーハンドリングする

boto3でエラーハンドリングするやり方は、大きく分けて2つあります。

  • client.exceptions
  • botocore.exceptions

client.exceptions

client.exceptionsを使えば、とてもシンプルに例外処理を書くことが出来ます。 宣言したboto3 clientから、client.exceptions.<エラー名>でcatchします。

import boto3


client = boto3.client('s3')
try:
    client.create_bucket(
        Bucket='examplebucket',
        CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1',},
    )
except client.exceptions.BucketAlreadyExists as e:
    print(e.response['Error'])

なお、boto3 clientではなくresourceを宣言している場合はresource.meta.client.exceptions.<エラー名>でcatchできます。

import boto3


resource = boto3.resource('s3')
try:
    resource.create_bucket(
        Bucket='examplebucket',
        CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1',},
    )
except resource.meta.client.exceptions.BucketAlreadyExists as e:
    print(e.response['Error'])

ただし、全てのエラーがこのやり方でcatch出来るわけではない、という点に注意が必要です。 対応できるエラーは、boto3ドキュメントの各APIごとに記載があります

一方、次に紹介するbotocore.exceptionsを使ったやり方では、全てのエラーをcatchできます。

botocore.exceptions

botocore.exceptionsを使う場合は少し冗長な書き方になってしまいますが、全ての例外を処理できるメリットがあります。

import botocore
import boto3


client = boto3.client('s3')
try:
    client.create_bucket(
        Bucket='examplebucket',
        CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1',},
    )
except botocore.exceptions.ClientError as e:
    if e.response['Error']['Code'] == 'BucketAlreadyExists':
        print(e.response['Error'])
    else:
        raise e

botocore.exceptions.ClientErrorでエラーをcatchしたのち、エラーのresponse['Error']['Code']から対応を振り分けます。

コード量が少し多くなってしまうのでなるべくclient.exceptionsを使ったやり方で書きたいところですが、やむを得ない場合はこちらで実装しましょう。

コード例

では、いくつかboto3で例外処理をするコード例を書いてみます。 xxxAlreadyExists系エラーやNoSuchEntitiy系のエラーはcatchしたいケースが多いので、これら中心にメモします。


S3 create_bucketにおけるBucketAlreadyExistsエラーのハンドリング例

import boto3


client = boto3.client('s3')
try:
    client.create_bucket(
        Bucket='examplebucket',
        CreateBucketConfiguration={'LocationConstraint': 'ap-northeast-1',},
    )
except client.exceptions.BucketAlreadyExists as e:
    print(f'S3 bucket already exists: {e.response["Error"]["BucketName"]}')


S3 get_objectにおけるNoSuchKeyエラーのハンドリング例

import boto3


client = boto3.client('s3')
try:
    client.get_object(
        Bucket='target-bucket-001',
        Key='no-such/ocject-key',
    )
except client.exceptions.NoSuchKey as e:
    print(f'No such key: {e.response["Error"]["Key"]}')


IAM get_roleにおけるNoSuchEntityExceptionエラーのハンドリング例

import boto3


client = boto3.client('iam')
try:
    client.get_role(
        RoleName='NoSuchRoleName'
    )
except client.exceptions.NoSuchEntityException as e:
    print(e.response['Error']['Message'])


Glue create_databaseにおけるAlreadyExistsExceptionエラーのハンドリング例

import boto3


client = boto3.client('glue')
try:
    client.create_database(
        DatabaseInput={
            'Name': 'test_db'
        }
    )
except client.exceptions.AlreadyExistsException as e:
    print(e.response['Error']['Message'])

おわりに

以上、boto3でエラーハンドリングする方法を整理しました。

上手にエラーハンドリングして、デバッグしやすいコードを心掛けたいものです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

www.bioerrorlog.work

参考

Error handling — Boto3 Docs 1.26.14 documentation

python - Properly catch boto3 Errors - Stack Overflow

python - How to handle errors with boto3? - Stack Overflow

人工生命 "Lenia" を動かす | ALife

人工生命 "Lenia" を触って遊びます。

はじめに

"Lenia" という美しい人工生命モデルがあります。

このデモ動画を見てみて下さい。

顕微鏡を覗いているような、"何か"が生きているかのような動きが美しいですね。

今回は、この人工生命"Lenia"を手元で動かして遊んでみます。

作業環境

Windows 10
Python 3.7.6
Git bash

で作業しました。

人工生命 "Lenia" を動かす

Leniaとは

Lenia is a system of continuous cellular automata, a form of artificial life. It was derived from Conway's Game of Life by making everything smooth, continuous and generalized.

Lenia Portalより

(意訳) Leniaは、連続値セルオートマトンの人工生命です。 コンウェイのライフゲームから、あらゆる要素をスムーズで連続的に汎化して作られています。

とのことで、ライフゲームベースの時空間的に連続な人工生命モデルとして、Bert Chan氏が作成したものです。

論文もあるので、詳細はこちらを参照ください。

Lenia - Biology of Artificial Life

https://direct.mit.edu/isal/proceedings/isal2020/32/221/98400

Leniaを動かす

それではLeniaを動かしていきます。

github.com

Python, Matlab, Javascriptなど複数の言語がサポートされていますが、今回はPythonバージョンを動かしてみます。

# ソースコードをclone
$ git clone https://github.com/Chakazul/Lenia.git

# Pythonバージョンのディレクトリに移動
$ cd Lenia/Python

# virtualenvがインストールされていない場合はインストールする
$ pip install virtualenv

# 仮想環境作成
$ virtualenv .venv

# 仮想環境有効化
$ source .venv/Scripts/activate

# パッケージインストール
$ pip install -r requirements.txt

これで準備は完了です。

いよいよLeniaを実行してみます。

$ python LeniaND.py 

このようにLeniaのウィンドウが表示されれば成功です。

いろんなLenia

ウィンドウ上部のタブからパラメータを操作すると、いろいろな形態を表現することが出来ます。

いくつか例を紹介します。

非常に多様な形態が創発されて面白いですね。

ぜひいろいろ遊んでみてください。

おわりに

今回は人工生命 "Lenia" を動かして遊びました。

セルオートマトンは古くより研究されてきた分野ですが、このLeniaのようにいかにも"生き物っぽい"動きを目の当たりにすると、この分野の奥深さを実感します。

人工生命のモデルは面白そうなものがたくさんあるので、いろいろと触ってみたいものです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

www.bioerrorlog.work

参考

GitHub - Chakazul/Lenia: Lenia - Mathematical Life Forms

Lenia Portal

Lenia - Biology of Artificial Life

https://direct.mit.edu/isal/proceedings/isal2020/32/221/98400

Stanford Seminar - Lenia: Biology of Artificial Life | YouTube

Bert Chan

boto3 clientのendpoint URLを確認する | AWS SDK for Python

作成済みboto3 clientのendpoint URLを確認する方法をまとめます。

はじめに

こんにちは、@bioerrorlogです。

boto3クライアントの宣言時には、endpoint URLを独自のURLに上書きすることが出来ます。

import boto3


client = boto3.client('s3', endpoint_url='http://my.endpoint:4566')

これはLocalStackなどを使う際に特に便利で、boto3がアクセスするエンドポイントを実際のAWSエンドポイントから独自のURLに振り向けることが出来ます。

しかし逆に、設定されているエンドポイントURLをclientから取得する方法はパッと思い浮かびませんでした。 もしそれが出来れば、デバッグ時に重宝しそうです。

今回は、作成済みboto3 clientから、設定されているendpoint URLを確認する方法まとめます。

環境

boto3 1.18.31 で動作確認してます。

boto3 clientのendpoint URLを確認する

作成したboto3 clientのendpoint URLを確認するには、(少なくとも)以下のふたつの方法があります。

  • client.meta.endpoint_url
  • client._endpoint.host

client.meta.endpoint_url

まず一つ目のやり方は、client.meta.endpoint_urlです。

import boto3


client = boto3.client('s3')
print(client.meta.endpoint_url)
# 出力例: https://s3.ap-northeast-1.amazonaws.com

client = boto3.client('s3', endpoint_url='http://localstack:4566')
print(client.meta.endpoint_url)
# 出力: http://localstack:4566

このようにboto3クライアントに設定されたendpoint URLを都度取得することが出来ます。

※ botocoreにおけるmeta.endpoint_url実装部分のソースコードはこちら: botocore/client.py at 6451ae1fad57f4453af97649e7ed9192b0f623be · boto/botocore · GitHub

client._endpoint.host

もう一つのやり方は、client._endpoint.hostです。

import boto3


client = boto3.client('s3')
print(client._endpoint.host)
# 出力例: https://s3.ap-northeast-1.amazonaws.com

client = boto3.client('s3', endpoint_url='http://localstack:4566')
print(client._endpoint.host)
# 出力: http://localstack:4566

client.meta.endpoint_urlと同様、boto3クライアントに設定されたendpoint URLを取得することが出来ます。

ただ、アンダースコアで始まる属性値_endpointに直接アクセスするのは憚られるので、client.meta.endpoint_urlでアクセスした方が自然に思います。

※ botocoreにおける_endpoint.host実装部分のソースコードはこちら: botocore/client.py at 6451ae1fad57f4453af97649e7ed9192b0f623be · boto/botocore · GitHub

おわりに

以上、boto3 clientのendpoint URLを確認する方法をまとめました。

なかなかboto3ドキュメントを漁っても該当のやり方を見つけることが出来ず、やり方を調べるのも一苦労でした。 (ネット上のコード例を探したうえで、boto3/botocoreソースコードを辿る必要がありました)

どなたかの参考になれば幸いです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

www.bioerrorlog.work

参考

botocore/client.py at 6451ae1fad57f4453af97649e7ed9192b0f623be · boto/botocore · GitHub

botocore/client.py at 6451ae1fad57f4453af97649e7ed9192b0f623be · boto/botocore · GitHub

Add environment variable to override endpoint_url (#2099) by rwillmer · Pull Request #2746 · boto/boto3 · GitHub

django models - Upload to Amazon S3 using Boto3 and return public url - Stack Overflow

GitHub - localstack/localstack: 💻 A fully functional local AWS cloud stack. Develop and test your cloud & Serverless apps offline!

Pythonでユーザー定義クラスのオブジェクトを等価比較する

Pythonで独自に定義したユーザー定義クラスのオブジェクト同士を等価比較する方法を整理します。

はじめに

こんにちは、@bioerrorlogです。

Pythonでは、ユーザー定義クラスは等価演算子で比較することが出来ません。

例えば、次のようなクラス/pytestコードを定義した場合、オブジェクト比較部分のassertでエラーが発生します。

class MyClass:
    def __init__(self, amount):
        self.amount = amount


def test_equality():
    obj_1 = MyClass(5)
    obj_2 = MyClass(5)
    assert obj_1 == obj_2 # これが出来ない


pytest実行結果:

=============================== FAILURES ================================
_____________________________ test_equality _____________________________

    def test_equality():
        obj_1 = MyClass(5)
        obj_2 = MyClass(5)
>       assert obj_1 == obj_2
E       assert <equality.MyClass object at 0x0000014F05071848> == <equality.MyClass object at 0x0000014F05071C88>

equality.py:14: AssertionError
======================== short test summary info ========================
FAILED equality.py::test_equality - assert <equality.MyClass object at ...
=========================== 1 failed in 0.03s ===========================


今回は、上のように独自に定義したオブジェクト同士を比較できるようにする方法を記録します。

動作環境

Python 3.7.6 で動作確認しています。

Pythonでユーザー定義クラスのオブジェクトを等価比較する

ユーザー定義クラスのオブジェクトを等価比較できるようにするには、特殊メソッド__eq__を利用します。

__eq__メソッドは、等価演算子==が使われたときに呼び出されるものです。
例えば x==y が実行されたときには x.__eq__(y)が呼び出されます。
※参考: 3. Data model — Python 3.10.5 documentation

以下、やり方を見ていきます。

__eq__メソッドでインスタンス変数を比較する

まず、__eq__メソッドでインスタンス変数を比較することでオブジェクトを等価比較できるようになります。

実装例:

class MyClass:
    def __init__(self, amount):
        self.amount = amount

    def __eq__(self, other):
        return self.amount == other.amount # インスタンス変数を比較

def test_equality():
    obj_1 = MyClass(5)
    obj_2 = MyClass(5)
    assert obj_1 == obj_2 # このテストは通る


ただ、このやり方だとインスタンス変数が変更されるたびに__eq__メソッドを更新する必要が出てきます。 そこで次は__dict__を利用する方法を紹介します。

__eq__メソッドで__dict__を比較する

__dict__は、インスタンス変数がdictで格納されている特殊属性です。 これを__eq__メソッドで比較することで、インスタンス変数をひとつひとつ比較せずにも等価比較が可能です。

実装例:

class MyClass:
    def __init__(self, amount):
        self.amount = amount

    def __eq__(self, other):
        return self.__dict__ == other.__dict__ # インスタンス変数を比較

def test_equality():
    obj_1 = MyClass(5)
    obj_2 = MyClass(5)
    assert obj_1 == obj_2 # このテストは通る

同一クラスかを比較する

上に挙げたやり方では、インスタンス変数を比較しているのみなのでクラスの種類までは比較できていません。 なので、仮に全く同じインスタンス変数を持つ別々のクラス同士を比較した場合も、同一のものとして判定されてしまいます:

class MyClass:
    def __init__(self, amount):
        self.amount = amount

    def __eq__(self, other):
        return self.__dict__ == other.__dict__

class DummyClass:
    def __init__(self, amount):
        self.amount = amount

    def __eq__(self, other):
        return self.__dict__ == other.__dict__


def test_equality():
    obj_1 = MyClass(5)
    obj_2 = DummyClass(5)
    assert obj_1 == obj_2 # このテストが通る


異なるクラスのオブジェクトを異なるものとして判定させるためには、__eq__メソッド内でクラス__class__を比較させます。 __class__は、クラス名が格納されている特殊属性です。

実装例:

class MyClass:
    def __init__(self, amount):
        self.amount = amount

    def __eq__(self, other):
        return (
            isinstance(other, self.__class__) and
            self.__dict__ == other.__dict__
        )


class DummyClass:
    def __init__(self, amount):
        self.amount = amount

    def __eq__(self, other):
        return (
            isinstance(other, self.__class__) and
            self.__dict__ == other.__dict__
        )


def test_equality():
    obj_1 = MyClass(5)
    obj_2 = MyClass(5)
    assert obj_1 == obj_2 # このテストは通る
    
    obj_3 = DummyClass(5)
    assert obj_1 == obj_3 # このテストは通らない

同一クラスでない時はNotImplementedを投げる

同一クラスでない時には、等価演算がサポートされていない意を示すNotImplementedを投げる実装もできます。

実装例:

class MyClass:
    def __init__(self, amount):
        self.amount = amount

    def __eq__(self, other):
        if not isinstance(other, self.__class__):
            return NotImplemented
        return self.__dict__ == other.__dict__

class DummyClass:
    def __init__(self, amount):
        self.amount = amount

    def __eq__(self, other):
        if not isinstance(other, self.__class__):
            return NotImplemented
        return self.__dict__ == other.__dict__


def test_equality():
    obj_1 = MyClass(5)
    obj_2 = MyClass(5)
    assert obj_1 == obj_2 # このテストは通る
    
    obj_3 = DummyClass(5)
    assert obj_1 == obj_3 # このテストは通らない

おわりに

以上、Pythonでユーザー定義クラスのオブジェクト同士を等価比較する方法をまとめました。

今回取り上げた問題は、Kent BeckのTDD本をPythonでやり直しているときに遭遇したものです。

Java/JUnitではassertEquals()ですぐオブジェクトを比較できたものが、Pythonだとちょっとした工夫が必要なことに気付き、備忘録を書きました。

どなたかの参考になれば幸いです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

参考

3. Data model — Python 3.10.5 documentation

Built-in Constants — Python 3.10.5 documentation

python - Check if two objects have equal content in Pytest - Stack Overflow

Is there a way to check if two object contain the same values in each of their variables in python? - Stack Overflow

Assert custom objects are equal in Python unit test - Gems

aws-cli/__init__.py at develop · aws/aws-cli · GitHub

aws-cli/test_alias.py at develop · aws/aws-cli · GitHub

Pythonでメモリ/ディスク容量を確認する

Pythonを用いて、実行環境のメモリおよびディスク容量を確認する方法をメモします。

はじめに

おはよう。@bioerrorlogです。

あるとき、Pythonを用いて実行環境のメモリおよびディスク容量を確認する必要に迫られました。

そのときに使ったやり方を備忘録に残します。

環境

OS: Linux (Amazon Linux 2)
Python 3.8

ディスク/メモリ容量を取得する

ディスク容量

ディスク容量を確認するには、以下のコードを実行します。

import shutil

total, used, free = shutil.disk_usage('/')

print(f'Total: {total / (2**30)} GiB')
print(f'Used: {used / (2**30)} GiB')
print(f'Free: {free / (2**30)} GiB')

# 10の9乗をギガとする場合
# print(f'Total: {total / (10**9)} GB')
# print(f'Used: {used / (10**9)} GB')
# print(f'Free: {free / (10**9)} GB')

Pythonの標準ライブラリshutilを用いて、ディスク容量を確認します。

shutil.disk_usage(path)path部分に任意のパスを与えることで、そのパス割り当てのディスク総量/使用済み容量/空き容量を取得できます。

※shutil.disk_usageのドキュメント

Return disk usage statistics about the given path as a named tuple with the attributes total, used and free, which are the amount of total, used and free space, in bytes. path may be a file or a directory.

メモリ容量

続いて物理メモリ容量を確認するには、以下のコードを実行します。

import os
mem_bytes = os.sysconf('SC_PAGE_SIZE') * os.sysconf('SC_PHYS_PAGES')
print(f'Memory total: {mem_bytes / (2**30)} GiB')

# 10の9乗をギガとする場合
# print(f'Memory total: {mem_bytes / (10**9)} GB')

メモリのページサイズos.sysconf('SC_PAGE_SIZE')と、物理メモリのページ数os.sysconf('SC_PHYS_PAGES')から、物理メモリの総容量が算出できます。

※Windows環境ではこのsysconfメソッドは利用できないのでご注意ください。

おわりに

以上、Pythonでメモリおよびディスク容量を調べる方法を記しました。

ローカルの端末あるいは普通のサーバであれば、システム情報やターミナルからメモリ/ディスク容量を確認できるかと思います。 ただ、何かの事情でPythonの実行環境しか渡されない場合、Pythonからこれら情報を取得する必要がでてきます。

そうしたケースに参考にしていただければ幸いです。

[関連記事]

www.bioerrorlog.work

www.bioerrorlog.work

www.bioerrorlog.work

参考

macos - Get hard disk size in Python - Stack Overflow

linux - Get total physical memory in Python - Stack Overflow

shutil — High-level file operations — Python 3.9.6 documentation

os — Miscellaneous operating system interfaces — Python 3.9.6 documentation

sysconf(3): config info at run time - Linux man page