AgentEvolver実践ガイド：自己進化するAIエージェントシステムをローカル環境で構築する

// 仮想環境作成
python3 -m venv agentevolver-env
source agentevolver-env/bin/activate  // Windowsの場合: agentevolver-env\Scripts\activate

// 基本パッケージのインストール
pip install --upgrade pip
pip install torch transformers langchain langchain-community

// リポジトリのクローン
git clone https://github.com/modelscope/AgentEvolver.git
cd AgentEvolver

// 依存関係のインストール
pip install -r requirements.txt

// 開発モードでインストール
pip install -e .

// Ollamaのインストール（Linuxの場合）
curl -fsSL https://ollama.com/install.sh | sh

// モデルのダウンロード
ollama pull llama3.1:8b

// Ollamaサーバーの起動確認
ollama list

// config/agent_config.yaml
llm:
  type: ollama
  model: llama3.1:8b
  base_url: http://localhost:11434
  temperature: 0.7

evolution:
  strategy: gradient_based
  max_iterations: 5
  convergence_threshold: 0.95

evaluation:
  metrics:
    - accuracy
    - completeness
    - efficiency
  weights:
    accuracy: 0.5
    completeness: 0.3
    efficiency: 0.2

from agentevolver import Agent
from langchain_community.llms import Ollama

class CodeReviewAgent(Agent):
    def __init__(self, llm):
        prompt_template = """
あなたは経験豊富なPythonエンジニアです。
以下のコードをレビューし、改善点を3つ以上提案してください。

コード:
{code}

レビュー観点:

コードの可読性
パフォーマンス
セキュリティ
Pythonic な書き方

各改善点について、理由と具体的な修正例を示してください。
"""
        super().__init__(llm, prompt_template)

    def review_code(self, code):
        return self.execute(code)

// エージェントの初期化
llm = Ollama(model="llama3.1:8b")
agent = CodeReviewAgent(llm)

from agentevolver import Evaluator

class CodeReviewEvaluator(Evaluator):
    def evaluate(self, code, review_output):
        score = 0.0
        feedback = []

        // 評価基準1: 改善点の数
        num_improvements = self._count_improvements(review_output)
        if num_improvements >= 3:
            score += 0.3
        else:
            feedback.append(f"改善点が{num_improvements}個しかありません。最低3個必要です。")

        // 評価基準2: 具体的なコード例の有無
        has_code_examples = self._has_code_examples(review_output)
        if has_code_examples:
            score += 0.3
        else:
            feedback.append("具体的なコード例が不足しています。")

        // 評価基準3: 説明の明確さ
        clarity_score = self._evaluate_clarity(review_output)
        score += clarity_score * 0.4

        return score, feedback

    def _count_improvements(self, text):
        // 改善点を数える簡易実装
        return text.count("改善点") + text.count("提案")

    def _has_code_examples(self, text):
        return "```" in text or "def " in text

    def _evaluate_clarity(self, text):
        // 説明の明確さを評価（簡易版）
        word_count = len(text.split())
        if word_count < 50:
            return 0.3  // 説明が短すぎる
        elif word_count > 500:
            return 0.5  // 冗長すぎる
        else:
            return 1.0  // 適切な長さ

evaluator = CodeReviewEvaluator()

from agentevolver import Evolver

class CodeReviewEvolver(Evolver):
    def evolve(self, agent, evaluation_results):
        // 評価が低い部分を分析
        avg_score = sum(r['score'] for r in evaluation_results) / len(evaluation_results)

        if avg_score < 0.5:
            // プロンプトに具体的な指示を追加
            improved_prompt = agent.prompt_template + """

重要: 各改善点について、必ず以下を含めてください：

問題の説明（なぜ改善が必要か）
修正前のコード例
修正後のコード例
期待される効果
"""
        elif avg_score < 0.8:
            // 評価基準を強調
            improved_prompt = agent.prompt_template.replace(
                "3つ以上提案してください",
                "正確に5つ提案してください。各提案は具体的なコード例を含めること"
            )
        else:
            // 既に高品質なので微調整のみ
            improved_prompt = agent.prompt_template

        // 新しいエージェントを返す
        return CodeReviewAgent(agent.llm)

evolver = CodeReviewEvolver()

