エージェントファクトリー完全ガイド2025 - モジュール型AI自動化パイプライン構築

エージェントファクトリーは、再利用可能な AI コンポーネントを組み合わせて、 複雑な自動化パイプラインを構築するためのアーキテクチャパターンです。 本記事では、モジュール型設計によるスケーラブルな AI システムの構築方法を解説します。

目次

1. エージェントファクトリーとは
2. アーキテクチャ設計
3. コアコンポーネントの実装
4. パイプライン構築システム
5. エージェント間の協調メカニズム
6. 動的ワークフロー生成
7. スケーラビリティとパフォーマンス
8. 実装例とユースケース

エージェントファクトリーとは

エージェントファクトリーは、AI エージェントを「製造」するための設計パターンです。 工場のように、標準化されたパーツ（モジュール）から、用途に応じたエージェントを効率的に生成します。

主な特徴

アーキテクチャ設計

レイヤード・アーキテクチャ

コアアーキテクチャの実装

# src/core/factory.py
from abc import ABC, abstractmethod
from typing import Dict, List, Any, Type
import asyncio
from dataclasses import dataclass

@dataclass
class AgentConfig:
    """エージェント構成の定義"""
    name: str
    modules: List[Dict[str, Any]]
    pipeline: List[str]
    metadata: Dict[str, Any] = None

class Module(ABC):
    """基底モジュールクラス"""
    @abstractmethod
    async def execute(self, context: 'Context') -> 'Context':
        pass
    
    @abstractmethod
    def validate_config(self, config: Dict) -> bool:
        pass

class Context:
    """実行コンテキスト"""
    def __init__(self, initial_data: Any):
        self.data = initial_data
        self.history = []
        self.metadata = {}
        self.error = None
    
    def add_result(self, module_name: str, result: Any):
        self.history.append({
            'module': module_name,
            'result': result,
            'timestamp': asyncio.get_event_loop().time()
        })

class AgentFactory:
    """エージェントファクトリーのメイン実装"""
    def __init__(self):
        self.module_registry: Dict[str, Type[Module]] = {}
        self.templates: Dict[str, AgentConfig] = {}
    
    def register_module(self, name: str, module_class: Type[Module]):
        """モジュールの登録"""
        self.module_registry[name] = module_class
    
    def create_agent(self, config: AgentConfig) -> 'Agent':
        """設定からエージェントを生成"""
        modules = self._instantiate_modules(config)
        pipeline = self._build_pipeline(config, modules)
        
        return Agent(
            name=config.name,
            modules=modules,
            pipeline=pipeline,
            config=config
        )
    
    def _instantiate_modules(self, config: AgentConfig) -> Dict[str, Module]:
        """モジュールのインスタンス化"""
        modules = {}
        for module_config in config.modules:
            module_type = module_config['type']
            module_name = module_config['name']
            
            if module_type not in self.module_registry:
                raise ValueError(f"Unknown module type: {module_type}")
            
            module_class = self.module_registry[module_type]
            modules[module_name] = module_class(**module_config.get('params', {}))
        
        return modules
    
    def _build_pipeline(self, config: AgentConfig, modules: Dict[str, Module]) -> List[Module]:
        """パイプラインの構築"""
        pipeline = []
        for module_name in config.pipeline:
            if module_name not in modules:
                raise ValueError(f"Module not found in pipeline: {module_name}")
            pipeline.append(modules[module_name])
        
        return pipeline

コアコンポーネントの実装

1. LLMモジュール

# src/modules/llm_module.py
from typing import Dict, Any, Optional
import asyncio
from langchain_core.language_models import BaseLLM
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

