ゼロから作って仕組みを理解するAIエージェント

この記事について

昨今「AIエージェント」という言葉が流行し、さまざまな場面で見聞きするようになりました。AIエージェントの作り方を解説した記事も数多くありますが、その多くはフレームワークを用いた実装方法の紹介に寄っています。フレームワークを使えば少ないコードで手早く形にできる一方で、内部で何が起きているのか、どのような仕組みで動いているのかを掴みにくいという課題があります。

そこで本記事では、フレームワークを使わず最小構成でAIエージェントを実装し、動作の「手触り感」を持って仕組みを理解することを目指します。

記事内で紹介している、実装したコードは下記に置いています。
https://github.com/raucha/simple-ai-agent

※ 0からと書いていますが、LLM本体は実装しません(一応)。APIを利用します。

AIエージェントの定義

何をAIエージェントと呼ぶかは色々な説明があり、まだ各人のイメージも揺れているように思いますが、概ね下記のようなものをAIエージェントと呼ぶようです。

与えられた目標を達成するために、状況判断や外部環境とのインタラクションを自律的に行うAIシステム。

最近のAIエージェントという単語が出てくる文脈では、一般的には内部でLLMが利用されます。

LLMとAIエージェントの間にあるギャップ

一方、LLMは突き詰めると 「文字列を入力として受け取り、文字列を出力するステートレスな関数」 というモデルと解釈できます。

LLM単体では外部環境とのインタラクション手段を持たないため自律的に情報を取りに行ったり何かを実行したりすることはできません。また、ステートレスで情報を保持することができないので、例えば「調査して結果をレポートにまとめる」といった複数段に分けた処理ステップを実行することができません。
LLMをAIエージェントとして利用するには、このあたりの課題を解決する必要があります。

文字列しか入出力できないLLMで、どうやって外部環境側とインタラクションするのか？

外部環境側で実行したい処理名とパラメータを構造化(json化)した文字列として出力させ、Pythonインタプリタなどで関数として実行します。また、実行結果を文字列化として再度LLMに返すことでLLMが結果を把握できるようにします。
一般的にtool callingなどと呼ばれる方法です。

具体的には下記のような処理の流れになります。

1. LLMに「実行可能な関数名・利用すべきシーン・入力形式・出力形式」の説明をあらかじめ文字列として渡す。
2. LLMが状況に応じて「関数呼び出し用の文字列」を出力する。
3. LLMの外側で実際に関数を実行する。
4. 実行結果を文字列化して再度LLMに返す。
   (実行結果はweb検索するツールであれば調査結果、ファイル生成するツールであれば生成されたパス、など)

ステートレスなのに、どうやって複数段ステップで実行して結果を得るのか？

AIエージェントでは、LLMの各実行結果を外部で履歴として管理します。
LLMには、ユーザーの入力だけでなく、これまでの実行履歴もあわせて渡します。これによりLLMは、ユーザーの当初の入力と現在までに得られている情報を踏まえて、次に取るべき行動を判断できます。

具体的な処理の流れは以下のとおりです。

1. ユーザーが依頼文を入力する
2. LLMが、その時点で行うべき処理が「ツール呼び出し」か「最終回答」かを判断する
3. ツール呼び出しと判断された場合は、LLMが生成した引数を用いてツールを実行する
4. ツールの実行結果を履歴に追加し、その履歴を含めて再度LLMに入力する
5. 最終回答と判断された場合は、その結果をユーザーに返す

このように、LLMは履歴を参照しながら複数回呼び出され、必要に応じてツールを使いながら最終回答を生成します。

このLLM自身に判断させるループ構造は色々な所で出てくるのですが、はっきりとした呼び名は分かりませんでした。。
agent loop / tool call loopなどの呼び名で出てくることが多いように思います。
参考: https://github.com/humanlayer/12-factor-agents?tab=readme-ov-file#agents-as-loops

実装編

上記を踏まえて実装します。
また実行可能な全体のコードはgithubに配置しています。

AIエージェントに任意コード実行の権限を渡しており、予期しない操作をされる危険性があるのでご注意ください。
上記リポジトリのdevcontainerを利用して仮想環境内で実行することを推奨します。