def evolution_cycle(agent, evaluator, evolver, test_cases, max_iterations=5):
    current_agent = agent

    for iteration in range(max_iterations):
        print(f"\n=== Iteration {iteration + 1} ===")

        // テストケースでエージェントを評価
        evaluation_results = []
        for code in test_cases:
            review = current_agent.review_code(code)
            score, feedback = evaluator.evaluate(code, review)
            evaluation_results.append({
                'score': score,
                'feedback': feedback,
                'review': review
            })
            print(f"Score: {score:.2f}, Feedback: {feedback}")

        // 平均スコアを計算
        avg_score = sum(r['score'] for r in evaluation_results) / len(evaluation_results)
        print(f"Average Score: {avg_score:.2f}")

        // 収束判定
        if avg_score >= 0.95:
            print("収束しました。進化を終了します。")
            break

        // エージェントを進化
        current_agent = evolver.evolve(current_agent, evaluation_results)

    return current_agent

// テストケースの準備
test_cases = [
    """
def calculate(a, b):
    return a + b
""",
    """
data = []
for i in range(1000):
    data.append(i * 2)
""",
]

// 進化サイクルの実行
final_agent = evolution_cycle(agent, evaluator, evolver, test_cases)

def composite_evaluation(output, expected):
    accuracy = calculate_accuracy(output, expected)
    completeness = calculate_completeness(output)
    efficiency = calculate_efficiency(output)

    // 重み付き平均
    total_score = (
        accuracy * 0.5 +
        completeness * 0.3 +
        efficiency * 0.2
    )
    return total_score

class HumanInTheLoopEvaluator(Evaluator):
    def evaluate(self, task, output):
        // 自動評価
        auto_score = self.auto_evaluate(output)

        // 低スコアの場合のみ人間に確認
        if auto_score < 0.7:
            human_score = self.request_human_review(task, output)
            final_score = (auto_score + human_score) / 2
        else:
            final_score = auto_score

        return final_score

class ErrorAnalyzer:
    def analyze_failures(self, evaluation_results):
        failure_patterns = {
            'missing_details': [],
            'incorrect_format': [],
            'logic_errors': []
        }

        for result in evaluation_results:
            if result['score'] < 0.6:
                error_type = self.classify_error(result['output'])
                failure_patterns[error_type].append(result)

        return failure_patterns

    def generate_improvement_suggestions(self, failure_patterns):
        suggestions = []

        if len(failure_patterns['missing_details']) > 3:
            suggestions.append("プロンプトに詳細な出力形式の例を追加する")

        if len(failure_patterns['logic_errors']) > 2:
            suggestions.append("推論ステップを明示的に指示する")

        return suggestions

def compare_evolution_strategies(agent, test_cases):
    strategies = [
        GradientBasedEvolver(),
        GeneticEvolver(),
        ReinforcementEvolver()
    ]

    results = {}

    for strategy in strategies:
        evolved_agent = strategy.evolve(agent, test_cases)
        performance = evaluate_agent(evolved_agent, test_cases)
        results[strategy.__class__.__name__] = performance

    // 最良の戦略を選択
    best_strategy = max(results, key=results.get)
    print(f"最良の戦略: {best_strategy} (スコア: {results[best_strategy]:.2f})")

    return best_strategy

import logging
from datetime import datetime

class AgentMonitor:
    def __init__(self):
        self.metrics = []
        self.logger = logging.getLogger(__name__)

    def log_execution(self, agent_id, task, result, duration):
        metric = {
            'timestamp': datetime.now().isoformat(),
            'agent_id': agent_id,
            'task': task,
            'success': result.get('success', False),
            'score': result.get('score', 0),
            'duration_ms': duration * 1000
        }
        self.metrics.append(metric)
        self.logger.info(f"Agent {agent_id}: score={metric['score']:.2f}, duration={metric['duration_ms']:.0f}ms")

    def get_performance_summary(self, agent_id, time_window_hours=24):
        recent_metrics = [m for m in self.metrics
                         if m['agent_id'] == agent_id and
                         self._is_recent(m['timestamp'], time_window_hours)]

        if not recent_metrics:
            return None

        avg_score = sum(m['score'] for m in recent_metrics) / len(recent_metrics)
        avg_duration = sum(m['duration_ms'] for m in recent_metrics) / len(recent_metrics)
        success_rate = sum(1 for m in recent_metrics if m['success']) / len(recent_metrics)

        return {
            'avg_score': avg_score,
            'avg_duration_ms': avg_duration,
            'success_rate': success_rate,
            'total_executions': len(recent_metrics)
        }

