AutoGen Group Chat：多Agent協調と投票意思決定メカニズムの実践的実装

AutoGenはMicrosoftが開発したマルチエージェントAIフレームワークであり、複数のAIエージェントを協調させて複雑なタスクを処理できます。本稿では、HolySheep AIのAPIを活用したAutoGen Group Chatの基本設定から、投票ベース意思決定メカニズムの実装まで、筆者が実務で直面した課題と解決策を含めて解説します。

AutoGen Group Chatアーキテクチャの概要

AutoGenのGroup Chatは、複数のエージェントが一つの会話グループ内で相互に通信しながらタスクを解決する仕組みです。筆者が初めてこの機能を試みた際、単純なチェーン状の呼び出しでは対応できない複雑な判断が必要となる場面に直面しました。例えば、コードレビューエージェント、テスト生成エージェント、ドキュメント作成エージェントが連携する場合、各エージェントの提案を統合的に評価する仕組みが必須となります。

基本的なGroup Chatの実装

まずHolySheep AIをproviderとしてAutoGenエージェントを構成する基本的なコードを示します。HolySheepの無料クレジット付き登録を活用すれば、GPT-4.1が$8/MTok、DeepSeek V3.2が$0.42/MTokという破格の料金で実験できます。

import autogen
from autogen.agentchat.group.chat import GroupChat
from autogen.agentchat.agent import Agent

HolySheep AI API設定
config_list = [{
    "model": "gpt-4.1",
    "api_type": "openai",
    "base_url": "https://api.holysheep.ai/v1",
    "api_key": "YOUR_HOLYSHEEP_API_KEY",
}]

llm_config = {
    "config_list": config_list,
    "temperature": 0.7,
    "max_tokens": 2000,
}

専門エージェントの定義
reviewer = autogen.AssistantAgent(
    name="Reviewer",
    system_message="あなたはコードレビュー担当者です。コードの質と潜在的な問題を報告してください。",
    llm_config=llm_config,
)

tester = autogen.AssistantAgent(
    name="Tester",
    system_message="あなたはテスト設計担当者です。必要なテストケースを提案してください。",
    llm_config=llm_config,
)

optimizer = autogen.AssistantAgent(
    name="Optimizer",
    system_message="あなたはパフォーマンス最適化担当者です。コードの改善点を提案してください。",
    llm_config=llm_config,
)

Group Chatの設定
group_chat = GroupChat(
    agents=[reviewer, tester, optimizer],
    max_round=10,
    speaker_selection_method="round_robin",  #  발언자 순환 방식
)

manager = autogen.GroupChatManager(groupchat=group_chat, llm_config=llm_config)

グループチャット開始
initializer = autogen.UserProxyAgent(name="User")

initializer.initiate_chat(
    manager,
    message="以下のPythonコードを三人でレビューしてください：\n\ndef calculate_fibonacci(n):\n    if n <= 1:\n        return n\n    return calculate_fibonacci(n-1) + calculate_fibonacci(n-2)",
)

このコードを実行すると、三つのエージェントが順番に发言し、各エージェントの意見を他のエージェントが参照しながら会話が進みます。筆者が初めて実行した際は HolySheep AI のAPI呼び出しで ConnectionError: timeout が発生しましたが、これは後述するエラーハンドリングで解決できます。

投票ベース意思決定メカニズムの実装

Group Chatの基本機能だけでは、複数の提案から最適なものを選択する機能がありません。筆者が実務で開発した投票メカニズムの実装例を示します。

import autogen
from typing import List, Dict
from dataclasses import dataclass
from enum import Enum

class VoteOption(Enum):
    APPROVE = "承認"
    REJECT = "却下"
    REVISE = "修正依頼"

@dataclass
class Vote:
    agent_name: str
    option: VoteOption
    reason: str

class VotingManager:
    def __init__(self, agents: List[Agent], llm_config: dict):
        self.agents = agents
        self.llm_config = llm_config
        self.votes: List[Vote] = []
    
    def collect_votes(self, proposals: List[str]) -> Dict[str, int]:
        """各エージェントから投票を収集"""
        self.votes = []
        
        for agent in self.agents:
            vote_prompt = f"""
以下の提案から投票を行ってください：

{'='*50}
{proposals}
{'='*50}

投票形式の例：
オプション: APPROVE/REJECT/REVISE
理由: [投票理由]
"""
            # HolySheep APIを呼び出して投票を取得
            response = agent.generate_reply(messages=[{
                "role": "user", 
                "content": vote_prompt
            }])
            
            # 投票結果をパース（実際の実装ではより堅牢なパースが必要）
            self.votes.append(Vote(
                agent_name=agent.name,
                option=VoteOption.APPROVE,  # 実際の実装ではresponseをパース
                reason="投票理由"
            ))
        
        # 投票集計
        vote_counts = {opt.value: 0 for opt in VoteOption}
        for vote in self.votes:
            vote_counts[vote.option.value] += 1
        
        return vote_counts
    
    def majority_decision(self, proposals: List[str]) -> str:
        """多数決で最終決定を導く"""
        vote_counts = self.collect_votes(proposals)
        
        # 最多得票のオプションを決定
        winning_option = max(vote_counts, key=vote_counts.get)
        
        print(f"\n{'='*50}")
        print("投票結果:")
        for option, count in vote_counts.items():
            print(f"  {option}: {count}票")
        print(f"決定: {winning_option}")
        print(f"{'='*50}\n")
        
        return winning_option

