実践：最初のAgent Loopアプリケーションを構築

理論から実践への最後の一歩

これまでの4つのチュートリアルでは、Agent Loopの基礎理論、ツール呼び出し、Reflection、マルチエージェント協調について解説しました。今回は、これらの知識を統合する時です。要件理解から最終納品までの全プロセスを自動で完了できる、完全なAgent Loopアプリケーションを構築します。

このアプリケーションは「AIリサーチアシスタント」をシミュレートします。ユーザーが研究テーマを提案すると、エージェントが要件分析、タスク分解、情報検索、結果検証、レポート生成を自動的に実行します。プロセス全体に人手は不要で、実際のシナリオにおけるAgent Loopの完全な能力を示します。

アプリケーションアーキテクチャ設計

「AIリサーチアシスタント」は5段階のパイプラインアーキテクチャを採用しています：

1. 要件受付：ユーザーの研究テーマを理解し、範囲と深さの要件を明確化
2. タスク分解：研究テーマを複数の実行可能なサブタスクに分解
3. 段階的実行：サブタスクを一つずつ完了し、情報と分析結果を収集
4. 結果検証：情報の正確性を相互検証し、欠落内容を補完
5. 最終納品：すべての結果を統合し、構造化された研究レポートを生成

各段階でフィードバックループが発生する可能性があります。検証が不合格の場合は実行段階に戻って情報を再収集し、納品が不完全な場合は分解段階に戻って再計画します。

スキルチェーン分解：完全なAgent Loopアプリケーション

「完全なAgent Loopアプリケーション」スキルチェーンを例に、エンドツーエンドプロセスにおける各ノードの役割を分解します：

ノード1：要件受付（sp-15）—これはAgent Loop全体の入り口です。AIはユーザーの自然言語入力から重要な情報を抽出する必要があります：研究テーマ、期待される深さ、出力形式、特別な要件など。このノードの出力は構造化された要件ドキュメントです。

ノード2：タスク分解（sp-188）—要件ドキュメントを実行可能な行動計画に変換します。優れたタスク分解には次の条件があります：各サブタスクが独立して実行可能、サブタスク間に明確な論理的関係がある、総作業量が合理的で管理可能。

ノード3：段階的実行（sp-187）—これはAgent Loopの中核となる実行段階です。AIはタスク計画に従い、サブタスクを一つずつ完了します。各サブタスクにはツール呼び出し、情報検索、コード実行などの操作が含まれる場合があります。実行ログは記録され、後の検証に使用されます。

ノード4：結果検証（sp-185）—すべての実行結果に対して品質チェックを実施します。検証の次元には以下が含まれます：情報の正確性（事実誤認がないか）、網羅性（重要なポイントの見落としがないか）、論理的一貫性（各部分が矛盾していないか）。検証が不合格の場合、具体的な問題が実行ノードにフィードバックされ修正されます。

ノード5：最終納品（sp-122）—検証済みのすべての結果を最終成果物に統合します。ユーザーの要求形式（レポート、PPT、コードリポジトリなど）に応じて、要件を満たす出力を生成します。このノードではメタデータ（生成時刻、情報ソース、信頼度スコアなど）も追加されます。

完全なコード実装

以下は「AIリサーチアシスタント」の完全な実装コードです：

import json
import time
from datetime import datetime
from openai import OpenAI

client = OpenAI(
    api_key="your-deepseek-api-key",
    base_url="https://api.deepseek.com"
)

class AIResearchAssistant:
    """完全なAgent Loopアプリケーション：AIリサーチアシスタント"""

    def __init__(self, max_iterations=10):
        self.max_iterations = max_iterations
        self.execution_log = []
        self.start_time = None

    def log(self, phase, message):
        """実行ログを記録"""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "phase": phase,
            "message": message
        }
        self.execution_log.append(entry)
        print(f"[{phase}] {message}")

    def call_llm(self, system_prompt, user_prompt, response_format=None):
        """統一LLM呼び出しインターフェース"""
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ]
        kwargs = {
            "model": "deepseek-chat",
            "messages": messages
        }
        if response_format:
            kwargs["response_format"] = response_format

        response = client.chat.completions.create(**kwargs)
        return response.choices[0].message.content

    def phase1_understand(self, user_input):
        """フェーズ1：要件受付と分析"""
        self.log("要件分析", "ユーザー要件の分析を開始")

        prompt = f"""以下のユーザーの研究要件を分析し、重要な情報を抽出してください。

ユーザー入力：{user_input}

JSON形式で出力してください：
{{
    "topic": "研究テーマ",
    "scope": "研究範囲（技術/ビジネス/学術など）",
    "depth": "期待される深さ（概要/詳細/深掘り）",
    "format": "期待される出力形式",
    "keywords": ["キーワード1", "キーワード2"],
    "constraints": ["制約条件1", "制約条件2"],
    "target_audience": "対象読者"
}}"""

        result = self.call_llm(
            "あなたは要件分析の専門家であり、曖昧な記述から構造化された要件を抽出するのが得意です。",
            prompt
        )
        requirements = json.loads(result)
        self.log("要件分析", f"テーマを特定：{requirements['topic']}")
        re
