【実践ガイド】Agent Lightningで自作エージェントを強化学習する ─ ユースケース別導入手順

はじめに

前回の記事o3時代でもエージェントRLは必要か？では、ベースモデルの性能向上とエージェントRLが補完関係にあることを説明しました。

本記事では、具体的にどういう時に、何を、どうすればいいのかを、Agent Lightningを使った実践ガイドとして解説します。

---

対象者

この記事は下記のような人を対象にしています。

• LLMエージェントの開発経験がある方
• プロンプトエンジニアリングで性能改善に限界を感じている方
• Agent Lightningを使った実践的な導入手順を知りたい方
• APOとGRPO/PPOの違いを理解したい方

---

目次

• 前提知識：GPTモデルとAgent Lightningの関係
• Agent Lightningの全体像
• セットアップ手順
• ユースケース別実装例
• アルゴリズムの選び方
• 判断フローチャート：何を使うべきか
• よくあるつまずきポイント
• おわりに

---

前提知識：GPTモデルとAgent Lightningの関係

Agent Lightningは「GPTの使い方」を最適化します

Agent Lightningが最適化しているのは、GPTモデルそのものではなく、GPTモデルの使い方です。

普段の開発では、プロンプトを手動で試行錯誤していますよね。「うまくいかない → プロンプトを書き換える → また試す」の繰り返しです。Agent Lightning（特にAPO）は、この試行錯誤を自動化するツールになります。

2つのモードの違い

APOがやっていること:

1. 現在のプロンプトでタスクを実行します
2. 失敗したケースを分析します（これもGPT-4oに聞きます）
3. 改善されたプロンプトを生成します（これもGPT-4oに聞きます）
4. 新しいプロンプトで再度実行 → 繰り返します

GRPO/PPOがやっていること:

モデル内部のパラメータを直接更新して、振る舞いを改善します。これが本当の「強化学習」ですね。

なぜGPT-4oでは「モデル重み更新」ができないのか

GPT-4oなどOpenAIのモデルはクローズドで、内部のパラメータにはアクセスできません。API経由で「入力を送って出力を受け取る」ことしかできないのです。

一方、QwenやLlamaなどのオープンソースモデルは、重みが公開されているので、自分でダウンロードして、パラメータを更新できます。

最適化手法	何が変わる？	使えるモデル	GPU
APO	プロンプトのテキスト	GPT-4o可	不要
GRPO/PPO	モデルの重み	オープンソースのみ	必要

レイヤー構造

---

Agent Lightningの全体像

アーキテクチャ：Training-Agent Disaggregation

Agent Lightningの核心は「実行と訓練の分離」です。

重要な概念

概念	説明
Rollout	エージェントがタスクを1回実行すること
Span	1回のLLM呼び出し、ツール呼び出し、または報酬イベント
Transition	(状態, 行動, 報酬) の組。RLの訓練単位
Credit Assignment	複数ステップのうち、どの行動が成果に貢献したかを分析

---

セットアップ手順

前提条件

• Python 3.10以上
• OpenAI APIキー（または互換API）
• GPU（モデル重み更新を行う場合）

インストール

# 基本インストール
pip install --upgrade --index-url https://test.pypi.org/simple/ \
    --extra-index-url https://pypi.org/simple/ --pre agentlightning

# APO（プロンプト最適化）を使う場合
pip install agentlightning[apo]

# VERL（RL訓練）を使う場合
pip install agentlightning[verl]

最小構成の確認

# test_setup.py
import agentlightning as agl
from agentlightning import rollout, PromptTemplate

@rollout
def simple_agent(task: dict, prompt_template: PromptTemplate) -> float:
    """最小限のエージェント定義"""
    # タスクを受け取り、報酬（0.0〜1.0）を返します
    prompt = prompt_template.format(**task)
    # ... LLM呼び出し ...
    reward = 0.5  # 仮の報酬
    return reward

print("セットアップ完了！")

---

ユースケース別実装例