使用例
config_list = [{
    "model": "deepseek-v3.2",
    "api_type": "openai",
    "base_url": "https://api.holysheep.ai/v1",
    "api_key": "YOUR_HOLYSHEEP_API_KEY",
}]

llm_config = {"config_list": config_list, "temperature": 0.3}

voter1 = autogen.AssistantAgent(name="SecurityExpert", 
    system_message="セキュリティ観点から評価", llm_config=llm_config)
voter2 = autogen.AssistantAgent(name="PerformanceExpert",
    system_message="パフォーマンス観点から評価", llm_config=llm_config)
voter3 = autogen.AssistantAgent(name="MaintainabilityExpert",
    system_message="保守性観点から評価", llm_config=llm_config)

voting_manager = VotingManager(
    agents=[voter1, voter2, voter3],
    llm_config=llm_config
)

proposals = [
    "案A: 暗号化AES-256を使用",
    "案B: 暗号化AES-128を使用",
    "案C: ハッシュ化のみ実施"
]

decision = voting_manager.majority_decision(proposals)

この実装では、各エージェントが специализированные 관점에서提案を評価し、投票を行います。HolySheep AI の DeepSeek V3.2 を使用すれば、$0.42/MTok という低コストで大量のエージェント間通信を実験できます。WeChat Pay や Alipay にも対応しているため、日本の開発者でも簡単に決済できます。

実用的なネゴシエーションフローの構築

より実践的なシナリオとして、要件定義から実装方針決定までのネゴシエーションフローを実装しました。このシステムでは、各エージェントが自分の視点を主張しつつ、最終的にコンセンサスを得るまで議論を続けます。

import autogen
import json
from typing import Optional

class NegotiationAgent(autogen.AssistantAgent):
    """ネゴシエーション特化エージェント"""
    
    def __init__(self, name: str, perspective: str, llm_config: dict):
        super().__init__(
            name=name,
            system_message=f"""あなたは{perspective}の専門家です。
あなたの役割:
1. {perspective}観点から問題を分析すること
2. 他のエージェントと建設的に議論すること
3. 必要に応じて妥協案を提案すること
4. 最終的な合意形成に貢献すること

出力形式:
ANALYSIS: [あなたの分析]
PROPOSAL: [あなたの提案]
COMPROMISE: [妥協可能な点]
""",
            llm_config=llm_config
        )
        self.perspective = perspective
        self.initial_position: Optional[str] = None
    
    def set_position(self, position: str):
        self.initial_position = position
    
    def evaluate_proposal(self, proposal: str) -> str:
        """提案を自分の観点から評価"""
        evaluation = self.generate_reply(messages=[{
            "role": "user",
            "content": f"""あなたの観点({self.perspective})から以下の提案を評価してください:

{proposal}

評価基準:
1. この観点から見た优点
2. この観点から見た課題
3. 改善提案

簡潔に述べてください。"""
        }])
        return evaluation

def run_negotiation(
    topic: str,
    perspectives: list[str],
    holy_sheep_api_key: str
) -> dict:
    """ネゴシエーションセッションを実行"""
    
    config_list = [{
        "model": "gpt-4.1",
        "api_type": "openai", 
        "base_url": "https://api.holysheep.ai/v1",
        "api_key": holy_sheep_api_key,
        "timeout": 120,  # HolySheepは<50msレイテンシですが、安全策としてtimeout設定
    }]
    
    llm_config = {"config_list": config_list, "temperature": 0.6}
    
    # エージェント作成
    agents = {
        perspective: NegotiationAgent(
            name=f"{perspective}担当",
            perspective=perspective,
            llm_config=llm_config
        )
        for perspective in perspectives
    }
    
    # 初期立場を設定
    for perspective, agent in agents.items():
        agent.set_position(f"{perspective}の観点から最適な解決策")
    
    # ネゴシエーショ|round実行
    discussion_history = []
    
    for round_num in range(3):
        print(f"\n=== ネゴシエーショ|round {round_num + 1} ===\n")
        
        for perspective, agent in agents.items():
            # 各エージェントの意見を収集
            response = agent.generate_reply(messages=[{
                "role": "system",
                "content": f"トピック: {topic}\n現在の|round: {round_num + 1}"
            }])
            
            discussion_history.append({
                "round": round_num + 1,
                "agent": perspective,
                "response": response
            })
            
            print(f"[{perspective}]")
            print(response[:200] + "..." if len(response) > 200 else response)
            print()
    
    # 最終投票で合意形成
    voting_prompt = f"""
以下の{topic}に関する議論を踏まえ、全員が同意できる解決策を提案してください:

{json.dumps(discussion_history, ensure_ascii=False, indent=2)}
"""
    
    # 司会者（UserProxyAgent）で最終提案を生成
    facilitator = autogen.UserProxyAgent(name="司会者")
    final_proposal = facilitator.generate_reply(messages=[{
        "role": "user",
        "content": voting_prompt
    }])
    
    return {
        "discussion": discussion_history,
        "final_proposal": final_proposal
    }