turn requirements

    def phase2_decompose(self, requirements):
        """フェーズ2: タスク分解"""
        self.log("タスク分解", "研究タスクの分解を開始")

        prompt = f"""以下の要件に基づいて、研究タスクを3〜5つの実行可能なサブタスクに分解してください。

要件：{json.dumps(requirements, ensure_ascii=False)}

JSON形式で出力してください：
{{
    "subtasks": [
        {{
            "id": 1,
            "title": "サブタスクのタイトル",
            "description": "詳細な説明",
            "type": "research|analysis|synthesis|verification",
            "estimated_tokens": 500,
            "depends_on": []
        }}
    ],
    "execution_order": [1, 2, 3]
}}"""

        result = self.call_llm(
            "あなたは複雑なタスクを実行可能なサブタスクに分解するのが得意なプロジェクト管理の専門家です。",
            prompt
        )
        plan = json.loads(result)
        self.log("タスク分解", f"{len(plan['subtasks'])} 個のサブタスクに分解しました")
        return plan

    def phase3_execute(self, requirements, plan):
        """フェーズ3: 段階的実行"""
        self.log("実行フェーズ", "サブタスクの実行を開始")
        results = {}

        for task_id in plan["execution_order"]:
            task = next(t for t in plan["subtasks"] if t["id"] == task_id)
            self.log("実行フェーズ", f"サブタスク {task_id} を実行中: {task['title']}")

            # 依存タスクのコンテキストを収集
            context = ""
            for dep_id in task.get("depends_on", []):
                if dep_id in results:
                    context += f"\n前提タスク {dep_id} の結果：\n{results[dep_id]}"

            prompt = f"""以下の研究サブタスクを完了してください。

研究テーマ：{requirements['topic']}
サブタスク：{task['title']}
説明：{task['description']}
{"コンテキスト：" + context if context else ""}

要件：
1. 正確で深みのある情報を提供する
2. 具体的なデータと事例を引用する
3. 情報の不確実性を指摘する（もしあれば）
4. 情報のソースタイプを明記する（学術論文/業界レポート/ニュース/公式データなど）"""

            result = self.call_llm(
                "あなたは情報の収集・分析・統合に長けたシニアリサーチャーです。",
                prompt
            )
            results[task_id] = result
            self.log("実行フェーズ", f"サブタスク {task_id} 完了（{len(result)} 文字）")

        return results

    def phase4_verify(self, requirements, results):
        """フェーズ4: 結果の検証"""
        self.log("検証フェーズ", "結果の品質検証を開始")

        prompt = f"""以下の研究結果の完全性と正確性を検証してください。

研究テーマ：{requirements['topic']}
研究結果：
{json.dumps(results, ensure_ascii=False, indent=2)}

以下の観点から評価してください：
1. 情報の正確性（疑わしい点や矛盾があるか）
2. カバレッジの完全性（重要な側面が欠落していないか）
3. 論理的一貫性（各部分間の矛盾がないか）

JSON形式で出力してください：
{{
    "verified": true/false,
    "overall_score": 0-100,
    "issues": [
        {{"task_id": 1, "severity": "high/medium/low", "description": "問題の説明"}}
    ],
    "missing_topics": ["欠落したトピック"],
    "recommendation": "合格/補足が必要/やり直しが必要"
}}"""

        result = self.call_llm(
            "あなたは情報の欠陥を見逃さない厳格な品質監査の専門家です。",
            prompt
        )
        verification = json.loads(result)

        if verification["verified"]:
            self.log("検証フェーズ", f"検証合格、スコア：{verification['overall_score']}")
        else:
            self.log("検証フェーズ", f"{len(verification['issues'])} 件の問題を発見")

        return verification

    def phase5_deliver(self, requirements, results, verification):
        """フェーズ5: 最終納品"""
        self.log("納品フェーズ", "最終レポートを生成中")

        elapsed = time.time() - self.start_time

        prompt = f"""以下の研究結果に基づいて、最終的な研究レポートを生成してください。

研究テーマ：{requirements['topic']}
対象読者：{requirements['target_audience']}
期待される形式：{requirements['format']}
検証結果：{json.dumps(verification, ensure_ascii=False)}

研究データ：
{json.dumps(results, ensure_ascii=False, indent=2)}

レポート要件：
1. 明確な構成（要旨、本文、結論を含む）
2. 専門的でありながら理解しやすく、対象読者に適している
3. 

重要なデータと引用元を明記する
4. 最後に実行統計（所要時間、サブタスク数、品質スコア）を添付する

実行統計：
- 総所要時間：{elapsed:.1f} 秒
- サブタスク数：{len(results)}
- 品質スコア：{verification['overall_score']}"""

        report = self.call_llm(
            "あなたは、研究データを高品質なレポートに変換することに長けたシニアレポート作成専門家です。",
            prompt
        )
        self.log("納品フェーズ", f"レポート生成完了（{len(report)} 文字）")
        return report

    def run(self, user_input):
        """完全なAgent Loopを実行する"""
        self.start_time = time.time()
        print("=" * 60)
        print("AI研究アシスタント起動")
        print("=" * 60)

        for iteration in range(self.max_iterations):
            print(f"\n--- 第 {iteration + 1} 回目の反復 ---")

            # フェーズ1-2：要件分析とタスク分解
            if iteration == 0:
                requirements = self.phase1_understand(user_input)
                plan = self.phase2_decompose(requirements)

            # フェーズ3：実行
            results = self.phase3_execute(requirements, plan)

            # フェーズ4：検証
            verification = self.phase4_verify(requirements, results)

            if verification["verified"]:
                # フェーズ5：納品
                report = self.phase5_deliver(requirements, results, verification)

                print("\n" + "=" * 60)
                print("タスク完了！")
                elapsed = time.time() - self.start_time
                print(f"総所要時間：{elapsed:.1f} 秒")
                print(f"総反復回数：{iteration + 1} 回")
                print(f"サブタスク数：{len(results)}")
                print(f"品質スコア：{verification['overall_score']}")
                print("=" * 60)

                return {
                    "report": report,
                    "requirements": requirements,
                    "verification": verification,
                    "execution_log": self.execution_log,
                    "elapsed_seconds": elapsed
                }
            else:
                self.log("反復", "検証が不合格のため、計画を調整して再実行します")
                # 検証フィードバックに基づいて計画を調整
                plan = self.phase2_decompose({
                    **requirements,
                    "feedback": verification["issues"],
                    "missing": verification["missing_topics"]
                })

        return {"error": "最大反復回数に達したため、タスクは完了しませんでした"}

