エージェント型AI開発の手法とポイント¶

class ReflectiveAgent:
    def __init__(self, llm, task_memory):
        self.llm = llm
        self.task_memory = task_memory

    def execute_with_reflection(self, task):
        # 初期実行
        initial_result = self.execute_task(task)

        # 自己評価
        reflection_prompt = f"""
        タスク: {task}
        実行結果: {initial_result}

        以下の観点で自己評価してください：
        1. 目標達成度 (1-10)
        2. 改善可能な点
        3. 次回への学習事項

        改善案があれば、修正した解決策も提示してください。
        """

        reflection = self.llm.generate(reflection_prompt)

        # 必要に応じて再実行
        if self.needs_improvement(reflection):
            return self.execute_task(task, improvements=reflection)

        return initial_result

class ToolEnabledAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = {
            'web_search': self.web_search,
            'code_executor': self.execute_code,
            'file_manager': self.manage_files,
            'api_caller': self.call_api
        }

    def solve_with_tools(self, problem):
        planning_prompt = f"""
        問題: {problem}

        利用可能なツール:
        {list(self.tools.keys())}

        解決手順を計画してください:
        1. 必要なツールの特定
        2. ツール使用順序
        3. 各ステップの期待結果
        """

        plan = self.llm.generate(planning_prompt)
        return self.execute_plan(plan)

class PlanningAgent:
    def __init__(self, llm):
        self.llm = llm

    def create_dynamic_plan(self, goal, constraints):
        planning_prompt = f"""
        目標: {goal}
        制約: {constraints}

        以下を含む詳細な実行計画を作成してください：
        1. サブタスクの分解
        2. 依存関係の整理
        3. リソース要件
        4. リスクと軽減策
        5. 成功指標

        JSON形式で出力してください。
        """

        plan = self.llm.generate(planning_prompt)
        return self.execute_adaptive_plan(plan)

    def execute_adaptive_plan(self, plan):
        for step in plan['steps']:
            result = self.execute_step(step)

            # 計画の動的調整
            if not self.is_step_successful(result):
                updated_plan = self.replan(plan, step, result)
                return self.execute_adaptive_plan(updated_plan)

        return self.consolidate_results(plan)

class AgentTeam:
    def __init__(self):
        self.agents = {
            'analyst': AnalystAgent(),
            'developer': DeveloperAgent(),
            'tester': TesterAgent(),
            'reviewer': ReviewerAgent()
        }

    def collaborative_development(self, requirements):
        # 分析フェーズ
        analysis = self.agents['analyst'].analyze(requirements)

        # 開発フェーズ
        code = self.agents['developer'].implement(analysis)

        # テストフェーズ
        test_results = self.agents['tester'].test(code)

        # レビューフェーズ
        review = self.agents['reviewer'].review(code, test_results)

        # 協調的改善
        if not review['approved']:
            return self.iterative_improvement(requirements, review)

        return code

class WorkflowAgent:
    def __init__(self, workflow_definition):
        self.workflow = workflow_definition
        self.state = WorkflowState()

    def execute_workflow(self, input_data):
        current_step = self.workflow.get_entry_point()

        while current_step:
            # ステップの実行
            step_result = self.execute_step(current_step, input_data)

            # 状態の更新
            self.state.update(step_result)

            # 次のステップの決定
            next_step = self.determine_next_step(
                current_step, 
                step_result, 
                self.state
            )

            current_step = next_step

        return self.state.get_final_result()

class HierarchicalAgent:
    def __init__(self):
        self.coordinator = CoordinatorAgent()
        self.specialists = {
            'data_processing': DataProcessingAgent(),
            'analysis': AnalysisAgent(),
            'reporting': ReportingAgent()
        }

    def solve_complex_problem(self, problem):
        # コーディネーターが問題を分析
        decomposition = self.coordinator.decompose_problem(problem)

        # 専門エージェントに作業を委譲
        results = {}
        for task in decomposition['tasks']:
            specialist = self.specialists[task['type']]
            results[task['id']] = specialist.execute(task)

        # 結果の統合
        return self.coordinator.integrate_results(results)

from langgraph import StateGraph, END

