multi-agent-orchestrator

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

機能説明

マルチエージェント協調システムの設計と管理。

アーキテクチャパターン

┌─────────────┐
│ Orchestrator │ ← タスク分解、調整
└──────┬──────┘
       │
   ┌───┼───┐
   ▼   ▼   ▼
 ┌───┐┌───┐┌───┐
 │ A ││ B ││ C │ ← 専門Agent
 └───┘└───┘└───┘

コア実装

1. Agentベースクラス

interface AgentConfig {
  name: string;
  role: string;
  capabilities: string[];
  llm: LLMConfig;
  tools: Tool[];
  instructions: string;
}

class BaseAgent {
  protected config: AgentConfig;
  protected memory: AgentMemory;
  
  constructor(config: AgentConfig) {
    this.config = config;
    this.memory = new AgentMemory(config.name);
  }
  
  async think(task: Task): Promise<Response> {
    const context = await this.memory.buildContext(task.description);
    const prompt = this.buildPrompt(task, context);
    const response = await this.callLLM(prompt);
    await this.memory.add({ type: 'semantic', content: task.description + ' -> ' + response.content, importance: 8 });
    return response;
  }
  
  protected buildPrompt(task: Task, context: string): Message[] {
    return [
      { role: 'system', content: this.config.instructions },
      { role: 'system', content: context },
      { role: 'user', content: task.description }
    ];
  }
  
  protected async callLLM(messages: Message[]): Promise<Response> {
    const res = await fetch('https://api.openai.com/v1/chat/completions', {
      method: 'POST',
      headers: { 'Authorization': `Bearer ${process.env.OPENAI_API_KEY}` },
      body: JSON.stringify({ model: this.config.llm.model, messages, tools: this.config.tools.map(t => t.definition) })
    });
    return res.json();
  }
}

2. オーケストレーター

interface TaskResult {
  agentId: string;
  status: 'pending' | 'running' | 'done' | 'failed';
  output?: string;
  dependencies: string[];
  startTime?: number;
  endTime?: number;
}

class Orchestrator {
  private agents: Map<string, BaseAgent> = new Map();
  private taskGraph: DAG<Task>;
  
  constructor(private llmRouter: LLMRouter) {}
  
  registerAgent(agent: BaseAgent) {
    this.agents.set(agent.config.name, agent);
  }
  
  async execute(goal: string): Promise<string> {
    // 1. タスク分解
    const plan = await this.decompose(goal);
    
    // 2. DAG構築
    this.taskGraph = this.buildDAG(plan);
    
    // 3. 実行スケジューリング
    const results = await this.schedule();
    
    // 4. 結果集約
    return this.summarize(goal, results);
  }
  
  private async decompose(goal: string): Promise<Task[]> {
    const response = await this.llmRouter.route({
      prompt: `以下のタスクを実行可能なサブタスクに分解し、JSON配列で返却してください。
      
目標: ${goal}

要件：
- 各サブタスクは1つのAgentが担当する
- タスクの依存関係を明確にする
- 返却形式: [{"id":"t1","description":"...","agent":"researcher","depends":[]},...]`,
      system: 'あなたはタスク分解の専門家です。'
    });
    
    return JSON.parse(response.content);
  }
  
  private async schedule(): Promise<Map<string, TaskResult>> {
    const results = new Map<string, TaskResult>();
    const pending = new Set(this.taskGraph.nodes);
    const running: Promise<void>[] = [];
    const maxConcurrent = 3;
    
    while (pending.size > 0 || running.length > 0) {
      // 並列実行可能なタスクを起動
      while (running.length < maxConcurrent) {
        const next = this.findNextRunnable(pending, results);
        if (!next) break;
        
        pending.delete(next.id);
        const p = this.runTask(next, results).catch(console.error);
        running.push(p);
      }
      
      // 1つ完了を待つ
      await Promise.race(running);
      running.splice(running.findIndex(p => false), 1);
    }
    
    return results;
  }
  
  private async runTask(task: Task, results: Map<string, TaskResult>) {
    results.set(task.id, { agentId: task.agent, status: 'running', dependencies: task.depends || [], startTime: Date.now() });
    
    try {
      // 依存関係の完了を待つ
      for (const depId of task.depends || []) {
        const dep = results.get(depId);
        if (dep?.status !== 'done') {
          await this.waitFor(depId, results);
        }
      }
      
      const agent = this.agents.get(task.agent);
      const context = this.buildContext(task, results);
      const response = await agent.think({ id: task.id, description: task.description, context });
      
      results.set(task.id, { ...results.get(task.id)!, status: 'done', output: response.content, endTime: Date.now() });
    } catch (error) {
      results.set(task.id, { ...results.get(task.id)!, status: 'failed', output: String(error), endTime: Date.now() });
    }
  }
  