# 使用例
assistant = AIResearchAssistant(max_iterations=10)
result = assistant.run(
    "2024年のAI Agent技術の発展現状と将来動向を調査し、"
    "エンタープライズ向けアプリケーションシナリオに焦点を当て、"
    "技術的意思決定者向けに、構造化された調査レポートを出力してください。"
)
print(f"\n調査レポート：\n{result['report']}")

re>
実行結果と主要指標
上記のコードを実行すると、Agent Loop の完全な実行プロセスが表示されます：
要件分析：研究テーマ、範囲、深さ、対象読者などの情報を抽出
タスク分解：研究を背景調査、技術分析、応用シナリオ、競争環境、トレンド予測などのサブタスクに分解
段階的実行：各サブタスクが独立して実行され、詳細な分析結果を出力
結果検証：情報の正確性、網羅性、論理的一貫性をチェック
最終納品：実行統計を含む構造化された調査レポートを生成
主要な設計ポイント：
ログシステム：各段階で実行ログを記録し、デバッグと監査を容易にする
フィードバックループ：検証が失敗した場合、自動的にタスク分解段階に戻り、再計画する
コンテキスト受け渡し：依存関係を通じてサブタスク間でコンテキストを渡し、情報の一貫性を保証
品質の定量化：スコアリングメカニズムを使用して出力品質を定量化し、停止条件の判断基準とする
本番環境へのデプロイ推奨事項
この Agent Loop アプリケーションを本番環境にデプロイするには、以下の点を考慮する必要があります：
非同期実行：メッセージキュー（Redis、RabbitMQ など）を使用して非同期タスク処理を実装し、長時間のブロッキングを回避
状態永続化：Agent の実行状態をデータベースに保存し、レジューム機能と進捗クエリをサポート
並行性制御：LLM API 呼び出しにレート制限とリトライメカニズムを実装し、スロットリングを回避
コスト監視：各呼び出しのトークン消費を記録し、予算アラートを設定
人間によるレビューノード：重要な決定ポイント（最終納品前など）に人間のレビューステップを追加
A/B テスト：異なるプロンプト戦略の効果を比較し、Agent のパフォーマンスを継続的に最適化
まとめと展望
これら5つのチュートリアルを通じて、Agent Loop の基本概念から始まり、ツール呼び出し、Reflection、マルチエージェント協調へと徐々に深掘りし、最終的に完全な Agent Loop アプリケーションを構築しました。Agent Loop は遠い概念ではなく、すぐに取り組める開発パラダイムです。
Agent Loop の未来はエキサイティングです：自律プログラミング、自動化された科学研究、インテリジェント運用——これらのシナリオは想像から現実へと変わりつつあります。Agent Loop をマスターすれば、次世代 AI アプリケーションを構築する中核能力を手に入れたことになります。今すぐスキルチェーンを開いて、自分自身で最初の Agent Loop をオーケストレーションしましょう！
このスキルチェーンを自分でオーケストレーションしたいですか？
スキルチェーンで開く →