class LLMModule(Module):
    """汎用LLMモジュール"""
    
    def __init__(
        self,
        model_type: str = "openai",
        model_name: str = "gpt-4",
        temperature: float = 0.7,
        max_tokens: int = 2000,
        **kwargs
    ):
        self.model = self._create_model(model_type, model_name, temperature, max_tokens, **kwargs)
        self.prompt_template = kwargs.get('prompt_template', "{input}")
    
    def _create_model(self, model_type: str, model_name: str, temperature: float, max_tokens: int, **kwargs) -> BaseLLM:
        """モデルのファクトリーメソッド"""
        if model_type == "openai":
            return ChatOpenAI(
                model=model_name,
                temperature=temperature,
                max_tokens=max_tokens
            )
        elif model_type == "anthropic":
            return ChatAnthropic(
                model=model_name,
                temperature=temperature,
                max_tokens=max_tokens
            )
        else:
            raise ValueError(f"Unsupported model type: {model_type}")
    
    async def execute(self, context: Context) -> Context:
        """LLM実行"""
        try:
            prompt = self.prompt_template.format(input=context.data)
            response = await self.model.ainvoke(prompt)
            
            context.data = response.content
            context.add_result('llm', {
                'prompt': prompt,
                'response': response.content,
                'tokens_used': response.usage_metadata
            })
        except Exception as e:
            context.error = f"LLM execution failed: {str(e)}"
        
        return context
    
    def validate_config(self, config: Dict) -> bool:
        required_fields = ['model_type', 'model_name']
        return all(field in config for field in required_fields)

2. ツール実行モジュール

# src/modules/tool_module.py
from typing import List, Dict, Callable, Any
import asyncio
import json

class ToolModule(Module):
    """ツール実行モジュール"""
    
    def __init__(self, tools: List[Dict[str, Any]], selection_strategy: str = "auto"):
        self.tools = self._register_tools(tools)
        self.selection_strategy = selection_strategy
    
    def _register_tools(self, tool_configs: List[Dict]) -> Dict[str, Callable]:
        """ツールの登録"""
        tools = {}
        for config in tool_configs:
            tool_name = config['name']
            tool_func = self._create_tool_function(config)
            tools[tool_name] = tool_func
        return tools
    
    def _create_tool_function(self, config: Dict) -> Callable:
        """ツール関数の動的生成"""
        if config['type'] == 'api':
            return self._create_api_tool(config)
        elif config['type'] == 'function':
            return self._create_function_tool(config)
        else:
            raise ValueError(f"Unknown tool type: {config['type']}")
    
    async def execute(self, context: Context) -> Context:
        """ツールの実行"""
        # ツール選択
        selected_tool = await self._select_tool(context)
        
        if selected_tool:
            # ツール実行
            result = await self._execute_tool(selected_tool, context)
            context.data = result
            context.add_result('tool', {
                'tool_name': selected_tool,
                'result': result
            })
        
        return context
    
    async def _select_tool(self, context: Context) -> Optional[str]:
        """コンテキストに基づくツール選択"""
        if self.selection_strategy == "auto":
            # AIによる自動選択
            return await self._ai_select_tool(context)
        elif self.selection_strategy == "rule":
            # ルールベース選択
            return self._rule_based_selection(context)
        else:
            # 手動指定
            return context.metadata.get('selected_tool')

3. メモリモジュール

# src/modules/memory_module.py
from typing import List, Dict, Any, Optional
import asyncio
from datetime import datetime
import faiss
import numpy as np