ユースケースA：会議室予約エージェント（APO）

難易度: ★☆☆（初心者向け） 特徴: モデル重み更新なし、プロンプトのみ最適化

これが最も簡単に始められるパターンですね。GPU不要で、プロンプトを自動改善します。

# room_selector_apo.py
from typing import TypedDict
import openai
from agentlightning import rollout, PromptTemplate, Trainer
from agentlightning.algorithms import APO

# 1. タスクの型定義
class RoomSelectionTask(TypedDict):
    duration: int
    attendees: int
    equipment: list[str]
    accessibility: bool
    expected_room: str  # 正解ラベル

# 2. エージェント定義
@rollout
def room_selector(task: RoomSelectionTask, prompt_template: PromptTemplate) -> float:
    """会議室を選択するエージェント"""

    # プロンプトテンプレートを使って入力を構築します
    prompt = prompt_template.format(
        duration=task["duration"],
        attendees=task["attendees"],
        equipment=task["equipment"],
        accessibility=task["accessibility"]
    )

    # LLM呼び出し
    client = openai.OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )

    selected_room = response.choices[0].message.content.strip()

    # 報酬計算（正解なら1.0、不正解なら0.0）
    reward = 1.0 if selected_room == task["expected_room"] else 0.0
    return reward

# 3. 訓練データ
tasks = [
    {"duration": 30, "attendees": 4, "equipment": ["whiteboard"],
     "accessibility": False, "expected_room": "Room A"},
    {"duration": 60, "attendees": 10, "equipment": ["projector", "video"],
     "accessibility": True, "expected_room": "Room B"},
    # ... 他のタスク
]

# 4. 初期プロンプトテンプレート
initial_prompt = PromptTemplate("""
You are a meeting room booking assistant.
Select the best room for the following requirements:
- Duration: {duration} minutes
- Attendees: {attendees} people
- Equipment needed: {equipment}
- Accessibility required: {accessibility}

Available rooms: Room A (small, whiteboard), Room B (large, projector, accessible)
Reply with only the room name.
""")

# 5. APOアルゴリズムで訓練
algorithm = APO(
    initial_prompt=initial_prompt,
    optimizer_model="gpt-4o",      # プロンプトを改善するモデル
    critic_model="gpt-4o-mini",    # 批評を生成するモデル
    beam_width=3,                  # 並列で試すプロンプト数
    max_rounds=5                   # 最適化ラウンド数
)

trainer = Trainer(
    agent=room_selector,
    algorithm=algorithm,
    tasks=tasks,
    num_workers=4
)

# 訓練実行
trainer.fit()

# 最適化されたプロンプトを取得
best_prompt = trainer.get_best_resource("prompt_template")
print(f"最適化されたプロンプト:\n{best_prompt}")

APOが内部で何をしているか

上記コードで trainer.fit() を呼ぶと、以下のループが自動で回ります。

公式ドキュメントの実験結果

• 検証データ: 29件
• ベースライン精度: 56.9%
• 2ラウンド後: 72.1%（+15.2ポイント改善）
• 所要時間: 約10分（8ワーカー並列）

重要: この例ではGPT-4o自体は何も変わっていません。変わったのはプロンプトのテキストだけです。APOは「プロンプトの試行錯誤を自動化するツール」であり、モデルを変更する「強化学習」とは異なります。

---

ユースケースB：Text-to-SQLエージェント（LangGraph + RL）

難易度: ★★★（中〜上級者向け） 特徴: モデル重み更新あり、GPU必要

複数エージェントが協調するワークフローで、特定のエージェントのみをRL最適化します。

APOとの違い: こちらは本当の強化学習です。Qwenなどオープンソースモデルの重みを直接更新します。GPT-4oは使えません（重みにアクセスできないからです）。

# sql_agent_rl.py
from langgraph.graph import StateGraph
from agentlightning import LitAgent, Trainer
from agentlightning.algorithms import VERL