実行例
if __name__ == "__main__":
    result = run_negotiation(
        topic="ECサイトのデータベース設計方针決定",
        perspectives=["セキュリティ", "パフォーマンス", "スケーラビリティ", "コスト"],
        holy_sheep_api_key="YOUR_HOLYSHEEP_API_KEY"
    )
    
    print("\n" + "="*60)
    print("最終合意案:")
    print(result["final_proposal"])
    print("="*60)

筆者がこのシステムを実際に使った際の問題は、各エージェントの応答が有时候矛盾することです。例えば、セキュリティ担当者が「AES-256を强制」と主張し、パフォーマンス担当者が「レイテンシ増加を懸念」と反論した場合、人間の司会者なしに自動合意形成するのは困難でした。この問題を解決するため、最後にUserProxyAgentを司会者として配置し、議論を統合する設計にしています。

よくあるエラーと対処法

エラー1: ConnectionError: timeout

# 問題: API呼び出し時のタイムアウト
urllib3.exceptions.ReadTimeoutError: HTTPSConnectionPool
原因: ネットワーク遅延またはAPIの過負荷

解決策1: timeout設定の追加
config_list = [{
    "model": "gpt-4.1",
    "api_type": "openai",
    "base_url": "https://api.holysheep.ai/v1",
    "api_key": "YOUR_HOLYSHEEP_API_KEY",
    "timeout": 180,  # タイムアウトを180秒に設定
    "max_retries": 3,  # リトライ回数を設定
}]

解決策2: tenacity用于自动重试
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_api_with_retry(agent, message):
    try:
        return agent.generate_reply(messages=[{"role": "user", "content": message}])
    except Exception as e:
        print(f"API呼び出し失敗: {e}")
        raise

解決策3: 替代提供商切换
fallback_config_list = [
    # 主provider: HolySheep (¥1=$1, 85%節約)
    {
        "model": "gpt-4.1",
        "api_type": "openai",
        "base_url": "https://api.holysheep.ai/v1",
        "api_key": "YOUR_HOLYSHEEP_API_KEY",
    },
    # 替代provider: DeepSeek (更便宜)
    {
        "model": "deepseek-v3.2",
        "api_type": "openai",
        "base_url": "https://api.holysheep.ai/v1",
        "api_key": "YOUR_HOLYSHEEP_API_KEY",
    },
]

llm_config = {"config_list": fallback_config_list}

エラー2: 401 Unauthorized

# 問題: APIキーが無効または期限切れ
原因: 
1. APIキーが正しく設定されていない
2. キーが無効化されている
3. スコープが適切に設定されていない

解決策1: 環境変数からの安全なキー読み込み
import os
from dotenv import load_dotenv

load_dotenv()  # .envファイルから読み込み

config_list = [{
    "model": "gpt-4.1",
    "api_type": "openai",
    "base_url": "https://api.holysheep.ai/v1",
    "api_key": os.environ.get("HOLYSHEEP_API_KEY"),  # 環境変数から取得
}]

解決策2: キー有効性の事前検証
import requests

def verify_api_key(api_key: str) -> bool:
    """APIキーの有効性をチェック"""
    try:
        response = requests.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            },
            json={
                "model": "gpt-4.1",
                "messages": [{"role": "user", "content": "test"}],
                "max_tokens": 5
            },
            timeout=10
        )
        return response.status_code == 200
    except Exception as e:
        print(f"API検証エラー: {e}")
        return False

使用前の検証
if not verify_api_key(os.environ.get("HOLYSHEEP_API_KEY", "")):
    raise ValueError("無効なAPIキーです。https://www.holysheep.ai/register で再取得してください。")

解決策3: .envファイルの例
HOLYSHEEP_API_KEY=sk-your-actual-api-key-here
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1

エラー3: RateLimitError: too many requests