class MemoryModule(Module):
    """インテリジェントメモリモジュール"""
    
    def __init__(
        self,
        memory_type: str = "vector",
        capacity: int = 1000,
        embedding_model: str = "text-embedding-ada-002"
    ):
        self.memory_type = memory_type
        self.capacity = capacity
        self.memories: List[Dict] = []
        
        if memory_type == "vector":
            self.index = faiss.IndexFlatL2(1536)  # OpenAI埋め込みの次元
            self.embedder = self._create_embedder(embedding_model)
    
    async def execute(self, context: Context) -> Context:
        """メモリ操作の実行"""
        operation = context.metadata.get('memory_operation', 'store')
        
        if operation == 'store':
            await self._store_memory(context)
        elif operation == 'retrieve':
            memories = await self._retrieve_memories(context)
            context.metadata['retrieved_memories'] = memories
        elif operation == 'update':
            await self._update_memory(context)
        
        return context
    
    async def _store_memory(self, context: Context):
        """メモリの保存"""
        memory_entry = {
            'content': context.data,
            'timestamp': datetime.now().isoformat(),
            'metadata': context.metadata.copy(),
            'context_id': context.metadata.get('session_id')
        }
        
        if self.memory_type == "vector":
            # ベクトル化して保存
            embedding = await self._get_embedding(context.data)
            self.index.add(np.array([embedding]))
            memory_entry['embedding_id'] = len(self.memories)
        
        self.memories.append(memory_entry)
        
        # 容量制限
        if len(self.memories) > self.capacity:
            self._evict_old_memories()
    
    async def _retrieve_memories(self, context: Context, k: int = 5) -> List[Dict]:
        """関連メモリの取得"""
        query = context.data
        
        if self.memory_type == "vector":
            # ベクトル検索
            query_embedding = await self._get_embedding(query)
            distances, indices = self.index.search(np.array([query_embedding]), k)
            
            return [self.memories[idx] for idx in indices[0] if idx < len(self.memories)]
        else:
            # キーワード検索
            return self._keyword_search(query, k)

パイプライン構築システム

動的パイプラインビルダー

# src/core/pipeline_builder.py
from typing import List, Dict, Set, Tuple
import networkx as nx
from concurrent.futures import ThreadPoolExecutor
import asyncio

class PipelineBuilder:
    """動的パイプラインビルダー"""
    
    def __init__(self):
        self.optimization_rules = []
        self.validators = []
    
    def build_pipeline(
        self,
        modules: Dict[str, Module],
        config: Dict[str, Any]
    ) -> 'Pipeline':
        """パイプラインの構築"""
        # 依存関係グラフの作成
        dependency_graph = self._build_dependency_graph(modules, config)
        
        # 検証
        self._validate_pipeline(dependency_graph, modules)
        
        # 最適化
        optimized_graph = self._optimize_pipeline(dependency_graph)
        
        # 実行計画の生成
        execution_plan = self._generate_execution_plan(optimized_graph)
        
        return Pipeline(modules, execution_plan)
    
    def _build_dependency_graph(
        self,
        modules: Dict[str, Module],
        config: Dict[str, Any]
    ) -> nx.DiGraph:
        """依存関係グラフの構築"""
        graph = nx.DiGraph()
        
        # ノードの追加
        for module_name in modules:
            graph.add_node(module_name, module=modules[module_name])
        
        # エッジの追加（依存関係）
        for module_name, deps in config.get('dependencies', {}).items():
            for dep in deps:
                graph.add_edge(dep, module_name)
        
        return graph
    
    def _optimize_pipeline(self, graph: nx.DiGraph) -> nx.DiGraph:
        """パイプラインの最適化"""
        optimized = graph.copy()
        
        # 並列実行可能なモジュールの特定
        parallel_groups = self._identify_parallel_groups(optimized)
        
        # 不要なモジュールの削除
        self._remove_redundant_modules(optimized)
        
        # キャッシュ戦略の適用
        self._apply_caching_strategy(optimized)
        
        return optimized
    
    def _identify_parallel_groups(self, graph: nx.DiGraph) -> List[Set[str]]:
        """並列実行可能なグループの特定"""
        levels = nx.topological_generations(graph)
        return list(levels)

