LangChainで実装されたAIエージェントでポ◯モンのデータ分析とレポートをやってみたよ
こんにちは、すずきです。
最近、OpenAIが新しいAIエージェントを開発しているという記事が話題になっていました。AutoGPTやBabyAGIが昨年から話題になっていますが、これからAIエージェントがより身近な存在になってきそうです。
この技術動向に少しはキャッチアップしておこうと思い、RAGをつかったチャットボット開発のとき参考にした講座のAIエージェントに関するセクションを観なおしました。
講座で公開されているデータをつかうだけだと面白くないので、ポケモンの種類、能力、技、進化などのデータを管理するDB(SQLite)を設計して、このDBを基に特定の問い合わせを処理するAIエージェントを実装してみました。
AIエージェントとは
AIエージェントは、依頼されたタスクに対して自動で行動計画を立てて、さまざまなツールを活用してそれらのタスクを完遂するシステムです。
これらはLLMを基盤としていて、人間が行うような意思決定プロセスを模倣し、代行する能力をもっています。
DBと接続することで、企業が保有するデータを基に分析を行い、レポートの作成やSlack通知などのタスクを自動で実行できます。
また、AIエージェントは、ただの自動化ツールとは違い、個性、記憶、計画、行動という4つの要素が相互作用することで、より複雑で意思決定が必要な作業も効率的に実行できるよう設計されています。
そのため、ビジネスプロセスの自動化だけでなく、より戦略的な意思決定のサポートにも利用されるみたいです。
実装
pokemon_db.sqliteを用意し、run_query_tool, describe_tables_tool, write_report_toolの3つのツールを駆使して、依頼に沿ったDB分析とHTMLレポートの作成を行うようなエージェントを実装します。
なお、Open AIのAPIキーを.envに設定すれば、気軽に動作確認できるようになっています。
Dockerを利用することで、ローカル環境への影響を避けつつ安全にテストを行うことができるので、よろしければREADME.mdの手順で試してみてください。
DB構造
以下のテーブルを準備し、サンプルデータを何件か入れます。
pokemonsテーブル
内容: ポケモンの基本情報(ID、名前、タイプ、HP、攻撃力、防御力、特殊攻撃力、特殊防御力、速さ)を格納
活用例: 特定のタイプのポケモン検索や、能力値に基づく抽出などに使用
| カラム名 | 型 | 説明 |
|-------------------|----------|--------------------------------------------|
| pokemon_id | INTEGER | 一意のID（プライマリキー） |
| name | TEXT | 名前 |
| type1 | TEXT | 主タイプ |
| type2 | TEXT | 副タイプ（ない場合はNULL） |
| hp | INTEGER | HP |
| attack | INTEGER | 攻撃力 |
| defense | INTEGER | 防御力 |
| sp_attack | INTEGER | 特殊攻撃力 |
| sp_defense | INTEGER | 特殊防御力 |
| speed | INTEGER | 速さ |movesテーブル
内容: 技の詳細情報(技のID、名前、タイプ、カテゴリ、威力、命中率、PP)を格納
活用例: 特定の技を持つポケモンの検出や、タイプ別の技分析に使用
| カラム名 | 型 | 説明 |
|-------------------|----------|-------------------------------------------|
| move_id | INTEGER | 技の一意のID（プライマリキー） |
| name | TEXT | 技の名前 |
| type | TEXT | 技のタイプ |
| category | TEXT | 技のカテゴリ（物理/特殊/変化） |
| power | INTEGER | 技の威力（変化技の場合はNULL） |
| accuracy | INTEGER | 技の命中率（変化技の場合はNULL） |
| pp | INTEGER | 技のPP（使用回数） |pokemon_movesテーブル
内容: ポケモンとそれが覚える技の関連を示す。ポケモンIDと技IDの組み合わせで構成
活用例: 戦略的なバトルプラン策定のための技の組み合わせ分析に利用
| カラム名 | 型 | 説明 |
|-------------------|----------|------------------------------------------|
| pokemon_id | INTEGER | ポケモンのID（外部キー） |
| move_id | INTEGER | 技のID（外部キー） |evolutionsテーブル
内容: ポケモンの進化情報(進化前後のポケモンID、進化方法、必要な条件)を格納
活用例: 進化条件の分析や、特定の進化形態のポケモン抽出に使用
| カラム名 | 型 | 説明 |
|--------------------|---------|-------------------------------------------|
| evolution_id | INTEGER | 進化の一意のID（プライマリキー） |
| base_pokemon_id | INTEGER | 進化前のポケモンID（外部キー） |
| evolved_pokemon_id | INTEGER | 進化後のポケモンID（外部キー） |
| method | TEXT | 進化の方法（例: 「レベルアップ」「特定のアイテム」など） |
| condition | TEXT | 進化の条件（具体的なレベルやアイテム名など、方法によって異なる） |AIエージェントの設計
pokemon_db.sqliteを活用してデータ分析とHTMLレポートの作成を行うAIエージェントを実装します。
main.pyは、LangChainでOpenAIのGPTモデルとの対話、SQLiteへのクエリ実行、および結果のレポート化を行います。
main.pyのコード解説
main.pyでは、LangChainライブラリの様々なコンポーネントを活用してエージェントの動作を定義します。
from langchain.chat_models import ChatOpenAI
from langchain.prompts import (
ChatPromptTemplate,
HumanMessagePromptTemplate,
MessagesPlaceholder
)
from langchain.schema import SystemMessage
from langchain.agents import OpenAIFunctionsAgent, AgentExecutor
from langchain.memory import ConversationBufferMemory
from dotenv import load_dotenv
from tools.sql import run_query_tool, list_tables, describe_tables_tool
from tools.report import write_report_tool
from handlers.chat_model_start_handler import ChatModelStartHandler
# 環境変数の読み込み
load_dotenv()
# チャットモデルスタートハンドラの初期化
handler = ChatModelStartHandler()
chat = ChatOpenAI(callbacks=[handler])
# 利用可能なテーブルのリストを取得
tables = list_tables()
# プロンプトの設定
prompt = ChatPromptTemplate(
messages=[
SystemMessage(
content=(
"You are an AI that has access to a SQLite database.\n"
f"The database has tables of: {tables}\n"
"Do not make any assumptions about what tables exist "
"or what columns exist. Instead, use the 'describe_tables' function"
)
),
MessagesPlaceholder(variable_name="chat_history"),
HumanMessagePromptTemplate.from_template("{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
]
)
# メモリの設定
memory = ConversationBufferMemory(
memory_key="chat_history", return_messages=True)
# ツールの設定
tools = [run_query_tool, describe_tables_tool, write_report_tool]
# エージェントの初期化
agent = OpenAIFunctionsAgent(
llm=chat,
prompt=prompt,
tools=tools
)
# エージェントの実行
agent_executor = AgentExecutor(
agent=agent,
verbose=True,
tools=tools,
memory=memory
)
agent_executor(
"Calculate the average HP for each Pokemon type. Output the results in a HTML report.")環境変数の読み込み
dotenvライブラリを用いて.envファイルから環境変数を読み込み、APIキーなどの設定情報を取得します
チャットモデルとプロンプトの設定
ChatOpenAIオブジェクトを初期化し、ChatModelStartHandlerをコールバックとして設定することで、エージェントがユーザーの入力に対して適切な応答を生成できるようにします
ChatPromptTemplateを用いて、エージェントの対話フローを定義します。ここでは、以下のメッセージタイプを設定しています：
SystemMessage：エージェントがどのように応答すべきかに影響を与えます。エージェントがDBへのアクセス権を持っていることを設定したり、クエリエラーを起こさないように実在するカラムを利用するようにします
MessagesPlaceholder(variable_name="chat_history")：これまでのチャット履歴を保持し、エージェントがコンテキストを理解するのに役立ちます
HumanMessagePromptTemplate.from_template("{input}")：ユーザーの入力を受け取るためのテンプレート
MessagesPlaceholder(variable_name="agent_scratchpad")：エージェントが内部的に情報を一時的に記録するためのスクラッチパッド
DB操作とレポート生成のツール
run_query_tool, describe_tables_tool, write_report_tool：これらのツールを定義し、エージェントがクエリを実行し、テーブルの構造を調べ、結果をHTMLレポートとして出力する機能を実装します
エージェントの実行
OpenAIFunctionsAgentを初期化し、定義したプロンプトとツールを使用して、エージェントがユーザーのクエリに応じた操作を行えるようにします
AgentExecutorを用いて、エージェントの対話セッションを開始し、実際にユーザーからのクエリに対応します
agent_executorを複数回実行するとき、memoryによって過去の対話履歴を参照します
sql.py: DB操作
sql.pyモジュールでは、SQLiteに対する基本的な操作を行うための関数が定義されています。
このモジュールを通じて、AIエージェントはDBからの情報取得や、DB構造の説明などのタスクを実行できます。
conn = sqlite3.connect("pokemon_db.sqlite")
def list_tables():
c = conn.cursor()
c.execute("SELECT name FROM sqlite_master WHERE type='table';")
rows = c.fetchall()
return '\n'.join(row[0] for row in rows if row[0] is not None)
def run_sqlite_query(query):
c = conn.cursor()
try:
c.execute(query)
return c.fetchall()
except sqlite3.OperationalError as err:
return f"The following error occured: {str(err)}"list_tables関数は、DB内の全てのテーブル名をリストアップします。
run_sqlite_query関数は、任意のSQLクエリを実行し、その結果を返します。
これらの関数は、AIエージェントがDBと対話する際に必要となります。
report.py: レポートの生成
report.pyモジュールでは、クエリ結果をもとにHTML形式のレポートを生成し、ファイルに書き出す関数が定義されています。
このモジュールにより、AIエージェントは視覚的にわかりやすい形で分析結果を提供します。
def write_report(filename, html):
with open(filename, 'w') as f:
f.write(html)write_report関数は、指定されたファイル名でHTMLコンテンツをファイルに保存します。
この関数を利用することで、データ分析の結果をレポートとして出力できます。
chat_model_start_handler.py: エージェントの対話管理
chat_model_start_handler.pyモジュールは、AIエージェントがユーザーとの対話を開始する際に特定のアクションを実行するためのコールバック関数を定義します。
このモジュールにより、エージェントは対話の各段階で適切なフィードバックを提供し、ユーザーのクエリに対する応答の生成を管理できます。
def boxen_print(*args, **kwargs):
print(boxen(*args, **kwargs))
class ChatModelStartHandler(BaseCallbackHandler):
def on_chat_model_start(self, serialized, messages, **kwargs):
print("\n\n\n\n========= Senging Messages =========\n\n")
for message in messages[0]:
if message.type == "system":
boxen_print(message.content,
title=message.type, color="yellow")
elif message.type == "human":
boxen_print(message.content, title=message.type, color="green")
elif message.type == "ai" and "function_call" in message.additional_kwargs:
call = message.additional_kwargs["function_call"]
boxen_print(
f"Running tool {call['name']} with args {call['arguments']}",
title=message.type,
color="cyan"
)
elif message.type == "ai":
boxen_print(message.content,
title=message.type, color="blue")
elif message.type == "function":
boxen_print(message.content,
title=message.type, color="purple")
else:
boxen_print(message.content, title=message.type)このクラスは、エージェントが対話を開始する際に呼び出されるon_chat_model_startメソッドを持ち、対話中に送信されるメッセージの種類に応じて異なる処理を行います。
例えば、システムメッセージ、ユーザーメッセージ、AIの応答、関数の実行結果など、メッセージの種類に応じて異なる色で出力を装飾します。
エージェントの動作例
以下がポケモンDBを活用したエージェントへの指示例です。
1. 特定のタイプのポケモン数を調べる
agent_executor(
"How many Electric type Pokemons are there in the database? Write the result to a HTML report.")2. 最も高い攻撃力を持つポケモンを特定する
agent_executor(
"Identify the Pokemon with the highest Attack stat. Present the findings in a HTML report.")3. 特定の技を覚えるポケモンをリストアップする
agent_executor(
"List all Pokemons that can learn 'Thunderbolt'. Generate a report in HTML format.")4. 各タイプのポケモンの平均HPを計算する
agent_executor(
"Calculate the average HP for each Pokemon type. Output the results in a HTML report.")5. 進化可能なポケモンとその方法を表示する
agent_executor(
"Show all Pokemons that can evolve and list their evolution methods. Summarize the information in a HTML report.")6. 特定の条件を満たすポケモンを探索する
agent_executor(
"Find all Pokemons with Speed greater than 100 and HP less than 50. Display the results in a HTML report.")説明のため、試しに4だけ実行してみます。
> Entering new AgentExecutor chain...
========= Senging Messages =========
╭─ system ─────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│You are an AI that has access to a SQLite database. │
│The database has tables of: pokemons │
│moves │
│pokemon_moves │
│evolutions │
│Do not make any assumptions about what tables exist or what columns exist. Instead, use the 'describe_tables' function│
╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ human ────────────────────────────────────────────────────────────────────────────╮
│Calculate the average HP for each Pokemon type. Output the results in a HTML report.│
╰────────────────────────────────────────────────────────────────────────────────────╯
Invoking: `describe_tables` with `{'tables_names': ['pokemons']}`
CREATE TABLE pokemons (
pokemon_id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
type1 TEXT,
type2 TEXT,
hp INTEGER,
attack INTEGER,
defense INTEGER,
sp_attack INTEGER,
sp_defense INTEGER,
speed INTEGER
)
========= Senging Messages =========
╭─ system ─────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│You are an AI that has access to a SQLite database. │
│The database has tables of: pokemons │
│moves │
│pokemon_moves │
│evolutions │
│Do not make any assumptions about what tables exist or what columns exist. Instead, use the 'describe_tables' function│
╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ human ────────────────────────────────────────────────────────────────────────────╮
│Calculate the average HP for each Pokemon type. Output the results in a HTML report.│
╰────────────────────────────────────────────────────────────────────────────────────╯
╭─ ai ───────────────────────────────────╮
│Running tool describe_tables with args {│
│ "tables_names": ["pokemons"] │
│} │
╰────────────────────────────────────────╯
╭─ function ────────────────────────╮
│CREATE TABLE pokemons ( │
│ pokemon_id INTEGER PRIMARY KEY,│
│ name TEXT NOT NULL, │
│ type1 TEXT, │
│ type2 TEXT, │
│ hp INTEGER, │
│ attack INTEGER, │
│ defense INTEGER, │
│ sp_attack INTEGER, │
│ sp_defense INTEGER, │
│ speed INTEGER │
│) │
╰───────────────────────────────────╯
Invoking: `run_sqlite_query` with `{'query': 'SELECT type1, AVG(hp) AS avg_hp FROM pokemons GROUP BY type1'}`
[('Electric', 50.0), ('Normal', 55.0), ('Water', 130.0)]
========= Senging Messages =========
╭─ system ─────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│You are an AI that has access to a SQLite database. │
│The database has tables of: pokemons │
│moves │
│pokemon_moves │
│evolutions │
│Do not make any assumptions about what tables exist or what columns exist. Instead, use the 'describe_tables' function│
╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ human ────────────────────────────────────────────────────────────────────────────╮
│Calculate the average HP for each Pokemon type. Output the results in a HTML report.│
╰────────────────────────────────────────────────────────────────────────────────────╯
╭─ ai ───────────────────────────────────╮
│Running tool describe_tables with args {│
│ "tables_names": ["pokemons"] │
│} │
╰────────────────────────────────────────╯
╭─ function ────────────────────────╮
│CREATE TABLE pokemons ( │
│ pokemon_id INTEGER PRIMARY KEY,│
│ name TEXT NOT NULL, │
│ type1 TEXT, │
│ type2 TEXT, │
│ hp INTEGER, │
│ attack INTEGER, │
│ defense INTEGER, │
│ sp_attack INTEGER, │
│ sp_defense INTEGER, │
│ speed INTEGER │
│) │
╰───────────────────────────────────╯
╭─ ai ────────────────────────────────────────────────────────────────────╮
│Running tool run_sqlite_query with args { │
│ "query": "SELECT type1, AVG(hp) AS avg_hp FROM pokemons GROUP BY type1"│
│} │
╰─────────────────────────────────────────────────────────────────────────╯
╭─ function ─────────────────────────────────────────────╮
│[["Electric", 50.0], ["Normal", 55.0], ["Water", 130.0]]│
╰────────────────────────────────────────────────────────╯
Invoking: `run_sqlite_query` with `{'query': 'SELECT type2, AVG(hp) AS avg_hp FROM pokemons GROUP BY type2'}`
[(None, 71.25)]
========= Senging Messages =========
╭─ system ─────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│You are an AI that has access to a SQLite database. │
│The database has tables of: pokemons │
│moves │
│pokemon_moves │
│evolutions │
│Do not make any assumptions about what tables exist or what columns exist. Instead, use the 'describe_tables' function│
╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ human ────────────────────────────────────────────────────────────────────────────╮
│Calculate the average HP for each Pokemon type. Output the results in a HTML report.│
╰────────────────────────────────────────────────────────────────────────────────────╯
╭─ ai ───────────────────────────────────╮
│Running tool describe_tables with args {│
│ "tables_names": ["pokemons"] │
│} │
╰────────────────────────────────────────╯
╭─ function ────────────────────────╮
│CREATE TABLE pokemons ( │
│ pokemon_id INTEGER PRIMARY KEY,│
│ name TEXT NOT NULL, │
│ type1 TEXT, │
│ type2 TEXT, │
│ hp INTEGER, │
│ attack INTEGER, │
│ defense INTEGER, │
│ sp_attack INTEGER, │
│ sp_defense INTEGER, │
│ speed INTEGER │
│) │
╰───────────────────────────────────╯
╭─ ai ────────────────────────────────────────────────────────────────────╮
│Running tool run_sqlite_query with args { │
│ "query": "SELECT type1, AVG(hp) AS avg_hp FROM pokemons GROUP BY type1"│
│} │
╰─────────────────────────────────────────────────────────────────────────╯
╭─ function ─────────────────────────────────────────────╮
│[["Electric", 50.0], ["Normal", 55.0], ["Water", 130.0]]│
╰────────────────────────────────────────────────────────╯
╭─ ai ────────────────────────────────────────────────────────────────────╮
│Running tool run_sqlite_query with args { │
│ "query": "SELECT type2, AVG(hp) AS avg_hp FROM pokemons GROUP BY type2"│
│} │
╰─────────────────────────────────────────────────────────────────────────╯
╭─ functio─╮
│ [] │
╰──────────╯
Invoking: `write_report` with `{'filename': 'average_hp_report.html', 'html': '<html><body><h1>Average HP for each Pokemon type</h1><table><tr><th>Type</th><th>Average HP</th></tr><tr><td>Electric</td><td>50.0</td></tr><tr><td>Normal</td><td>55.0</td></tr><tr><td>Water</td><td>130.0</td></tr><tr><td>Null</td><td>71.25</td></tr></table></body></html>'}`
None
========= Senging Messages =========
╭─ system ─────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│You are an AI that has access to a SQLite database. │
│The database has tables of: pokemons │
│moves │
│pokemon_moves │
│evolutions │
│Do not make any assumptions about what tables exist or what columns exist. Instead, use the 'describe_tables' function│
╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ human ────────────────────────────────────────────────────────────────────────────╮
│Calculate the average HP for each Pokemon type. Output the results in a HTML report.│
╰────────────────────────────────────────────────────────────────────────────────────╯
╭─ ai ───────────────────────────────────╮
│Running tool describe_tables with args {│
│ "tables_names": ["pokemons"] │
│} │
╰────────────────────────────────────────╯
╭─ function ────────────────────────╮
│CREATE TABLE pokemons ( │
│ pokemon_id INTEGER PRIMARY KEY,│
│ name TEXT NOT NULL, │
│ type1 TEXT, │
│ type2 TEXT, │
│ hp INTEGER, │
│ attack INTEGER, │
│ defense INTEGER, │
│ sp_attack INTEGER, │
│ sp_defense INTEGER, │
│ speed INTEGER │
│) │
╰───────────────────────────────────╯
╭─ ai ────────────────────────────────────────────────────────────────────╮
│Running tool run_sqlite_query with args { │
│ "query": "SELECT type1, AVG(hp) AS avg_hp FROM pokemons GROUP BY type1"│
│} │
╰─────────────────────────────────────────────────────────────────────────╯
╭─ function ─────────────────────────────────────────────╮
│[["Electric", 50.0], ["Normal", 55.0], ["Water", 130.0]]│
╰────────────────────────────────────────────────────────╯
╭─ ai ────────────────────────────────────────────────────────────────────╮
│Running tool run_sqlite_query with args { │
│ "query": "SELECT type2, AVG(hp) AS avg_hp FROM pokemons GROUP BY type2"│
│} │
╰─────────────────────────────────────────────────────────────────────────╯
╭─ functio─╮
│ [] │
╰──────────╯
╭─ ai ───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│Running tool write_report with args { │
│ "filename": "average_hp_report.html", │
│ "html": "<html><body><h1>Average HP for each Pokemon type</h1><table><tr><th>Type</th><th>Average │
│HP</th></tr><tr><td>Electric</td><td>50.0</td></tr><tr><td>Normal</td><td>55.0</td></tr><tr><td>Water</td><td>130.0</td></tr><tr><td>Null</td><td>71│
│.25</td></tr></table></body></html>" │
│} │
╰────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ functio─╮
│ null │
╰──────────╯
I have generated the HTML report for the average HP of each Pokemon type. You can download it from [here](sandbox:/average_hp_report.html).
> Finished chain.解説
この実行例は、AIエージェントがユーザーからの指示に基づいて、ポケモンDBから情報を取得し、分析結果をHTMLレポートとして出力するプロセスを示しています。
以下が各ステップの詳細です。
システムの初期メッセージ
エージェントは、SQLiteへのアクセス権を持っており、DBにはpokemons、moves、pokemon_moves、evolutionsの4つのテーブルが含まれていることをユーザーに通知します
また、テーブルやカラムの存在を前提とせず、必要に応じてdescribe_tables関数を使用するよう指示しています
ユーザーのクエリ
ユーザーは、各ポケモンタイプの平均HPを計算し、結果をHTMLレポートで出力するようエージェントに依頼します
テーブル構造の確認
エージェントはdescribe_tables関数を呼び出して、pokemonsテーブルの構造を取得します。これにより、クエリを正確に生成するために必要なテーブルのカラム情報を確認します
DBクエリの実行
続いて、エージェントはrun_sqlite_query関数を使用して、ポケモンのタイプごとに平均HPを計算するSQLクエリを実行します
クエリの結果として、電気タイプの平均HPが50.0、ノーマルタイプが55.0、水タイプが130.0と計算されます
HTMLレポートの生成
最後に、エージェントはwrite_report関数を呼び出して、クエリ結果を含むHTMLレポートを生成します。レポートには、各ポケモンタイプの平均HPが表形式で記載されています
レポートの提供
エージェントはレポートの生成が完了し、ユーザーがレポートをダウンロードできることを通知します
HTMLファイルをブラウザで開くと以下のようにレポートが作成されています。
<html>
<body>
<h1>Average HP for each Pokemon type</h1>
<table>
<tr>
<th>Type</th>
<th>Average HP</th>
</tr>
<tr>
<td>Electric</td>
<td>50.0</td>
</tr>
<tr>
<td>Normal</td>
<td>55.0</td>
</tr>
<tr>
<td>Water</td>
<td>130.0</td>
</tr>
<tr>
<td>Null</td>
<td>71.25</td>
</tr>
</table>
</body>
</html>採用情報
この記事が気に入ったらサポートをしてみませんか？