[ad_1]

エージェント型AI（Agentic AI）の実装方法【2025年版実践ガイド】¶

はじめに¶

2025年の技術トレンドとして注目されているエージェント型AI（Agentic AI）は、従来のAIとは異なり、自律的に意思決定し、目標達成のために独自に計画を立てて実行する革新的なシステムです。この記事では、Python と OpenAI API を使用して、実際に動作するエージェント型AIシステムを構築する方法を詳しく解説します。

[ad_1]

完成すると、タスクを自動分解し、各ステップを順次実行するAIエージェントが完成し、複雑な問題解決を自動化できるようになります。

必要な知識・環境¶

前提知識¶

• Python基礎知識（クラス、関数、モジュール）
• REST API の基本概念
• 非同期プログラミングの理解

開発環境¶

• Python 3.9 以上
• OpenAI API キー
• VS Code または PyCharm（推奨）

使用するライブラリ・パッケージ¶

• openai>=1.0.0
• aiohttp>=3.8.0
• asyncio
• json
• dataclasses
• typing
• logging

プロジェクトのセットアップ¶

💡 開発環境をもっと快適に！
Visual Studio Code や WebStorm などの高機能エディタで開発効率アップ！

ディレクトリ構造¶

agentic_ai_system/
├── main.py
├── agents/
│   ├── __init__.py
│   ├── base_agent.py
│   ├── planner_agent.py
│   └── executor_agent.py
├── tools/
│   ├── __init__.py
│   ├── web_search.py
│   └── file_operations.py
├── config/
│   ├── __init__.py
│   └── settings.py
└── requirements.txt

必要なパッケージのインストール¶

# プロジェクトディレクトリ作成
mkdir agentic_ai_system
cd agentic_ai_system

# Poetry を使用したプロジェクト初期化
poetry init
poetry add openai aiohttp

設定ファイルの作成¶

config/settings.pyを作成：

import os
from dataclasses import dataclass

@dataclass
class Settings:
    OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "")
    MODEL_NAME: str = "gpt-4"
    MAX_ITERATIONS: int = 10
    TIMEOUT: int = 30

settings = Settings()

実装手順（ステップバイステップ）¶

🔧 開発効率アップツール
GitHub Copilot でAI支援コーディングを体験しよう！

ステップ1: ベースエージェントクラスの作成¶

agents/base_agent.py：

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
import logging
import asyncio
from openai import AsyncOpenAI
from config.settings import settings

@dataclass
class Task:
    """タスクを表すデータクラス"""
    id: str
    description: str
    status: str = "pending"  # pending, in_progress, completed, failed
    result: Optional[str] = None
    dependencies: List[str] = None

    def __post_init__(self):
        if self.dependencies is None:
            self.dependencies = []

class BaseAgent(ABC):
    """エージェントの基底クラス"""

    def __init__(self, name: str, role: str):
        self.name = name
        self.role = role
        self.client = AsyncOpenAI(api_key=settings.OPENAI_API_KEY)
        self.logger = logging.getLogger(f"Agent.{name}")

    @abstractmethod
    async def process(self, task: Task) -> Task:
        """タスクを処理する抽象メソッド"""
        pass

    async def call_llm(self, prompt: str, system_message: str = None) -> str:
        """LLMを呼び出すヘルパーメソッド"""
        messages = []
        if system_message:
            messages.append({"role": "system", "content": system_message})
        messages.append({"role": "user", "content": prompt})

        try:
            response = await self.client.chat.completions.create(
                model=settings.MODEL_NAME,
                messages=messages,
                temperature=0.7,
                max_tokens=1000
            )
            return response.choices[0].message.content.strip()
        except Exception as e:
            self.logger.error(f"LLM呼び出しエラー: {e}")
            raise

ステップ2: プランナーエージェントの実装¶

agents/planner_agent.py：

import json
from typing import List
from .base_agent import BaseAgent, Task