class Pipeline:
    """実行可能なパイプライン"""
    
    def __init__(self, modules: Dict[str, Module], execution_plan: List[List[str]]):
        self.modules = modules
        self.execution_plan = execution_plan
        self.metrics = PipelineMetrics()
    
    async def execute(self, context: Context) -> Context:
        """パイプラインの実行"""
        for parallel_group in self.execution_plan:
            if len(parallel_group) == 1:
                # シーケンシャル実行
                module_name = parallel_group[0]
                context = await self._execute_module(module_name, context)
            else:
                # 並列実行
                context = await self._execute_parallel(parallel_group, context)
        
        return context
    
    async def _execute_parallel(
        self,
        module_names: List[str],
        context: Context
    ) -> Context:
        """並列モジュール実行"""
        # コンテキストの分岐
        sub_contexts = [context.copy() for _ in module_names]
        
        # 並列実行
        tasks = [
            self._execute_module(name, ctx)
            for name, ctx in zip(module_names, sub_contexts)
        ]
        
        results = await asyncio.gather(*tasks)
        
        # 結果のマージ
        return self._merge_contexts(results)

エージェント間の協調メカニズム

マルチエージェントオーケストレーター

# src/orchestration/multi_agent.py
from typing import List, Dict, Any, Optional
import asyncio
from abc import ABC, abstractmethod
import uuid

class AgentMessage:
    """エージェント間メッセージ"""
    def __init__(
        self,
        sender: str,
        receiver: str,
        content: Any,
        message_type: str = "task"
    ):
        self.id = str(uuid.uuid4())
        self.sender = sender
        self.receiver = receiver
        self.content = content
        self.message_type = message_type
        self.timestamp = asyncio.get_event_loop().time()

class MultiAgentOrchestrator:
    """マルチエージェントオーケストレーター"""
    
    def __init__(self, coordination_strategy: str = "hierarchical"):
        self.agents: Dict[str, Agent] = {}
        self.message_queue = asyncio.Queue()
        self.coordination_strategy = coordination_strategy
        self.running = False
    
    def register_agent(self, agent: Agent):
        """エージェントの登録"""
        self.agents[agent.name] = agent
        agent.set_orchestrator(self)
    
    async def start(self):
        """オーケストレーターの開始"""
        self.running = True
        
        # メッセージルーターの起動
        router_task = asyncio.create_task(self._message_router())
        
        # 協調戦略の初期化
        if self.coordination_strategy == "hierarchical":
            await self._setup_hierarchical_structure()
        elif self.coordination_strategy == "mesh":
            await self._setup_mesh_network()
        
        await router_task
    
    async def _message_router(self):
        """メッセージルーティング"""
        while self.running:
            try:
                message = await asyncio.wait_for(
                    self.message_queue.get(),
                    timeout=1.0
                )
                
                if message.receiver == "broadcast":
                    # ブロードキャスト
                    for agent_name, agent in self.agents.items():
                        if agent_name != message.sender:
                            await agent.receive_message(message)
                else:
                    # 特定エージェントへの送信
                    if message.receiver in self.agents:
                        await self.agents[message.receiver].receive_message(message)
                
            except asyncio.TimeoutError:
                continue
    
    async def execute_task(self, task: Dict[str, Any]) -> Any:
        """タスクの実行"""
        if self.coordination_strategy == "hierarchical":
            return await self._hierarchical_execution(task)
        elif self.coordination_strategy == "pipeline":
            return await self._pipeline_execution(task)
        elif self.coordination_strategy == "voting":
            return await self._voting_execution(task)
        else:
            raise ValueError(f"Unknown coordination strategy: {self.coordination_strategy}")

エージェント協調の実装例

# src/agents/specialized_agents.py

class ResearchAgent(Agent):
    """リサーチ専門エージェント"""
    
    async def process_task(self, task: Dict) -> Dict:
        # 情報収集
        research_results = await self._gather_information(task['query'])
        
        # 分析
        analysis = await self._analyze_information(research_results)
        
        # 他エージェントへの委譲判断
        if self._needs_expert_opinion(analysis):
            expert_response = await self.delegate_to_agent(
                'expert_agent',
                {'analysis': analysis, 'question': task['query']}
            )
            analysis['expert_opinion'] = expert_response
        
        return {
            'research': research_results,
            'analysis': analysis,
            'confidence': self._calculate_confidence(analysis)
        }