ステートレスなLLMの関数の定義

import os
import httpx


def call_llm(prompt: str) -> str:
    """
    Stateless LLM call: input string -> output string.
    Uses an OpenAI-compatible endpoint for simplicity.
    """
    api_key = os.getenv("OPEN_AI_API_KEY")
    base_url = "https://api.openai.com"
    model = "gpt-4.1"

    if not api_key:
        raise RuntimeError("Set OPEN_AI_API_KEY in your environment.")

    payload = {
        "model": model,
        "messages": [
            {"role": "user", "content": prompt},
        ],
        "temperature": 0.0,
    }

    headers = {"Authorization": f"Bearer {api_key}"}
    url = base_url.rstrip("/") + "/v1/chat/completions"

    with httpx.Client(timeout=60) as client:
        resp = client.post(url, headers=headers, json=payload)
        resp.raise_for_status()
        data = resp.json()

    return data["choices"][0]["message"]["content"]

• 余分な考慮事項が入ってこないように、LLMのAPI呼び出しを文字入力->文字出力の関数でラップする。
• ツール呼び出しの効果がわかりやすくなることを期待して、古めのChatGPT 4.1モデルを選択。

利用するツールの定義

LLMから呼び出させるツールを2つ定義します。
ツールの実態はPythonの関数です。

• pythonを実行するツール

from io import StringIO
import sys

TOOL_NAME = "python_repl"

class PythonREPL:
    """Simulates a standalone Python REPL with persistent globals."""

    def __init__(self) -> None: 
        # Pythonインタプリタ内でグローバル変数を保持するための辞書を作成
        self._globals: dict = {}

    def run(self, command: str) -> str:
        # 標準出力先を退避
        old_stdout = sys.stdout
        
        # 標準出力をmystdoutに流し込む
        sys.stdout = mystdout = StringIO()
        try:
            exec(command, self._globals)

            # 標準出力先を復元
            sys.stdout = old_stdout

            # exec処理内のprint出力を取得
            output = mystdout.getvalue()
        except Exception as exc:
            sys.stdout = old_stdout
            output = str(exc)
        if output.strip() == "":
            return "[no output] Python実行完了。実行結果を取得した場合は、printを利用してください。"
        
        # exec処理内のprint出力を返す
        return output


def python_repl(command: str) -> str:
    return _PYTHON_REPL.run(command)

_PYTHON_REPL = PythonREPL()

• Web検索を行うツール

import os
import httpx

TOOL_NAME = "tavily_search"

def tavily_search(query: str) -> str:
    api_key = os.getenv("TAVILY_API_KEY")
    if not api_key:
        raise RuntimeError("Set TAVILY_API_KEY in your environment.")

    payload = {
        "api_key": api_key,
        "query": query,
        "search_depth": "basic",
        "max_results": 5,
    }

    # Tavily検索APIを呼び出し
    with httpx.Client(timeout=30) as client:
        resp = client.post("https://api.tavily.com/search", json=payload)
        resp.raise_for_status()
        data = resp.json()

    # 検索結果を整形
    results = data.get("results", [])
    lines = []
    for item in results:
        title = item.get("title", "")
        url = item.get("url", "")
        content = item.get("content", "")
        lines.append(f"- {title}\n  {url}\n  {content}")

    # 結果を文字列として返す
    return "\n".join(lines).strip()

ループ処理の定義/システムプロンプトの定義

import json
import sys

from dotenv import load_dotenv

from .llm import call_llm
from .tools_python import TOOL_NAME as PYTHON_REPL_NAME
from .tools_python import python_repl
from .tools_tavily import TOOL_NAME as TAVILY_NAME
from .tools_tavily import tavily_search
from datetime import datetime

# tool名称と実装関数のマッピング
TOOLS = {
    TAVILY_NAME: tavily_search,
    PYTHON_REPL_NAME: python_repl,
}