class PlannerAgent(BaseAgent):
    """タスクを分解し、実行計画を立てるエージェント"""

    def __init__(self):
        super().__init__("Planner", "タスク分解と計画立案")

    async def process(self, task: Task) -> List[Task]:
        """複雑なタスクを小さなサブタスクに分解"""
        self.logger.info(f"タスクを分解中: {task.description}")

        system_message = """
        あなたは優秀なプロジェクトマネージャーです。
        与えられた複雑なタスクを、実行可能な小さなサブタスクに分解してください。
        各サブタスクは明確で具体的である必要があります。
        """

        prompt = f"""
        以下のタスクを実行可能なサブタスクに分解してください：

        タスク: {task.description}

        以下のJSON形式で出力してください：
        {{
            "subtasks": [
                {{
                    "id": "unique_id",
                    "description": "具体的なタスク説明",
                    "dependencies": ["依存するタスクのID（あれば）"]
                }}
            ]
        }}
        """

        try:
            response = await self.call_llm(prompt, system_message)
            plan_data = json.loads(response)

            subtasks = []
            for i, subtask_data in enumerate(plan_data["subtasks"]):
                subtask = Task(
                    id=subtask_data.get("id", f"subtask_{i}"),
                    description=subtask_data["description"],
                    dependencies=subtask_data.get("dependencies", [])
                )
                subtasks.append(subtask)

            self.logger.info(f"{len(subtasks)}個のサブタスクに分解完了")
            return subtasks

        except json.JSONDecodeError as e:
            self.logger.error(f"JSON解析エラー: {e}")
            # フォールバック: 単一タスクとして返す
            return [task]
        except Exception as e:
            self.logger.error(f"タスク分解エラー: {e}")
            raise

[ad_2]

ステップ3: エグゼキューターエージェントの実装¶

agents/executor_agent.py：

import asyncio
from typing import Dict, Any
from .base_agent import BaseAgent, Task

class ExecutorAgent(BaseAgent):
    """個別のタスクを実行するエージェント"""

    def __init__(self):
        super().__init__("Executor", "タスク実行")
        self.tools = self._load_tools()

    def _load_tools(self) -> Dict[str, Any]:
        """利用可能なツールを読み込み"""
        # 実際の実装では、より多くのツールを追加
        return {
            "web_search": self._web_search,
            "file_read": self._file_read,
            "calculation": self._calculation,
        }

    async def process(self, task: Task) -> Task:
        """タスクを実行"""
        self.logger.info(f"タスク実行中: {task.description}")
        task.status = "in_progress"

        system_message = f"""
        あなたは{self.role}です。
        与えられたタスクを実行し、結果を返してください。
        利用可能なツール: {list(self.tools.keys())}

        タスクを実行できない場合は、その理由を明確に説明してください。
        """

        prompt = f"""
        以下のタスクを実行してください：

        タスク: {task.description}

        実行方法を説明し、必要に応じてツールを使用してください。
        最終的な結果を明確に示してください。
        """

        try:
            # タスクの種類を判断
            task_type = await self._analyze_task_type(task.description)

            if task_type in self.tools:
                result = await self.tools[task_type](task.description)
            else:
                result = await self.call_llm(prompt, system_message)

            task.result = result
            task.status = "completed"
            self.logger.info(f"タスク完了: {task.id}")

        except Exception as e:
            task.status = "failed"
            task.result = f"エラー: {str(e)}"
            self.logger.error(f"タスク実行失敗: {e}")

        return task

    async def _analyze_task_type(self, description: str) -> str:
        """タスクの種類を分析"""
        # 簡易実装：キーワードベース
        if "検索" in description or "調べ" in description:
            return "web_search"
        elif "ファイル" in description or "読み" in description:
            return "file_read"
        elif "計算" in description or "数値" in description:
            return "calculation"
        else:
            return "general"

    async def _web_search(self, query: str) -> str:
        """Web検索の模擬実装"""
        await asyncio.sleep(1)  # 模擬的な処理時間
        return f"'{query}'の検索結果: [模擬的な検索結果]"

    async def _file_read(self, description: str) -> str:
        """ファイル読み込みの模擬実装"""
        await asyncio.sleep(0.5)
        return f"ファイル操作完了: {description}"

    async def _calculation(self, description: str) -> str:
        """計算の模擬実装"""
        return f"計算実行: {description} = [計算結果]"

ステップ4: メインのエージェントシステム作成¶

[ad_2]

main.py：

import asyncio
import logging
from typing import List, Dict
from agents.planner_agent import PlannerAgent
from agents.executor_agent import ExecutorAgent
from agents.base_agent import Task