class WriterAgent(Agent):
    """執筆専門エージェント"""
    
    async def process_task(self, task: Dict) -> Dict:
        # リサーチ結果の取得
        if 'research' not in task:
            research = await self.request_from_agent(
                'research_agent',
                {'query': task['topic']}
            )
            task['research'] = research
        
        # コンテンツ生成
        content = await self._generate_content(
            topic=task['topic'],
            research=task['research'],
            style=task.get('style', 'professional')
        )
        
        # レビュー依頼
        if task.get('require_review', True):
            review = await self.request_from_agent(
                'reviewer_agent',
                {'content': content}
            )
            
            if review['needs_revision']:
                content = await self._revise_content(content, review['feedback'])
        
        return {
            'content': content,
            'metadata': self._generate_metadata(content)
        }

動的ワークフロー生成

# src/workflow/dynamic_generator.py

class DynamicWorkflowGenerator:
    """動的ワークフロー生成器"""
    
    def __init__(self, factory: AgentFactory):
        self.factory = factory
        self.workflow_templates = {}
        self.optimization_history = []
    
    async def generate_workflow(
        self,
        task_description: str,
        constraints: Dict[str, Any] = None
    ) -> Workflow:
        """タスクから動的にワークフローを生成"""
        # タスク分析
        task_analysis = await self._analyze_task(task_description)
        
        # 必要なコンポーネントの特定
        required_components = self._identify_components(task_analysis)
        
        # ワークフロー構造の設計
        workflow_structure = await self._design_workflow(
            task_analysis,
            required_components,
            constraints
        )
        
        # エージェントの生成と構成
        agents = self._create_agents(workflow_structure)
        
        # ワークフローの最適化
        optimized_workflow = self._optimize_workflow(
            workflow_structure,
            agents,
            self.optimization_history
        )
        
        return Workflow(
            name=f"dynamic_workflow_{uuid.uuid4().hex[:8]}",
            agents=agents,
            structure=optimized_workflow
        )
    
    async def _analyze_task(self, task_description: str) -> Dict:
        """AIによるタスク分析"""
        prompt = f"""
        以下のタスクを分析して、必要な処理ステップを特定してください：
        
        タスク: {task_description}
        
        分析項目:
        1. タスクの種類（データ処理、文書生成、分析など）
        2. 必要な入力データ
        3. 期待される出力
        4. 処理ステップ
        5. 必要なツールやAPI
        """
        
        analysis = await self.llm_analyzer.analyze(prompt)
        return self._parse_task_analysis(analysis)

スケーラビリティとパフォーマンス

分散実行アーキテクチャ

# src/distributed/executor.py
import ray
from typing import List, Dict, Any
import asyncio

@ray.remote
class DistributedAgent:
    """分散実行可能なエージェント"""
    
    def __init__(self, agent_config: AgentConfig):
        self.agent = AgentFactory().create_agent(agent_config)
    
    async def execute(self, task: Dict) -> Dict:
        return await self.agent.process(task)

class DistributedExecutor:
    """分散実行エンジン"""
    
    def __init__(self, num_workers: int = 4):
        ray.init(num_cpus=num_workers)
        self.worker_pool = []
        self.load_balancer = LoadBalancer()
    
    async def execute_workflow(
        self,
        workflow: Workflow,
        input_data: Any
    ) -> Any:
        # ワークフローを分散可能なタスクに分解
        tasks = self._decompose_workflow(workflow)
        
        # タスクの並列実行
        futures = []
        for task in tasks:
            worker = self._select_worker(task)
            future = worker.execute.remote(task)
            futures.append(future)
        
        # 結果の収集とマージ
        results = await asyncio.gather(*[
            self._ray_to_asyncio(f) for f in futures
        ])
        
        return self._merge_results(results)
    
    def _select_worker(self, task: Dict) -> DistributedAgent:
        """負荷分散によるワーカー選択"""
        return self.load_balancer.select_worker(
            self.worker_pool,
            task
        )