# 問題: API呼び出しがレート制限に達した
原因: 
1. 短时间に大量のリクエストを送信
2. プランの制限を超えた

解決策1: rate limiterの実装
import time
import threading
from collections import deque

class RateLimiter:
    def __init__(self, max_calls: int, time_window: float):
        self.max_calls = max_calls
        self.time_window = time_window
        self.calls = deque()
        self.lock = threading.Lock()
    
    def wait(self):
        with self.lock:
            now = time.time()
            # 時間枠外の呼び出し記録を削除
            while self.calls and self.calls[0] < now - self.time_window:
                self.calls.popleft()
            
            if len(self.calls) >= self.max_calls:
                sleep_time = self.time_window - (now - self.calls[0])
                if sleep_time > 0:
                    print(f"レート制限: {sleep_time:.1f}秒待機")
                    time.sleep(sleep_time)
            
            self.calls.append(time.time())

使用例: 1秒間に10リクエストまで
rate_limiter = RateLimiter(max_calls=10, time_window=1.0)

def throttled_api_call(agent, message):
    rate_limiter.wait()
    return agent.generate_reply(messages=[{"role": "user", "content": message}])

解決策2: 非同期リクエストのバッチ处理
import asyncio

async def batch_api_call(agents_messages: list, batch_size: int = 5):
    """複数のAPI呼び出しをバッチで処理"""
    results = []
    
    for i in range(0, len(agents_messages), batch_size):
        batch = agents_messages[i:i+batch_size]
        
        # バッチ内のリクエストを并行実行
        tasks = [
            throttled_api_call(agent, msg)
            for agent, msg in batch
        ]
        
        batch_results = await asyncio.gather(*tasks, return_exceptions=True)
        results.extend(batch_results)
        
        # HolySheepの<50msレイテンシを活かしつつ、批量間の待機
        if i + batch_size < len(agents_messages):
            await asyncio.sleep(0.5)
    
    return results

解決策3: 缓存机制用于重复请求
from functools import lru_cache
import hashlib

def cache_key(message: str, model: str) -> str:
    return hashlib.md5(f"{model}:{message}".encode()).hexdigest()

@lru_cache(maxsize=100)
def cached_api_call(message: str, model: str):
    """ 동일한 요청의 결과를 캐시 """
    # 実際の実装ではAPIを呼び出す
    return None  # プレースホルダー

エラー4: 無効なJSON応答の処理

# 問題: Agentからの応答が有効なJSON形式でない
原因:
1. LLMの出力フォーマットが不安定
2. 特殊文字によるJSON構造の破損

解決策1: JSONパースの再試行机制
import json
import re

def extract_json(text: str) -> dict:
    """テキストからJSONを抽出"""
    # 方法1: ``json `` ブロックを探す
    json_match = re.search(r'``json\s*([\s\S]*?)\s*``', text)
    if json_match:
        try:
            return json.loads(json_match.group(1))
        except json.JSONDecodeError:
            pass
    
    # 方法2: 最初の { から最後の } まで
    brace_start = text.find('{')
    brace_end = text.rfind('}')
    if brace_start != -1 and brace_end != -1:
        json_str = text[brace_start:brace_end+1]
        try:
            return json.loads(json_str)
        except json.JSONDecodeError:
            pass
    
    # 方法3: 後処理で修復
    cleaned = text.replace("'", '"').replace("\n", " ")
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError:
        return {"raw_response": text, "error": "JSON parsing failed"}

解決策2: Pydantic模型用于严格的输出验证
from pydantic import BaseModel, Field, validator

class VoteResult(BaseModel):
    option: str = Field(..., pattern="^(APPROVE|REJECT|REVISE)$")
    reason: str = Field(..., min_length=10)
    confidence: float = Field(..., ge=0.0, le=1.0)
    
    @validator('option')
    def validate_option(cls, v):
        valid_options = ["APPROVE", "REJECT", "REVISE"]
        if v.upper() in valid_options:
            return v.upper()
        raise ValueError(f"Invalid option: {v}")

def parse_vote_response(text: str) -> VoteResult:
    """投票応答を安全にパース"""
    try:
        json_data = extract_json(text)
        return VoteResult(**json_data)
    except Exception as e:
        print(f"パースエラー: {e}")
        # フォールバック: デフォルト値
        return VoteResult(
            option="REVISE",
            reason="パース失敗によるデフォルト",
            confidence=0.0
        )

#
関連リソース
📚 AI API 記事一覧
💰 料金を見る
📖 開発者ドキュメント
🚀 無料登録
関連記事
GPT-4o 統計分析：回帰分析予測モデリングAPI实战
Kimi K2搭載リサーチアシスタントの作り方：HolySheep AIで始める高性能AI活用
LangGraph Memory：長期記憶と短期コンテキスト管理の最適化戦略