def create_research_agent():
    workflow = StateGraph(AgentState)

    # ノードの定義
    workflow.add_node("researcher", research_node)
    workflow.add_node("analyzer", analysis_node)
    workflow.add_node("writer", writing_node)

    # エッジの定義
    workflow.add_edge("researcher", "analyzer")
    workflow.add_conditional_edges(
        "analyzer",
        should_continue_research,
        {
            "continue": "researcher",
            "write": "writer"
        }
    )
    workflow.add_edge("writer", END)

    # エントリーポイント
    workflow.set_entry_point("researcher")

    return workflow.compile()

from crewai import Agent, Task, Crew

# エージェントの定義
researcher = Agent(
    role='Research Specialist',
    goal='Conduct thorough research on given topics',
    backstory='Expert researcher with strong analytical skills',
    tools=[web_search_tool, document_analyzer]
)

writer = Agent(
    role='Technical Writer',
    goal='Create comprehensive documentation',
    backstory='Experienced technical writer',
    tools=[writing_assistant, grammar_checker]
)

# タスクの定義
research_task = Task(
    description='Research the latest trends in AI development',
    agent=researcher
)

writing_task = Task(
    description='Write a comprehensive report based on research',
    agent=writer
)

# クルーの作成
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    verbose=True
)

result = crew.kickoff()

from bee_agent_framework import Agent, LLMAdapter

# エージェントの設定
agent = Agent({
    "llm": LLMAdapter.fromModel(model),
    "memory": memory_instance,
    "tools": [calculator, web_browser, code_interpreter]
})

# 大規模ワークフロー実行
async def execute_enterprise_workflow():
    result = await agent.run({
        "prompt": "Analyze quarterly performance and generate insights",
        "context": enterprise_context
    })
    return result

class ShortTermMemory:
    def __init__(self, capacity=10):
        self.capacity = capacity
        self.conversations = deque(maxlen=capacity)
        self.current_context = {}

    def add_interaction(self, user_input, agent_response):
        self.conversations.append({
            'timestamp': datetime.now(),
            'input': user_input,
            'response': agent_response,
            'context': self.current_context.copy()
        })

    def get_recent_context(self, num_interactions=5):
        recent = list(self.conversations)[-num_interactions:]
        return self.format_context(recent)

class LongTermMemory:
    def __init__(self, vector_db, knowledge_graph):
        self.vector_db = vector_db
        self.knowledge_graph = knowledge_graph
        self.episodic_memory = EpisodicMemory()

    def store_experience(self, experience):
        # ベクターデータベースに保存
        embedding = self.generate_embedding(experience)
        self.vector_db.store(embedding, experience)

        # 知識グラフに関係性を追加
        entities = self.extract_entities(experience)
        self.knowledge_graph.add_relationships(entities)

        # エピソード記憶に保存
        self.episodic_memory.store(experience)

    def retrieve_relevant_knowledge(self, query):
        # 類似検索
        similar_experiences = self.vector_db.similarity_search(query)

        # 関連知識の取得
        related_knowledge = self.knowledge_graph.traverse(query)

        return self.combine_knowledge(similar_experiences, related_knowledge)

class AdaptiveLearningAgent:
    def __init__(self):
        self.performance_tracker = PerformanceTracker()
        self.strategy_optimizer = StrategyOptimizer()

    def learn_from_feedback(self, task, result, feedback):
        # パフォーマンスの記録
        self.performance_tracker.record(task, result, feedback)

        # 学習パターンの分析
        patterns = self.analyze_performance_patterns()

        # 戦略の調整
        if patterns.indicates_improvement_needed():
            new_strategy = self.strategy_optimizer.optimize(
                current_strategy=self.current_strategy,
                performance_data=patterns
            )
            self.update_strategy(new_strategy)

class IncrementalAgentBuilder:
    def __init__(self):
        self.development_phases = [
            "基本応答機能",
            "ツール統合",
            "メモリ機能",
            "計画立案",
            "自己改善"
        ]

    def build_phase(self, phase_name):
        if phase_name == "基本応答機能":
            return self.build_basic_responder()
        elif phase_name == "ツール統合":
            return self.add_tool_capabilities()
        # ... 各フェーズの実装

    def validate_phase(self, agent, test_cases):
        success_rate = 0
        for test_case in test_cases:
            result = agent.execute(test_case)
            if self.evaluate_result(result, test_case.expected):
                success_rate += 1
        return success_rate / len(test_cases)