class AgenticAISystem:
    """エージェント型AIシステムのメインクラス"""

    def __init__(self):
        self.planner = PlannerAgent()
        self.executor = ExecutorAgent()
        self.logger = logging.getLogger("AgenticAI")

    async def execute_goal(self, goal: str) -> Dict[str, any]:
        """目標を達成するためのタスクを実行"""
        self.logger.info(f"目標設定: {goal}")

        # 1. 目標をタスクに変換
        main_task = Task(id="main", description=goal)

        # 2. タスクを分解
        subtasks = await self.planner.process(main_task)

        # 3. 依存関係を考慮した実行順序を決定
        execution_order = self._resolve_dependencies(subtasks)

        # 4. タスクを順次実行
        results = []
        for task_id in execution_order:
            task = next(t for t in subtasks if t.id == task_id)
            completed_task = await self.executor.process(task)
            results.append(completed_task)

            self.logger.info(f"進捗: {len(results)}/{len(subtasks)} 完了")

        return {
            "goal": goal,
            "subtasks_completed": len([r for r in results if r.status == "completed"]),
            "total_subtasks": len(results),
            "results": [{"id": r.id, "description": r.description, "result": r.result, "status": r.status} for r in results]
        }

    def _resolve_dependencies(self, tasks: List[Task]) -> List[str]:
        """タスクの依存関係を解決して実行順序を決定"""
        # 簡易実装：依存関係のないタスクから順に実行
        no_deps = [t.id for t in tasks if not t.dependencies]
        with_deps = [t.id for t in tasks if t.dependencies]
        return no_deps + with_deps

async def main():
    """メイン実行関数"""
    # ロギング設定
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )

    # エージェントシステム初期化
    ai_system = AgenticAISystem()

    # 目標設定と実行
    goal = "Python Webアプリケーションの開発計画を立てて、必要な技術スタックを調査する"

    print(f"🎯 目標: {goal}")
    print("=" * 60)

    try:
        result = await ai_system.execute_goal(goal)

        print(f"\n✅ 実行完了!")
        print(f"📊 完了タスク: {result['subtasks_completed']}/{result['total_subtasks']}")
        print(f"\n📋 実行結果:")

        for i, task_result in enumerate(result['results'], 1):
            status_emoji = "✅" if task_result['status'] == "completed" else "❌"
            print(f"\n{i}. {status_emoji} {task_result['description']}")
            print(f"   結果: {task_result['result'][:100]}...")

    except Exception as e:
        print(f"❌ エラーが発生しました: {e}")

if __name__ == "__main__":
    asyncio.run(main())

テストとデバッグ¶

📚 エラー解決スキルアップ！
リーダブルコード でコード品質を向上させましょう。

動作確認方法¶

1. 環境変数設定：

export OPENAI_API_KEY="your-api-key-here"

1. 基本実行テスト：

python main.py

1. 個別エージェントテスト：

# test_agents.py
import asyncio
from agents.planner_agent import PlannerAgent
from agents.base_agent import Task

async def test_planner():
    planner = PlannerAgent()
    task = Task(id="test", description="Webサイトを作成する")
    subtasks = await planner.process(task)
    print(f"分解されたタスク数: {len(subtasks)}")
    for subtask in subtasks:
        print(f"- {subtask.description}")

asyncio.run(test_planner())

よくあるエラーと対処法¶

[ad_3]

1. OpenAI API キーエラー

openai.AuthenticationError: Incorrect API key provided

対処法：
– API キーが正しく設定されているか確認
– 環境変数 OPENAI_API_KEY が適切に設定されているか確認

2. JSON解析エラー

json.JSONDecodeError: Expecting value

対処法：
– LLMレスポンスの前後処理を追加
– フォールバック処理を実装

3. タスク実行タイムアウト

asyncio.TimeoutError

対処法：
– settings.py の TIMEOUT 値を調整
– 長時間タスクの分割を検討

応用例・カスタマイズ¶

1. 専門領域エージェントの追加¶

class DataAnalysisAgent(BaseAgent):
    """データ分析専門エージェント"""

    async def process(self, task: Task) -> Task:
        # pandas, matplotlib等を使用した分析処理
        pass

class CodeReviewAgent(BaseAgent):
    """コードレビュー専門エージェント"""

    async def process(self, task: Task) -> Task:
        # ソースコード解析とレビュー処理
        pass