  private buildContext(task: Task, results: Map<string, TaskResult>): string {
    return (task.depends || []).map(depId => {
      const dep = results.get(depId);
      return dep?.output || '';
    }).join('\n\n');
  }
}

3. メッセージバス

class MessageBus {
  private subscriptions = new Map<string, Subscriber[]>();
  
  publish(channel: string, message: Message) {
    const subs = this.subscriptions.get(channel) || [];
    for (const sub of subs) {
      sub.handler(message);
    }
  }
  
  subscribe(channel: string, handler: (msg: Message) => void): () => void {
    if (!this.subscriptions.has(channel)) {
      this.subscriptions.set(channel, []);
    }
    const sub = { id: crypto.randomUUID(), handler };
    this.subscriptions.get(channel)!.push(sub);
    return () => this.unsubscribe(channel, sub.id);
  }
  
  unsubscribe(channel: string, subId: string) {
    const subs = this.subscriptions.get(channel) || [];
    const idx = subs.findIndex(s => s.id === subId);
    if (idx >= 0) subs.splice(idx, 1);
  }
}

// メッセージタイプ
interface Message {
  id: string;
  type: 'request' | 'response' | 'broadcast' | 'event';
  from: string;
  to?: string;
  content: any;
  timestamp: number;
}

4. ステートマシン

type AgentState = 'idle' | 'thinking' | 'waiting' | 'acting' | 'error';

interface AgentSession {
  id: string;
  agentId: string;
  state: AgentState;
  currentTask?: string;
  history: Turn[];
  sharedContext: Record<string, any>;
}

class StateManager {
  private sessions = new Map<string, AgentSession>();
  
  transition(sessionId: string, newState: AgentState) {
    const session = this.sessions.get(sessionId);
    if (!session) return;
    
    const oldState = session.state;
    session.state = newState;
    
    // 状態遷移フック
    this.onTransition(sessionId, oldState, newState);
  }
  
  // 状態遷移ルール
  private canTransition(from: AgentState, to: AgentState): boolean {
    const rules: Record<AgentState, AgentState[]> = {
      idle: ['thinking'],
      thinking: ['waiting', 'acting', 'error', 'idle'],
      waiting: ['thinking', 'error', 'idle'],
      acting: ['thinking', 'error', 'idle'],
      error: ['idle', 'thinking']
    };
    return rules[from]?.includes(to) || false;
  }
}

通信パターン

パターン	説明	適用シナリオ
ブロードキャスト	全Agentが受信	グローバル通知
ポイントツーポイント	指定Agentが受信	タスク割り当て
パブリッシュ/サブスクライブ	トピックごとに配信	イベント駆動
ブラックボード	共有知識スペース	協調推論

よくあるパターン

ロールプレイ

class RolePlayOrchestrator extends Orchestrator {
  async execute(goal: string) {
    // 役割の割り当て
    const planner = this.getAgent('planner');
    const executor = this.getAgent('executor');
    const critic = this.getAgent('critic');
    
    const plan = await planner.think({ description: goal });
    const result = await executor.think({ description: plan.output, context: '' });
    const review = await critic.think({ description: result.output, context: '' });
    
    return review.output;
  }
}

ディベート

async debate(topic: string, rounds = 3) {
  const pro = this.getAgent('pro');
  const con = this.getAgent('con');
  const judge = this.getAgent('judge');
  
  let context = '';
  for (let i = 0; i < rounds; i++) {
    const proArg = await pro.think({ description: `賛成論点 (第${i+1}ラウンド): ${topic}`, context });
    context += `\n賛成: ${proArg.output}`;
    
    const conArg = await con.think({ description: `反対論点 (第${i+1}ラウンド): ${topic}`, context });
    context += `\n反対: ${conArg.output}`;
  }
  
  return judge.think({ description: `判定: ${topic}`, context });
}

ベストプラクティス

1. 単一責任: 各Agentは明確な専門分野を持つ
2. 疎結合: メッセージバスを介して通信し、直接依存を避ける
3. タイムアウト制御: Agentのフリーズを防ぐ
4. サーキットブレーカー: 障害が多発した場合に自動的に機能を縮退させる
5. 可観測性: 完全なログとトレース

使い方

1. スキルをインストールする
2. 必要に応じて設定する
3. OpenClawで実行する