monitor = AgentMonitor()

class PerformanceAlerter:
    def __init__(self, monitor, thresholds):
        self.monitor = monitor
        self.thresholds = thresholds

    def check_alerts(self, agent_id):
        summary = self.monitor.get_performance_summary(agent_id)

        if not summary:
            return

        alerts = []

        if summary['avg_score'] < self.thresholds['min_score']:
            alerts.append(f"低スコア警告: {summary['avg_score']:.2f}")

        if summary['avg_duration_ms'] > self.thresholds['max_duration_ms']:
            alerts.append(f"遅延警告: {summary['avg_duration_ms']:.0f}ms")

        if summary['success_rate'] < self.thresholds['min_success_rate']:
            alerts.append(f"成功率低下: {summary['success_rate']:.1%}")

        for alert in alerts:
            self.send_alert(agent_id, alert)

    def send_alert(self, agent_id, message):
        // Slack、メールなどに通知
        print(f"[ALERT] Agent {agent_id}: {message}")

alerter = PerformanceAlerter(monitor, {
    'min_score': 0.7,
    'max_duration_ms': 5000,
    'min_success_rate': 0.9
})

from concurrent.futures import ThreadPoolExecutor, as_completed

class AgentPool:
    def __init__(self, agent_factory, pool_size=5):
        self.agents = [agent_factory() for _ in range(pool_size)]
        self.executor = ThreadPoolExecutor(max_workers=pool_size)

    def execute_batch(self, tasks):
        futures = []

        for i, task in enumerate(tasks):
            agent = self.agents[i % len(self.agents)]
            future = self.executor.submit(agent.execute, task)
            futures.append((task, future))

        results = []
        for task, future in futures:
            try:
                result = future.result(timeout=30)
                results.append({'task': task, 'result': result, 'success': True})
            except Exception as e:
                results.append({'task': task, 'error': str(e), 'success': False})

        return results

// 使用例
pool = AgentPool(lambda: CodeReviewAgent(llm), pool_size=5)
tasks = [code1, code2, code3, code4, code5]
results = pool.execute_batch(tasks)

import hashlib
import json

class AgentCache:
    def __init__(self):
        self.cache = {}

    def get_cache_key(self, agent_id, task):
        task_hash = hashlib.md5(json.dumps(task, sort_keys=True).encode()).hexdigest()
        return f"{agent_id}:{task_hash}"

    def get(self, agent_id, task):
        key = self.get_cache_key(agent_id, task)
        return self.cache.get(key)

    def set(self, agent_id, task, result, ttl_seconds=3600):
        key = self.get_cache_key(agent_id, task)
        self.cache[key] = {
            'result': result,
            'timestamp': datetime.now(),
            'ttl_seconds': ttl_seconds
        }

    def execute_with_cache(self, agent, task):
        // キャッシュをチェック
        cached = self.get(agent.id, task)
        if cached and not self._is_expired(cached):
            return cached['result']

        // キャッシュミス：実行して結果をキャッシュ
        result = agent.execute(task)
        self.set(agent.id, task, result)
        return result

cache = AgentCache()

class AgentVersionManager:
    def __init__(self):
        self.versions = {}

    def save_version(self, agent, version_tag):
        self.versions[version_tag] = {
            'prompt_template': agent.prompt_template,
            'config': agent.config,
            'timestamp': datetime.now()
        }
        print(f"バージョン {version_tag} を保存しました")

    def load_version(self, version_tag):
        if version_tag not in self.versions:
            raise ValueError(f"バージョン {version_tag} が見つかりません")

        version_data = self.versions[version_tag]
        print(f"バージョン {version_tag} を読み込みました")
        return version_data

    def rollback(self, agent, version_tag):
        version_data = self.load_version(version_tag)
        agent.prompt_template = version_data['prompt_template']
        agent.config = version_data['config']
        print(f"バージョン {version_tag} にロールバックしました")

version_manager = AgentVersionManager()

// バージョンの保存
version_manager.save_version(agent, "v1.0-stable")

// 進化後に問題があればロールバック
// version_manager.rollback(agent, "v1.0-stable")