実装例とユースケース

1. カスタマーサポート自動化

# examples/customer_support.py

# カスタマーサポート用エージェント設定
support_config = AgentConfig(
    name="customer_support_agent",
    modules=[
        {
            "name": "intent_classifier",
            "type": "llm",
            "params": {
                "model_type": "openai",
                "model_name": "gpt-4",
                "prompt_template": """
                顧客の問い合わせを分類してください：
                問い合わせ: {input}
                
                カテゴリ: [技術的問題, 請求関連, 製品情報, その他]
                """
            }
        },
        {
            "name": "knowledge_retriever",
            "type": "vector_search",
            "params": {
                "index_name": "support_knowledge_base",
                "top_k": 5
            }
        },
        {
            "name": "response_generator",
            "type": "llm",
            "params": {
                "model_type": "anthropic",
                "model_name": "claude-3-opus",
                "prompt_template": """
                顧客の問い合わせに対して、以下の情報を基に回答を生成してください：
                
                問い合わせ: {query}
                関連情報: {knowledge}
                
                回答は親切で分かりやすく、具体的な解決策を含めてください。
                """
            }
        },
        {
            "name": "escalation_checker",
            "type": "rule_engine",
            "params": {
                "rules": [
                    {"condition": "sentiment < -0.5", "action": "escalate"},
                    {"condition": "complexity > 0.8", "action": "escalate"}
                ]
            }
        }
    ],
    pipeline=["intent_classifier", "knowledge_retriever", "response_generator", "escalation_checker"]
)

# エージェントの生成と実行
factory = AgentFactory()
support_agent = factory.create_agent(support_config)

# 顧客問い合わせの処理
async def handle_customer_inquiry(inquiry: str) -> Dict:
    context = Context(inquiry)
    result = await support_agent.process(context)
    
    if result.metadata.get('escalate'):
        # 人間のオペレーターにエスカレーション
        return await escalate_to_human(result)
    
    return {
        'response': result.data,
        'confidence': result.metadata.get('confidence', 0.9),
        'suggested_actions': result.metadata.get('actions', [])
    }

2. データ分析パイプライン

# examples/data_analysis_pipeline.py

analysis_config = AgentConfig(
    name="data_analysis_pipeline",
    modules=[
        {
            "name": "data_ingestion",
            "type": "etl",
            "params": {
                "sources": ["database", "api", "files"],
                "validation_rules": {...}
            }
        },
        {
            "name": "data_cleaning",
            "type": "processor",
            "params": {
                "operations": ["remove_duplicates", "handle_missing", "normalize"]
            }
        },
        {
            "name": "feature_engineering",
            "type": "ml_transformer",
            "params": {
                "strategies": ["auto_feature", "polynomial", "interactions"]
            }
        },
        {
            "name": "model_selection",
            "type": "automl",
            "params": {
                "algorithms": ["xgboost", "lightgbm", "neural_net"],
                "optimization_metric": "f1_score"
            }
        },
        {
            "name": "insight_generator",
            "type": "llm",
            "params": {
                "model_type": "openai",
                "model_name": "gpt-4",
                "prompt_template": """
                分析結果から重要なインサイトを抽出してください：
                
                データ概要: {data_summary}
                モデル結果: {model_results}
                特徴量重要度: {feature_importance}
                
                ビジネスに活用できる具体的な提案を含めてください。
                """
            }
        }
    ],
    pipeline=["data_ingestion", "data_cleaning", "feature_engineering", "model_selection", "insight_generator"]
)

ベストプラクティス

まとめ

エージェントファクトリーの導入により、AI システムの開発速度が 3 倍になり、 新しい要求への対応時間が数週間から数日に短縮されました。 モジュール型アーキテクチャのおかげで、チーム間の並行開発も可能になり、 全体的な生産性が大幅に向上しています。