2. 外部APIとの連携¶

async def _web_search_real(self, query: str) -> str:
    """実際のWeb検索API連携"""
    import aiohttp
    async with aiohttp.ClientSession() as session:
        # Google Custom Search API、Bing API等との連携
        pass

3. 永続化とログ機能¶

import sqlite3
from datetime import datetime

class TaskLogger:
    """タスク実行履歴の永続化"""

    def __init__(self, db_path: str = "tasks.db"):
        self.db_path = db_path
        self._init_db()

    def log_task(self, task: Task):
        # SQLiteにタスク情報を保存
        pass

パフォーマンス最適化¶

並列処理の実装¶

async def execute_parallel_tasks(self, tasks: List[Task]) -> List[Task]:
    """依存関係のないタスクを並列実行"""
    independent_tasks = [t for t in tasks if not t.dependencies]

    # 並列実行
    results = await asyncio.gather(*[
        self.executor.process(task) for task in independent_tasks
    ])

    return results

キャッシュ機能¶

from functools import lru_cache
import hashlib

class CachedLLMCalls:
    """LLM呼び出し結果のキャッシュ"""

    @lru_cache(maxsize=100)
    async def cached_call_llm(self, prompt_hash: str, prompt: str) -> str:
        return await self.call_llm(prompt)

    def get_prompt_hash(self, prompt: str) -> str:
        return hashlib.md5(prompt.encode()).hexdigest()

セキュリティ考慮事項¶

1. API キーの安全な管理¶

# .env ファイルの使用
from dotenv import load_dotenv
load_dotenv()

# 実行時のキー検証
def validate_api_key():
    if not settings.OPENAI_API_KEY:
        raise ValueError("OpenAI API キーが設定されていません")

2. 入力値の検証¶

import re

def sanitize_task_description(description: str) -> str:
    """タスク説明文のサニタイズ"""
    # 危険な文字列の除去
    dangerous_patterns = [r'<script.*?</script>', r'javascript:', r'eval\(']
    for pattern in dangerous_patterns:
        description = re.sub(pattern, '', description, flags=re.IGNORECASE)
    return description.strip()

3. レート制限の実装¶

import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, calls_per_minute: int = 60):
        self.calls_per_minute = calls_per_minute
        self.calls = defaultdict(list)

    async def wait_if_needed(self):
        now = time.time()
        minute_ago = now - 60

        # 1分以内の呼び出し数をチェック
        recent_calls = [t for t in self.calls['api'] if t > minute_ago]

        if len(recent_calls) >= self.calls_per_minute:
            wait_time = 60 - (now - recent_calls[0])
            await asyncio.sleep(wait_time)

        self.calls['api'].append(now)

まとめとNext Steps¶

この記事では、エージェント型AI（Agentic AI）の基本的な実装方法を学びました：

実装のポイント¶

1. モジュール化設計: 各エージェントが独立した役割を持つ
2. 非同期処理: 効率的なタスク実行
3. エラーハンドリング: 堅牢性の確保
4. 拡張性: 新しいエージェントやツールの追加が容易

Next Steps¶

1. 実際のAPI連携: Web検索、データベース、外部サービス
2. 機械学習モデル統合: カスタムモデルの組み込み
3. GUI開発: Streamlit、Gradio等を使用したユーザーインターフェース
4. 本格運用: Docker化、CI/CD、監視システム

エージェント型AIは2025年の重要技術トレンドです。この基盤を元に、より高度な自律的AIシステムの開発に挑戦してみてください。

関連記事：
– [OpenAI API活用ガイド【2025年版】]
– [Python非同期プログラミング入門]
– [AI エージェント設計パターン集]

おすすめ関連商品¶

実装スキル向上におすすめの書籍¶

1. Clean Architecture
2. 保守性の高いアーキテクチャ設計

3. 実装パターンの理解

4. Web API実践設計

5. REST API設計のベストプラクティス

6. 実装時の考慮事項

7. プログラマーの数学

8. アルゴリズムの数学的基礎
9. 効率的な実装のための数学知識

開発効率を上げるツール・サービス¶

• GitHub Copilot – AI支援コーディング
• AWS Free Tier – クラウド環境で実践
• O’Reilly Learning – 技術書読み放題

[ad_3]