はじめに
「現場で活用するためのAIエージェント実践入門」の第5章で私がつまずいたことのメモです。
（このメモのほかの章へ：1章 / 2章 / 3章 / 4章 / 5章 / 6章 / 7章 / 8章 / 9章 / 10章 / まとめ）
第5章 データ分析者を支援する
この章では、データ分析を手伝ってくれるAIエージェントを作っていきます。本で掲載されているソースコードはGitHubで公開されています。
5.1 データ分析を伴う業務
私はデータ分析にそれほど明るくないので（自然言語処理をやっていたのでテキストマイニングに少し関わった程度）勉強になりました。特にコラムで紹介されていた「Data Interpreter」が、自然文の指示から動的に有向非巡回グラフを作成して、それに基づきコードを自動生成＆実行してくれるというのは興味深いです。
5.2 データ分析エージェントの設計
ここからは、ステップを踏んでAIエージェントを作っていきます。
5.3 実装準備
今回使うコード実行用のサンドボックス「E2B」は、アカウントを作ると無料のHobbyプランで100ドル分のクレジットをもらえます。私がこの章の内容をいろいろ試した時は1ドルくらいしか消費しなかったので、勉強にはHobbyプランで十分です。サイトのデザインが昔のMacintoshのUIを彷彿とさせるものでいい感じです。
テンプレートエンジン「Jinja2」は、テンプレートに分岐やループのロジックを埋められるのが便利です。
なお、プロジェクト名は「Jinja」なのですが、パッケージ名がJinja2なので「Jinja2」とも呼ばれているようです。でも、すでにバージョンは3.xなのでややこしいですね。最新版はJinja v3系ですが、パッケージ名はJinja2のままです。詳細はこちらです。
データ概要を返すdescribe_dataframe()は、本では説明されていませんが、GitHubのscripts/03_describe_dataframe.pyで試すことができます。実行するとこんな感じになりました。
2025-11-02 15:59:38.562 | INFO | __main__:main:39 - ```python
>>> df.info()
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 500 entries, 0 to 499
Data columns (total 19 columns):
# Column Non-Null Count Dtype
--- ------ -------------- -----
0 campaign_id 500 non-null object
1 channel_id 500 non-null object
2 campaign_type 500 non-null object
3 start_date 500 non-null object
4 end_date 500 non-null object
5 user_id 500 non-null object
6 login_at 500 non-null object
7 login_interval 500 non-null int64
8 purchase_date 500 non-null object
9 purchase_amount 500 non-null float64
10 purchase_items 500 non-null int64
11 click_date 500 non-null object
12 click_count 500 non-null int64
13 conversion_rate 500 non-null float64
14 score 500 non-null float64
15 is_successful 500 non-null bool
16 weekday 500 non-null object
17 time_of_day 500 non-null object
18 week_of_month 500 non-null int64
dtypes: bool(1), float64(3), int64(4), object(11)
memory usage: 70.9+ KB
>>> df.sample(5).to_markdown()
| | campaign_id | channel_id | campaign_type | start_date | end_date | user_id | login_at | login_interval | purchase_date | purchase_amount | purchase_items | click_date | click_count | conversion_rate | score | is_successful | weekday | time_of_day | week_of_month |
|----:|:--------------|:--------------|:----------------|:-------------|:-----------|:----------|:-----------|-----------------:|:----------------|------------------:|-----------------:|:-------------|--------------:|------------------:|--------:|:----------------|:----------|:--------------|----------------:|
| 197 | CAMP_4 | Email | Bundle Offer | 2024-06-10 | 2024-06-17 | USER_843 | 2024-06-05 | 29 | 2024-06-06 | 143.35 | 5 | 2024-06-05 | 9 | 11.11 | 4.32 | True | Wednesday | Morning | 1 |
| 440 | CAMP_1 | Display Ads | Discount | 2024-06-19 | 2024-06-25 | USER_497 | 2024-06-10 | 15 | 2024-06-10 | 0 | 0 | 2024-06-08 | 1 | 100 | 9.91 | True | Monday | Evening | 2 |
| 477 | CAMP_10 | Email | Bundle Offer | 2024-06-21 | 2024-06-23 | USER_67 | 2024-06-30 | 29 | 2024-07-03 | 377.6 | 9 | 2024-06-30 | 7 | 14.29 | 13.18 | True | Sunday | Morning | 5 |
| 351 | CAMP_5 | Display Ads | Discount | 2024-06-07 | 2024-06-09 | USER_231 | 2024-06-26 | 19 | 2024-07-01 | 35.81 | 1 | 2024-06-25 | 7 | 0 | 0.89 | False | Wednesday | Afternoon | 4 |
| 21 | CAMP_9 | Search Engine | Discount | 2024-06-06 | 2024-06-13 | USER_102 | 2024-06-19 | 26 | 2024-06-24 | 363.14 | 10 | 2024-06-18 | 4 | 25 | 27.99 | True | Wednesday | Morning | 3 |
>>> df.describe()
| | login_interval | purchase_amount | purchase_items | click_count | conversion_rate | score | week_of_month |
|:------|-----------------:|------------------:|-----------------:|--------------:|------------------:|----------:|----------------:|
| count | 500 | 500 | 500 | 500 | 500 | 500 | 500 |
| mean | 15.264 | 161.215 | 3.566 | 5.422 | 26.7615 | 10.3128 | 2.696 |
| std | 8.52994 | 167.582 | 3.61937 | 2.7942 | 26.1302 | 7.81839 | 1.28798 |
| min | 1 | 0 | 0 | 1 | 0 | 0.22 | 1 |
| 25% | 8 | 0 | 0 | 3 | 12.5 | 4.5025 | 2 |
| 50% | 15 | 107.955 | 2.5 | 5.5 | 16.67 | 8.71 | 3 |
| 75% | 22 | 306.957 | 7 | 8 | 33.33 | 13.7825 | 4 |
| max | 30 | 499.76 | 10 | 10 | 100 | 45.67 | 5 |
「プログラムリスト 5.11」のLLMラッパーは、コストまで計算してくれて便利です。ただ、プロンプトキャッシュによるコスト削減は加味されていないので、実際は安くなることもあります。また、コストを計算するために価格表をメンテするのがちょっと面倒ですね。価格表が更新されたら追従させないといけません。コストもLLMの応答で返してくれると便利なのですが。
5.4 プログラム生成を行うシングルエージェントワークフロー
ここからはプログラマーに相当する部分の実装です。
5.4.1 コード生成（計画）
「プログラムリスト 5.15」のように、プロンプトで「あなたは優秀なデータサイエンティストです。」といったプロフィールを指定することで精度を上げる手法は、次の論文で有名になりました。なお、この論文では、与えるプロフィール情報の自動生成もしています。
その後、複数の専門家を用意して、そこから選抜すると精度が上がるという論文もありました。プロンプトもまだまだ奥が深そうです。
「プログラムリスト 5.16」を試したところ、私の場合はこんなコードを書いてくれました。
def display_dataframe_overview(dataframe):
"""
データフレームの概要を表示します。
Args:
dataframe (pd.DataFrame): 概要を表示したいデータフレーム。
"""
# データフレームのサイズ
print("データフレームのサイズ:", dataframe.shape)
# カラム名とデータ型
print("カラム名とデータ型:")
print(dataframe.dtypes)
# NA値の数
print("各カラムのNA値の数:")
print(dataframe.isnull().sum())
# データの概要を表示
display_dataframe_overview(df)
このコードでは本のようなdf.describe()は使っていませんが、再度実行するとdf.describe()を使うコードが生成されました。注釈24では、再現性を担保したい場合はseedパラメータを指定するようにとの補足がありますが、このメモの「3.1 OpenAI APIの基本」にも書いたように、ランダム性を消すことはできないので注意が必要です。
5.4.2 コード実行（行動）
「プログラムリスト 5.20」は、GitHubのコードだとインデント付きで結果が表示されるようになっていました。以降も、GitHubのコードの方が洗練されているので、そちらも見ながら読み進めるのがおすすめです。
5.4.3 実行結果のレビュー（知覚）
「プログラムリスト 5.21」でレビュー結果を定義するReviewクラスは、GitHubのコードの方がdescriptionをきっちり指定しています。また、その定義を、プロンプト内の出力形式の指定でもそのまま書いています。
### 出力形式
以下の Pydantic.BaseModel のデータ型に従って、構造化されたデータを出力してください。
```python
class Review(BaseModel):
observation: str = Field(
title="実行結果に対する客観的受け止め",
description=(
"まずはコードの実行結果に対する客観的な事実を記述する。例えば「正常に終了し、〇〇という結果を得た。」「エラーが発生した。」などを記述する。"
"その後、コードの実行結果がユーザーから与えられた要求に対して最低限担保できているかを評価する。"
"要求を満たさない場合は、その修正方針を追記する。"
),
)
is_completed: bool = Field(
title="タスク達成条件",
description=(
"実行結果がユーザーから与えられた要求に対して最低限担保できているかを評価する。"
"タスク要求を満たさない場合はFalse、改善点はあれど最低限要求を満たす場合はTrueとする。"
),
)
```
Structured Outputsで指定すればPydantic.BaseModelから派生したクラスの情報は伝わるはずなので冗長な感じもするのですが、LLMにきっちり理解させるためのテクニックなのかもしれません。
また、「プログラムリスト 5.22」は、会話履歴messagesの途中でsystemロールが何度も出てくるのが特徴的です。このような書き方もあるんですね、なんとなく会話の先頭にしか書かないものなのかと思っていました。
messages = [
{"role": "system", "content": system_instruction},
{"role": "user", "content": user_request},
{"role": "assistant", "content": data_thread.code},
*([{"role": "system", "content": results}] if has_results else []),
{"role": "system", "content": f"stdout: {data_thread.stdout}"},
{"role": "system", "content": f"stderr: {data_thread.stderr}"},
{
"role": "user",
"content": "実行結果に対するフィードバックを提供してください。",
},
]
5.4.4 コード生成・コード実行・実行結果のレビューのサイクル実行
「プログラムリスト 5.25」は、GitHubではscripts/08_programmer.pyとscripts/programmer.pyとに分かれています。なお、GitHubのコードをそのまま動かすと、以下のような感じで[Errno 2] No such file or directory: 'outputs/process_id/score_distribution.png'が発生してしまいます。
2025-11-03 01:51:51.141 | DEBUG | src.modules.execute_code:execute_code:16 - execution=Execution(Results: [Result(<PIL.Image.Image image mode=RGBA size=1000x600>), Result(<Figure size 1000x600 with 1 Axes>)], Logs: Logs(stdout: [], stderr: []), Error: ExecutionError(name='FileNotFoundError', value="[Errno 2] No such file or directory: 'outputs/process_id/score_distribution.png'..."
これは、プロンプトでoutputs/process_id以下にファイルを保存するよう指定しているにも関わらず、そのディレクトリが存在しないためです。エラーを回避するため、src/prompts/generate_code.jinjaのプロンプトを修正するのがよさそうです。
<コード生成の制約条件>
(中略)
- グラフや新しいデータは "{{ remote_save_dir }}" のディレクトリ下に適切なファイル名で保存すること。
</コード生成の制約条件>
<コード生成の制約条件>
(中略)
- グラフや新しいデータを保存する際は、必ず最初に "{{ remote_save_dir }}" のディレクトリを作成してから、そのディレクトリ下に適切なファイル名で保存すること。ディレクトリの作成には `import os` と `os.makedirs('{{ remote_save_dir }}', exist_ok=True)` を使用すること。
</コード生成の制約条件>
なお、ここでファイル作成に失敗しているのに画像ファイルが取得できる理由は、E2B Sandboxで動くヘッドレスJupyterサーバーに、Matplotlibでの画像生成処理をフックしてexecution.resultsへ格納する機能があるためです。詳細はこちらです。
5.5 データ分析レポートの作成
プログラマーに相当する部分ができたので、その手前の実行計画立案と、最後のレポート作成を実装します。
5.5.1 分析計画の立案（仮説構築）
「プログラムリスト 5.26」のTaskクラスやPlanクラスも、GitHubのコードではdescriptionやexamplesがしっかり定義されています。
「プログラムリスト 5.29」は、GitHubのコードでは結果がインデント付きで表示されるように改善されていました。
5.5.2 プログラムの実行
「プログラムリスト 5.30」を実行すると、同じ画像が、2ファイル生成されることがあります。あまり深くは追っていないのですが、コードによる保存と、前述のE2B SandboxがMatplotlibの画像生成処理をフックしてexecution.resultsへ格納する機能とで重複してしまうことが原因かもしれません。
5.5.3 実行結果を反映したレポートの作成
プログラムで生成したグラフの画像を、そのままLLMに渡すところがおもしろいです。最近のLLMは画像の理解もお手の物なので、こういったことができる訳ですね。
5.5.4 作成された分析レポート
私の場合はこんな感じのレポートができあがりました。
グラフ付きのレポートができあがるのはインパクトありますね、すごい！
5.6 課題と展望
やってみて分かりましたが、実行するたびに結果は変わりますし、調整も大変そうですし、実務で使うにはいろいろ足りない点も多そうです。本にもあるように、今回作ったものは、AIエージェント実装における第一歩としてのサンプルという位置づけになりそうです。
また、これも本にありますが、あくまでもAIエージェントは手段の一つなので、AIエージェントを導入することが目的になってしまわないよう注意が必要です。
5.7 まとめ
GitHubにはLangGraphによる実装もあるので、そちらも見てみました。
親のグラフはsrc/graph/data_analysis.pyのbuild_data_analysis_graph()で作成していますが、ここではadd_node()だけで、4章で使ったadd_edge()やadd_conditional_edge()は出てきません。子のグラフも同様です。
def build_data_analysis_graph() -> CompiledStateGraph:
checkpointer = InMemorySaver()
graph = StateGraph(DataAnalysisState)
graph.add_node("generate_plan", generate_plan_node)
graph.add_node("approve_plan", approve_plan)
graph.add_node("programmer", build_programmer_graph(_close_programmer))
graph.add_node("open_programmer", open_programmer)
graph.add_node("generate_report", generate_report_node)
graph.set_entry_point("generate_plan")
return graph.compile(checkpointer=checkpointer)
ノード間のエッジは事前定義されておらず、各ノードに対応する関数がCommandオブジェクトを返すことで、動的に次のノードを指定する形になっています。たとえば、generate_planノードの実装はこんな感じです。
def generate_plan_node(state: DataAnalysisState) -> dict:
logger.info("|--> generate_plan")
with open(state["data_file"], "rb") as fi:
file_object = io.BytesIO(fi.read())
data_info = describe_dataframe(
file_object=file_object,
template_file=TEMPLATE_FILE,
)
llm_response = generate_plan(
data_info=data_info,
user_request=state["user_goal"],
)
plan = llm_response.content
tasks = plan.tasks
sub_tasks = [SubTask(state=False, task=task) for task in tasks]
return Command(
goto="approve_plan",
update={
"data_info": data_info,
"sub_tasks": sub_tasks,
"next_node": "approve_plan",
},
)
Commandクラスの詳細はこちらです。
計画の生成後に、人間による承認フェーズ、ヒューマンインザループ（Human-in-the-Loop; HITL）が実装されています。workflow.invoke()でワークフローを開始後、制御が返ってきた時に次のノードがapprove_planなのかどうかで判定しています。
def invoke_workflow(
workflow: CompiledStateGraph,
input_data: dict | Command,
config: dict,
) -> dict:
result = workflow.invoke(
input=input_data,
config=config,
)
logger.debug(result)
if result["next_node"] == "approve_plan":
user_input = str(input("User Feedback: Approval? (y/n): "))
return invoke_workflow(
workflow=workflow,
input_data=Command(resume=user_input),
config=config,
)
return result
なお、通常だとワークフローを開始したら途中で制御は戻りませんが、approve_planノードには、途中で制御を返すためにinterrupt()が入っています。
def approve_plan(state: DataAnalysisState) -> Command[Literal["programmer", "generate_plan"]]:
logger.info("|--> approve_plan")
is_approval = interrupt(
{
"sub_tasks": state["sub_tasks"],
},
)
if is_approval.lower() == "y":
return Command(
goto="open_programmer",
update={
"user_approval": True,
"next_node": "open_programmer",
},
)
return Command(
goto="generate_plan",
update={
"user_approval": False,
"next_node": "generate_plan",
},
)
interrupt()で一度制御が戻るので、Pythonのジェネレータで、yieldで呼び出し元に制御を返して、呼び出し元からsend()で送られた情報を受け取り再開する流れと近い感じです。
ただし、interrupt()の内部実装は例外を送出して関数から戻っているので同じ位置からは再開できません（yieldはPython内部でフレームが保存されて再開時に復元されるので実行位置まで戻れるのです）。再開時はそのノードの頭から再実行になるのが注意点です。再開後にinterrupt()まで来たところで、今度は制御を戻すのではなくCommand(resume=hogehoge)で指定した内容がinterrupt()から返ります。yieldとはこの点が違うので、「yieldと同じね！」と理解するのは危険です。
最後に、グラフ構造を図示しておきます（Claude Codeに書いてもらいました）。
この章はなかなか読み応えがありました。それにしても、グラフ付きのレポートができあがるのはインパクトありますね（2回目）。可能性を感じてワクワクする章でした。
次回は「情報収集者の支援」です。
（このメモのほかの章へ：1章 / 2章 / 3章 / 4章 / 5章 / 6章 / 7章 / 8章 / 9章 / 10章 / まとめ）