# 1. LangGraphでSQLエージェントを定義
def build_sql_agent(db_path: str, model: str):
    """
    ワークフロー:
    write_query → execute → check_query → (失敗なら) rewrite_query → execute → ...
    """

    class SQLState(TypedDict):
        question: str
        query: str
        result: str
        error: str
        iteration: int

    def write_query(state: SQLState) -> SQLState:
        """自然言語からSQLを生成します"""
        # LLM呼び出し
        ...
        return {"query": generated_sql}

    def execute_query(state: SQLState) -> SQLState:
        """SQLを実行します"""
        try:
            result = run_sql(state["query"], db_path)
            return {"result": result, "error": ""}
        except Exception as e:
            return {"result": "", "error": str(e)}

    def check_query(state: SQLState) -> str:
        """結果を検証し、次のステップを決定します"""
        if state["error"] or state["iteration"] >= 3:
            return "end"
        return "rewrite"

    def rewrite_query(state: SQLState) -> SQLState:
        """エラーを修正したSQLを再生成します"""
        # LLM呼び出し（エラーメッセージを含めます）
        ...
        return {"query": rewritten_sql, "iteration": state["iteration"] + 1}

    # グラフ構築
    graph = StateGraph(SQLState)
    graph.add_node("write", write_query)
    graph.add_node("execute", execute_query)
    graph.add_node("check", check_query)
    graph.add_node("rewrite", rewrite_query)

    graph.set_entry_point("write")
    graph.add_edge("write", "execute")
    graph.add_conditional_edges("execute", check_query, {
        "rewrite": "rewrite",
        "end": "__end__"
    })
    graph.add_edge("rewrite", "execute")

    return graph.compile()

# 2. LitAgentでラップ
class SQLAgent(LitAgent):
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.graph = None

    def training_rollout(self, task, rollout_id, resources):
        """1回の訓練実行"""
        # リソースからLLMエンドポイントを取得します（訓練中は更新されます）
        llm = resources["llm"]

        # グラフを構築します（LLMエンドポイントを注入します）
        self.graph = build_sql_agent(
            db_path=self.db_path,
            model=llm.model
        )

        # 実行
        result = self.graph.invoke({"question": task["question"]})

        # 報酬計算（正解SQLと実行結果を比較します）
        reward = evaluate_sql(
            predicted=result["query"],
            ground_truth=task["ground_truth_query"],
            db_path=self.db_path
        )

        return reward

def evaluate_sql(predicted, ground_truth, db_path):
    """SQLの実行結果が一致すれば1.0、そうでなければ0.0を返します"""
    result_pred = run_sql(predicted, db_path)
    result_true = run_sql(ground_truth, db_path)
    return 1.0 if result_pred == result_true else 0.0

# 3. VERL設定
verl_config = {
    "algorithm": {
        "adv_estimator": "grpo",      # GRPOアルゴリズム
        "use_kl_in_reward": False
    },
    "data": {
        "train_batch_size": 32,
        "max_prompt_length": 4096,
        "max_response_length": 2048,
    },
    "actor_rollout_ref": {
        "rollout": {
            "name": "vllm",            # vLLMで推論
            "n": 4,                    # GRPOのグループサイズ
            "multi_turn": {"format": "hermes"}
        },
        "actor": {
            "ppo_mini_batch_size": 32,
            "optim": {"lr": 1e-6}
        },
        "model": {
            "path": "Qwen/Qwen2.5-Coder-1.5B-Instruct"  # ベースモデル
        }
    },
    "trainer": {
        "n_gpus_per_node": 1,
        "total_epochs": 2,
        "save_freq": 64
    }
}

# 4. 訓練
algorithm = VERL(
    config=verl_config,
    agent_match="write|rewrite"  # write_queryとrewrite_queryのみ最適化
)

trainer = Trainer(
    agent=SQLAgent(db_path="spider.db"),
    algorithm=algorithm,
    train_tasks=spider_train_tasks,
    val_tasks=spider_val_tasks,
)