# ツール呼び出しを繰り返して目的達成させるためのプロンプト
SYSTEM_PROMPT = """あなたは親切なAIアシスタントです。
必要に応じてtoolを呼び出し、十分な情報が揃ったら最終回答してください。
以下にユーザー入力と前回までのツール実行結果が記載されたscratchpadが与えられます。

使用可能なtool:
- tavily_search: 
  - 概要: web検索を実行します。
  - 利用するシーン: WEBから情報を収集する必要がある場合に利用してください。
  - 引数: 検索キーワードを文字列で指定します。
  - 戻り値: 検索結果の文字列が返されます。
- python_repl: 
  - 概要: pythonコードを実行します。
  - 利用するシーン: 数字の計算を行う場合に必ず利用してください。
  - 引数: pythonスクリプト。実行結果を取得するにはprintで値を出力してください。
  - 戻り値: 実行結果の文字列が返されます。

ルール:
- 新しい情報が必要ならtoolを使う。
- actionは1つのみ。
- 出力は必ずJSON 1つだけで、余計なテキストは出力しない。
  - actionは必ず "tool" または "final" のどちらかです。

今日の日時:
{datetime}

出力JSONスキーマ:
下記のいずれか1つ
- {{"action":"tool","tool":"tavily_search","input":"..."}}
- {{"action":"final","output":"..."}}

User Input: {user_query}
Scratchpad :
{scratchpad}
"""


def run_agent(user_query: str) -> str:
    scratchpad: list[dict] = []

    max_steps = 10
    for step in range(1, max_steps + 1):
        print(f"\n==== Start STEP {step} ====")
        
        prompt = SYSTEM_PROMPT.format(
            datetime=datetime.now(),
            user_query=user_query,
            scratchpad=json.dumps(scratchpad, ensure_ascii=False, indent=2),
        )
        raw = call_llm(prompt)
        action = json.loads(raw)

        if action.get("action") == "tool":
            print(f"Processing tool action.")

            tool_name = action.get("tool")
            tool_input = action.get("input", "")
            tool_func = TOOLS.get(tool_name)
            
            print(f"TOOL CALL: {tool_name}({tool_input})")
            tool_output = tool_func(tool_input)
            print(f"TOOL OUTPUT (preview): {tool_output[:30]}")
            
            scratchpad.append(
                {
                    "action": "tool",
                    "tool": tool_name,
                    "input": tool_input,
                    "observation": tool_output,
                }
            )
            continue

        if action.get("action") == "final":
            print(f"Processing final action.")
            return str(action.get("output", "")).strip()

        # print(f"[step {step}] invalid action: {action}")

        scratchpad.append({"error": f"Unknown action: {action.get('action')}"})

    return "Max steps reached without a final answer."


def main() -> int:
    load_dotenv()
    if len(sys.argv) < 2:
        print("Usage: python -m src.agent \"your question\"")
        return 1

    user_query = " ".join(sys.argv[1:])
    answer = run_agent(user_query)
    print(answer)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

• ツール説明や出力フォーマットを含めたプロンプトの定義、及びLLMの出力を用いたループ処理の定義を行っています。
• LLMには実行ごとに次に行う行動を判定させます。判定の結果はaction属性のtoolorfinalとして出力されます。
  • actionがtoolだった場合: 該当のツールを実行し、呼び出した引数や実行結果をscratchpadに追加する。ループを継続
  • actionがfinalだった場合: LLMの出力を最終結果として出力

※ プロンプトは英語で記載するのが一般的ですが、今回は分かりやすさのために日本語としています。

実行結果

作成した簡易AIエージェントで色々なプロンプトを試してみます。
STEP1, STEP2,....の順で途中経過が表示され、最後のSTEPで最終結果が出力されます。

1. web検索