class AgentTestSuite:
    def __init__(self):
        self.test_scenarios = []

    def add_scenario(self, name, input_data, expected_behavior):
        self.test_scenarios.append({
            'name': name,
            'input': input_data,
            'expected': expected_behavior,
            'evaluation_criteria': self.define_criteria(expected_behavior)
        })

    def run_tests(self, agent):
        results = {}
        for scenario in self.test_scenarios:
            result = agent.execute(scenario['input'])
            evaluation = self.evaluate_against_criteria(
                result, 
                scenario['evaluation_criteria']
            )
            results[scenario['name']] = evaluation
        return results

class AgentMonitor:
    def __init__(self):
        self.metrics = AgentMetrics()
        self.logger = StructuredLogger()

    def monitor_execution(self, agent, task):
        start_time = time.time()

        try:
            # 実行前の状態記録
            self.logger.log_state("pre_execution", agent.get_state())

            # タスク実行
            result = agent.execute(task)

            # 実行後の状態記録
            self.logger.log_state("post_execution", agent.get_state())

            # メトリクスの更新
            execution_time = time.time() - start_time
            self.metrics.record_execution(task, result, execution_time)

            return result

        except Exception as e:
            self.logger.log_error("execution_failed", str(e), agent.get_state())
            raise

    def generate_performance_report(self):
        return {
            'success_rate': self.metrics.get_success_rate(),
            'average_execution_time': self.metrics.get_avg_execution_time(),
            'error_patterns': self.metrics.get_error_patterns(),
            'improvement_suggestions': self.analyze_performance()
        }

class SecureAgent:
    def __init__(self, security_config):
        self.security_config = security_config
        self.action_validator = ActionValidator()
        self.access_controller = AccessController()

    def execute_secure_action(self, action, context):
        # アクセス権限の確認
        if not self.access_controller.has_permission(action, context.user):
            raise UnauthorizedActionError()

        # アクションの妥当性検証
        if not self.action_validator.is_safe(action):
            raise UnsafeActionError()

        # サンドボックス環境での実行
        return self.execute_in_sandbox(action)

    def execute_in_sandbox(self, action):
        # リソース制限の設定
        with ResourceLimiter(
            cpu_limit=self.security_config.cpu_limit,
            memory_limit=self.security_config.memory_limit,
            network_access=self.security_config.network_access
        ):
            return action.execute()

class AgentGovernance:
    def __init__(self):
        self.policy_engine = PolicyEngine()
        self.audit_logger = AuditLogger()
        self.compliance_checker = ComplianceChecker()

    def enforce_governance(self, agent, action):
        # ポリシー適用
        policy_result = self.policy_engine.evaluate(action)
        if not policy_result.allowed:
            self.audit_logger.log_policy_violation(action, policy_result.reason)
            raise PolicyViolationError(policy_result.reason)

        # コンプライアンスチェック
        compliance_result = self.compliance_checker.check(action)
        if not compliance_result.compliant:
            self.audit_logger.log_compliance_issue(action, compliance_result)

        # 実行の記録
        self.audit_logger.log_action_execution(agent.id, action)

        return policy_result

class OptimizedReasoningAgent:
    def __init__(self):
        self.reasoning_cache = ReasoningCache()
        self.pattern_matcher = PatternMatcher()

    def optimized_reasoning(self, problem):
        # キャッシュされた推論結果の確認
        cached_result = self.reasoning_cache.get(problem)
        if cached_result:
            return cached_result

        # パターンマッチングによる高速解決
        pattern_match = self.pattern_matcher.find_similar(problem)
        if pattern_match.confidence > 0.8:
            result = self.adapt_solution(pattern_match.solution, problem)
        else:
            result = self.full_reasoning(problem)

        # 結果のキャッシュ
        self.reasoning_cache.store(problem, result)
        return result

class ResourceManager:
    def __init__(self):
        self.resource_pool = ResourcePool()
        self.scheduler = TaskScheduler()

    def optimize_resource_usage(self, agents, tasks):
        # リソース使用量の予測
        resource_predictions = self.predict_resource_usage(agents, tasks)

        # タスクスケジューリングの最適化
        optimal_schedule = self.scheduler.optimize(
            tasks, 
            resource_predictions,
            self.resource_pool.available_resources()
        )

        return optimal_schedule