# デバッグモード（10タスクだけ試します）
trainer.dev()

# 本番訓練
trainer.fit()

ポイント:

• agent_match="write|rewrite" で、check_queryは最適化対象外になります
• 報酬は「SQLの実行結果が正解と一致するか」という検証可能な指標です
• ground_truth_queryは訓練中にモデルには見せません

APOとの違い（再確認）:

APOの場合:
  プロンプト: 変わる
  GPT-4o: 変わらない ← OpenAIのサーバーにある同じモデル

GRPO/PPOの場合（この例）:
  プロンプト: 変わらない（または変わる）
  Qwen: 変わる ← 自分のGPUで動かすモデルの重みが更新される

---

ユースケースC：RAGエージェント（OpenAI Agents SDK）

難易度: ★★☆（中級者向け） 特徴: 検索クエリ生成と回答生成を同時最適化

# rag_agent_rl.py
from openai import OpenAI
from agentlightning import rollout, emit_reward

# 検索バックエンド（BGE + Faiss）
from retriever import WikipediaRetriever

retriever = WikipediaRetriever(index_path="wikipedia_bge.faiss")

@rollout
def rag_agent(task: dict, llm) -> float:
    """
    マルチホップQAエージェント
    1. 検索クエリを生成します
    2. 検索実行します
    3. 必要なら追加検索します
    4. 回答生成します
    """
    client = OpenAI(base_url=llm.endpoint, api_key=llm.api_key or "dummy")

    question = task["question"]
    context = []

    for hop in range(3):  # 最大3ホップ
        # 検索クエリ生成
        query_response = client.chat.completions.create(
            model=llm.model,
            messages=[
                {"role": "system", "content": "Generate a search query to answer the question."},
                {"role": "user", "content": f"Question: {question}\nContext so far: {context}"}
            ]
        )
        search_query = query_response.choices[0].message.content

        # 検索実行
        results = retriever.search(search_query, top_k=3)
        context.extend(results)

        # 中間報酬を発行します（検索結果の関連度）
        relevance = compute_relevance(results, question)
        emit_reward(relevance * 0.1, name=f"search_hop_{hop}")

        # 回答可能か判断します
        can_answer_response = client.chat.completions.create(
            model=llm.model,
            messages=[
                {"role": "user", "content": f"Can you answer '{question}' with this context? {context}"}
            ]
        )
        if "yes" in can_answer_response.choices[0].message.content.lower():
            break

    # 最終回答生成
    answer_response = client.chat.completions.create(
        model=llm.model,
        messages=[
            {"role": "system", "content": "Answer the question based on the context."},
            {"role": "user", "content": f"Question: {question}\nContext: {context}"}
        ]
    )
    answer = answer_response.choices[0].message.content

    # 最終報酬（正解との一致度）
    final_reward = compute_answer_similarity(answer, task["expected_answer"])
    return final_reward

def compute_relevance(results, question):
    """検索結果と質問の関連度を計算します"""
    # BGE embeddingsでコサイン類似度を計算
    ...
    return relevance_score

def compute_answer_similarity(predicted, expected):
    """回答の類似度を計算します（F1スコアなど）"""
    ...
    return f1_score

ポイント:

• emit_reward() で中間報酬を発行し、スパース報酬問題を軽減します
• 検索クエリ生成と回答生成の両方が最適化されます

---

アルゴリズムの選び方

比較表

アルゴリズム	モデル更新	GPU必要	最適化対象	向いているケース
APO	なし	不要	プロンプト	素早く試したい、小規模データ
GRPO	あり	必要	モデル重み	検証可能な報酬がある、中規模データ
PPO	あり	必要	モデル重み	複雑な報酬設計、大規模データ
SFT	あり	必要	モデル重み	良い例が大量にある

選択フローチャート

---

判断フローチャート：何を使うべきか

大前提：本当にエージェントRLが必要か？

エージェントRL導入の判断基準

やるべき場合:

• 複数ステップのツール呼び出しがある
• 報酬が自動計算可能（テスト成否、SQL実行結果、検索適合率など）
• 同じタイプのタスクが大量にある（100件以上推奨）
• プロンプトエンジニアリングで頭打ちになった

やらないほうがいい場合:

• 単発の質問応答タスク → フロンティアモデル使用で十分です
• 報酬設計が困難（創作、オープンエンド対話）
• タスク数が少ない（10件以下）
• まだプロンプトを十分に試していない

---

よくあるつまずきポイント

問題1：報酬が常に0になる

原因: 報酬計算ロジックのバグ、またはタスクが難しすぎます

解決策:

# デバッグモードで確認します
trainer.dev()  # 10タスクだけ実行してSpanを表示

# 報酬の分布を確認します
for span in trainer.get_spans():
    print(f"Task: {span.task_id}, Reward: {span.reward}")

問題2：訓練しても改善しない

原因:

• データ量が少なすぎます
• 報酬のバリエーションが少ないです（全部0か全部1）

解決策:

# 中間報酬を追加してスパース報酬問題を軽減します
from agentlightning import emit_reward

@rollout
def my_agent(task, llm):
    # ステップ1完了
    emit_reward(0.2, name="step1_done")

    # ステップ2完了
    emit_reward(0.3, name="step2_done")

    # 最終報酬
    return final_reward

問題3：マルチエージェントのどれを最適化すべきかわからない

解決策: agent_match で選択的に最適化します

# 例：write_queryとrewrite_queryだけ最適化、check_queryは固定
algorithm = VERL(
    config=verl_config,
    agent_match="write|rewrite"  # 正規表現でマッチ
)

問題4：GPUメモリが足りない

解決策:

# バッチサイズを小さくします
verl_config["data"]["train_batch_size"] = 8

# 小さいモデルを使います
verl_config["actor_rollout_ref"]["model"]["path"] = "Qwen/Qwen2.5-Coder-1.5B-Instruct"

# 勾配累積を使います
verl_config["actor_rollout_ref"]["actor"]["gradient_accumulation"] = 4

---

おわりに

基礎知識のおさらい

用語	意味
APO	プロンプトを自動改善するツールです。モデル自体は変わりません。GPT-4o使用可能です。
GRPO/PPO	モデルの重みを更新する強化学習です。オープンソースモデルのみ対応で、GPU必要です。
Agent Lightning	上記両方をサポートするフレームワークです

クイックリファレンス

やりたいこと	使うもの	所要時間目安
プロンプトを自動改善	APO	10分〜1時間
Text-to-SQLの精度向上	GRPO + LangGraph	数時間〜1日
RAGの検索クエリ改善	GRPO + 中間報酬	数時間〜1日
コーディングエージェント改善	PPO + テスト報酬	1日〜数日

始め方のステップ

1. まずAPOから始めましょう

   • GPU不要、数分で結果が見えます
   • プロンプトがどう改善されるか学べます

2. 効果が確認できたらGRPOへ

   • 検証可能な報酬を設計します
   • 小さいモデル（1.5B〜3B）で試します

3. 本番導入

   • ベースラインとの比較を必ず行います
   • A/Bテストで実際の効果を測定します

本記事では、Agent Lightningを使った自作エージェントの強化学習について、ユースケース別の導入手順をまとめました。この記事がどなたかの参考になれば幸いです。

Agent Lightningは、既存のエージェントコードをほぼ修正せずに強化学習を導入できる画期的なフレームワークです。まずは手元のエージェントにAPOを試してみることから始めてみてください。プロンプトがどう改善されるかを見るだけでも、エージェントRLの可能性を実感できるはずです。

---

参考

• Agent Lightning 公式ドキュメント
• GitHub リポジトリ
• Agent Lightning Paper (arXiv)
• Train the First Agent - 公式チュートリアル
• examples/apo - APOのサンプル
• examples/spider - Text-to-SQLのサンプル