(プロンプト: "2つのニュースサイトを訪問して、それぞれのトップニュースを取得して。またその結果を200文字のレポートにまとめて)

• ニュースを調べて、レポート作成させてみます。
• 3ステップ(検索2回+最終結果作成1回)で処理を完了して、最終的な結果の文章が出力されます。

$ uv run python -m src.agent "2つのニュースサイトを訪問して、それぞれのトップニュースを取得して。またその結果を200文字のレポートにまとめて"

==== Start STEP 1 ====
Processing tool action.
TOOL CALL: tavily_search(主要ニュースサイトのトップニュース 2025年12月24日)
TOOL OUTPUT (preview): - 2025年12月24日の記事一覧
  https://w

==== Start STEP 2 ====
Processing tool action.
TOOL CALL: tavily_search(読売新聞 トップニュース 2025年12月24日)
TOOL OUTPUT (preview): - 読売新聞 - 12月 24, 2025
  https:

==== Start STEP 3 ====
Processing final action.
2025年12月24日の主要ニュースサイト（朝日新聞・読売新聞）のトップニュースは以下の通りです。

朝日新聞では、トルコでのジェット機墜落事故やウクライナ情勢、国内では出生数の減少や高額療養費の見直し、外国人労働者政策などが大きく報じられています。

読売新聞では、国内金価格の史上最高値更新、電通過労死事件から10年の母親の訴え、高市内閣の高支持率、能登半島地震の孤立集落問題などが注目されています。

両紙とも国際情勢と国内の社会・経済問題を幅広く取り上げており、年末らしい総括や回顧記事も目立ちました。

2. フィボナッチ数の計算

プロンプト: フィボナッチ数の30番目の数字を計算して

• pythonを呼び出して処理計算しています。
• フィボナッチ数計算用の関数を定義後に実行して、結果を取得しています。

$ uv run python -m src.agent "フィボナッチ数の30番目の数字を計算して"

==== Start STEP 1 ====
Processing tool action.
TOOL CALL: python_repl(def fibonacci(n):
    a, b = 0, 1
    for _ in range(n):
        a, b = b, a + b
    return a
print(fibonacci(30)))
TOOL OUTPUT (preview): 832040


==== Start STEP 2 ====
Processing final action.
フィボナッチ数列の30番目の数字は832040です。

3. 確率計算

プロンプト: サイコロを 2 個振って合計 が 12 以上になる確率を厳密計算とモンテカルロ法の両方で求めて

• サイコロを2回降って合計値が12以上になる確率を2種類の方法で計算させてみます。
• それぞれの手法をPythonで実装して確認しています。

uv run python -m src.agent "サイコロを 2 個振って合計 が 12 以上になる確率を厳密計算とモンテカルロ法の両方で求めて"

==== Start STEP 1 ====
Processing tool action.
TOOL CALL: python_repl(# サイコロ2個の合計が12以上になる確率（厳密計算）
# サイコロは6面なので、2個振ると出る目の合計は2～12
# それぞれの目の出る組み合わせを数える

total = 0
success = 0
for i in range(1, 7):
    for j in range(1, 7):
        total += 1
        if i + j >= 12:
            success += 1
exact_prob = success / total
print(exact_prob)
)
TOOL OUTPUT (preview): 0.027777777777777776


==== Start STEP 2 ====
Processing tool action.
TOOL CALL: python_repl(import random
# モンテカルロ法でサイコロ2個の合計が12以上になる確率を推定
trials = 1000000
success = 0
for _ in range(trials):
    dice1 = random.randint(1, 6)
    dice2 = random.randint(1, 6)
    if dice1 + dice2 >= 12:
        success += 1
monte_carlo_prob = success / trials
print(monte_carlo_prob)
)
TOOL OUTPUT (preview): 0.027617


==== Start STEP 3 ====
Processing final action.
サイコロを2個振って合計が12以上になる確率は、

・厳密計算では約0.0278（=2.78%）
・モンテカルロ法による推定では約0.0276（=2.76%）

となります。どちらもほぼ同じ値で、サイコロ2個で合計12以上（つまり12のみ）は非常にまれなことが分かります。

4. iris(アヤメ)のデータセットをDLしてヒストグラム画像を作成

データ分析でお約束の、irisデータセットを使ってデータの可視化にチャレンジしてみます。

プロンプト: アヤメのデータセットを取得して、各特徴量をヒストグラムとして画像化して

• irisデータセットをDLする方法をweb検索し、見つかったURLでcsvファイルとしてDLした後、matplotlibで可視化した画像を生成しているようです。
• 興味深い部分として、途中可視化にseabornライブラリを利用しようとしたものの、ライブラリ未インストールのエラーが出たため、可視化方法をmatplotlibに自動で切り替えています。

$ uv add matplotlib pandas # 可視化に必要なライブラリをあらかじめインストールする(`matplotlib`, `pandas`)。
$ uv run python -m src.agent "アヤメのデータセットを取得して、各特徴量をヒストグラムとして画像化して"

==== Start STEP 1 ====
Processing tool action.
TOOL CALL: tavily_search(アヤメ データセット ダウンロード)
TOOL OUTPUT (preview): - Irisデータセットのダウンロード | Excel VB

==== Start STEP 2 ====
Processing tool action.
TOOL CALL: python_repl(import matplotlib.pyplot as plt
from sklearn.datasets import load_iris
import pandas as pd

# アヤメデータセットの取得
data = load_iris()
df = pd.DataFrame(data.data, columns=data.feature_names)

# 各特徴量のヒストグラムを描画
fig, axes = plt.subplots(2, 2, figsize=(10, 8))
axes = axes.flatten()
for i, col in enumerate(df.columns):
    axes[i].hist(df[col], bins=20, color='skyblue', edgecolor='black')
    axes[i].set_title(col)
plt.tight_layout()
plt.savefig('iris_hist.png')
print('iris_hist.png'))
TOOL OUTPUT (preview): No module named 'sklearn'

==== Start STEP 3 ====
Processing tool action.
TOOL CALL: python_repl(import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
import urllib.request

# Irisデータセットを直接ダウンロード
url = 'https://raw.githubusercontent.com/pandas-dev/pandas/main/pandas/tests/io/data/csv/iris.csv'
filename = 'iris.csv'
urllib.request.urlretrieve(url, filename)

# データ読み込み
cols = ['sepal_length', 'sepal_width', 'petal_length', 'petal_width', 'species']
df = pd.read_csv(filename, names=cols, header=0)

# 各特徴量のヒストグラムを描画
fig, axes = plt.subplots(2, 2, figsize=(10, 8))
axes = axes.flatten()
for i, col in enumerate(cols[:-1]):
    sns.histplot(df[col], bins=20, ax=axes[i], color='skyblue', edgecolor='black')
    axes[i].set_title(col)
plt.tight_layout()
plt.savefig('iris_hist.png')
print('iris_hist.png'))
TOOL OUTPUT (preview): No module named 'seaborn'

==== Start STEP 4 ====
Processing tool action.
TOOL CALL: python_repl(import matplotlib.pyplot as plt
import pandas as pd
import urllib.request

# Irisデータセットを直接ダウンロード
url = 'https://raw.githubusercontent.com/pandas-dev/pandas/main/pandas/tests/io/data/csv/iris.csv'
filename = 'iris.csv'
urllib.request.urlretrieve(url, filename)

# データ読み込み
cols = ['sepal_length', 'sepal_width', 'petal_length', 'petal_width', 'species']
df = pd.read_csv(filename, names=cols, header=0)

# 各特徴量のヒストグラムを描画
fig, axes = plt.subplots(2, 2, figsize=(10, 8))
axes = axes.flatten()
for i, col in enumerate(cols[:-1]):
    axes[i].hist(df[col], bins=20, color='skyblue', edgecolor='black')
    axes[i].set_title(col)
plt.tight_layout()
plt.savefig('iris_hist.png')
print('iris_hist.png'))
TOOL OUTPUT (preview): iris_hist.png


==== Start STEP 5 ====
Processing final action.
アヤメ（Iris）データセットの各特徴量（sepal_length, sepal_width, petal_length, petal_width）のヒストグラム画像を作成しました。

画像ファイル名：iris_hist.png

この画像には、4つの特徴量それぞれの値の分布がヒストグラムとして表示されています。ご希望の可視化が完了しました。

まとめ

AIエージェントの中身が理解できるよう、フレームワークを使わずに、シンプルなAIエージェントを実装してみました。
誰かの理解の助けになれば幸いです。