Agent Workflow Builder

エージェントワークフロービルダー

エージェントワークフロービルダーのスキルは、複雑なタスクを達成するために、計画を立て、推論し、ツールを使い、連携できるマルチエージェントAIシステムの設計と実装を支援します。現代のAIアプリケーションでは、LLMが単にテキストを生成するのではなく、アクションを調整する推論エンジンとして機能する、エージェント型アーキテクチャへの依存が高まっています。

このスキルでは、エージェントの設計パターン、ツール統合、状態管理、エラー処理、ヒューマン・イン・ザ・ループのワークフローについて説明します。安全性と制御性を維持しながら、現実世界の複雑さに対応できる堅牢なエージェントシステムの構築を支援します。

自律型アシスタント、ワークフロー自動化、複雑な推論システムのいずれを構築する場合でも、このスキルを使えば、エージェントアーキテクチャを適切に設計し、本番環境に対応できる状態にできます。

主なワークフロー

ワークフロー 1：エージェントアーキテクチャを設計する

1. エージェントの範囲を定義します。
   • どのタスクを自律的に処理する必要がありますか？
   • 人間の承認が必要なのはどのタスクですか？
   • 明示的に対象外とする範囲は何ですか？
2. アーキテクチャパターンを選択します。

   パターン	説明	適用場面
   シングルエージェント	1つのLLMとツール	単純なタスク、明確な範囲
   ルーターエージェント	分類して委任	複数の異なる分野
   シーケンシャルチェーン	順番に実行するエージェント	パイプライン処理
   階層型	マネージャーとワーカーエージェント	複雑で、分解可能なタスク
   協調型	議論するピアエージェント	多様な専門知識が必要

3. ツールセットを設計します。
   • エージェントにはどのような機能が必要ですか？
   • ツールはどのように定義し、ドキュメント化しますか？
   • 安全性の境界はどこですか？
4. 状態管理を計画します。
   • 会話履歴
   • タスクの状態と進捗
   • 外部システムの状態
5. アーキテクチャ上の決定を文書化します。

ワークフロー 2：エージェントループを実装する

1. エージェントのコアループを構築します。

   class Agent:
       def __init__(self, llm, tools, system_prompt):
           self.llm = llm
           self.tools = {t.name: t for t in tools}
           self.system_prompt = system_prompt
   
       async def run(self, user_input, max_steps=10):
           messages = [
               {"role": "system", "content": self.system_prompt},
               {"role": "user", "content": user_input}
           ]
   
           for step in range(max_steps):
               response = await self.llm.chat(messages, tools=self.tools)
   
               if response.tool_calls:
                   # Execute tools
                   for call in response.tool_calls:
                       result = await self.execute_tool(call)
                       messages.append({"role": "tool", "content": result})
               else:
                   # Final response
                   return response.content
   
           raise MaxStepsExceeded()
   
       async def execute_tool(self, call):
           tool = self.tools[call.name]
           return await tool.execute(call.arguments)
   

2. 明確なインターフェースを持つツールを実装します。
3. エラー処理と再試行を追加します。
4. ログ記録と可観測性を組み込みます。
5. 多様なシナリオでテストします。

ワークフロー 3：マルチエージェントシステムを構築する

1. エージェントの役割を定義します。

   agents = {
       "planner": Agent(
           llm=gpt4,
           tools=[search, create_task],
           system_prompt="You decompose complex tasks into steps..."
       ),
       "researcher": Agent(
           llm=claude,
           tools=[web_search, read_document],
           system_prompt="You gather and synthesize information..."
       ),
       "executor": Agent(
           llm=gpt4,
           tools=[code_interpreter, file_system],
           system_prompt="You execute tasks and produce outputs..."
       ),
       "reviewer": Agent(
           llm=claude,
           tools=[validate, provide_feedback],
           system_prompt="You review work for quality and correctness..."
       )
   }
   

2. オーケストレーションを実装します。
   • エージェント同士はどのように通信しますか？
   • どの処理をいつ実行するかは、誰が決定しますか？
   • エージェント間で作業をどのように引き渡しますか？
3. 共有状態を管理します。
   • タスクボードまたは作業キュー
   • 共有メモリまたはコンテキスト
   • 成果物の保存先
4. 障害に適切に対処します。
5. 必要な箇所に人間によるチェックポイントを追加します。

クイックリファレンス

アクション	コマンド／トリガー
エージェントを設計する	「[task] 用のエージェントを設計」
ツールを追加する	「[agent type] 用のツールは？」
マルチエージェントを構築する	「[goal] のためのマルチエージェントシステムを構築」
エラーに対処する	「エージェントのエラー処理パターン」
ヒューマン・イン・ザ・ループを追加する	「エージェントワークフローに人間の承認を追加」
エージェントをデバッグする	「エージェントワークフローをデバッグ」

ベストプラクティス

• シンプルに始める：マルチエージェントにする前に、ツールを使う単一エージェントから始めます

  • 最小限の複雑さで価値を実証します
  • 必要な場合にのみエージェントを追加します
  • 各エージェントには、明確で固有の責任を持たせます

• ツールを慎重に設計する：ツールはエージェントの手です

  • 明確で説明的な名前とドキュメントを付けます
  • 入出力スキーマを明確に定義します
  • 適切なエラー処理とメッセージを用意します
  • 可能な場合は、冪等性のある操作にします

• エージェントの自律性を制限する：影響範囲を抑えます

  • エージェントが実行できない操作を定義します
  • 影響の大きい操作には承認を必須にします
  • 支出やレートの上限を設定します
  • 監査用にすべての操作を記録します

• 状態を明示的に管理する：LLM のメモリーだけに頼らないようにします

  • 会話とタスクの状態を永続化します
  • 長いコンテキストを要約して、コンテキストウィンドウに収めます
  • 試したこと、完了したことを追跡します

• 適切に失敗させる：エージェントはエラーに遭遇します

  • エージェントが状況を推論できる明確なエラーメッセージを用意します
  • バックオフ付きの再試行ロジックを実装します
  • フォールバック戦略を用意します
  • 人間にエスカレーションする経路を用意します

• すべてを観測する：エージェントのデバッグは困難です

  • すべての LLM 呼び出しとツールの実行を記録します
  • 推論の連鎖と判断を追跡します
  • タスクの種類ごとに成功率を測定します

高度なテクニック

ReAct パターン（推論＋行動）

エージェントの思考を明示的に構成します：

REACT_PROMPT = """
You are an agent that solves tasks step by step.

For each step:
1. Thought: Analyze the current situation and decide what to do
2. Action: Choose a tool and provide arguments
3. Observation: Review the tool result

Continue until you can provide a final answer.

Available tools: {tool_descriptions}

Current task: {task}

Begin:
"""

タスク分解を行う計画エージェント

複雑なタスクを、取り組みやすいステップに分解します：

class PlanningAgent:
    async def solve(self, task):
        # Step 1: Create plan
        plan = await self.create_plan(task)

        # Step 2: Execute each step
        results = []
        for step in plan.steps:
            result = await self.execute_step(step, context=results)
            results.append(result)

            # Replan if needed
            if result.status == "blocked":
                plan = await self.replan(task, results)

        # Step 3: Synthesize final output
        return await self.synthesize(task, results)

振り返りと自己修正

エージェントが自身の作業を確認し、改善できるようにします：

async def solve_with_reflection(self, task, max_attempts=3):
    for attempt in range(max_attempts):
        # Generate solution
        solution = await self.generate_solution(task)

        # Self-critique
        critique = await self.critique_solution(task, solution)

        if critique.is_acceptable:
            return solution

        # Improve based on critique
        task = f"{task}\n\nPrevious attempt issues: {critique.issues}"

    return solution  # Return best effort

ヒューマン・イン・ザ・ループのチェックポイント

ワークフローに人間の承認を組み込みます：

class HumanApprovalTool:
    async def execute(self, action_description, risk_level):
        if risk_level == "low":
            return {"approved": True, "auto": True}

        # Send to approval queue
        approval_request = await self.create_request(action_description)

        # Wait for human response (with timeout)
        response = await self.wait_for_approval(
            approval_request.id,
            timeout_minutes=30
        )

        return {
            "approved": response.approved,
            "feedback": response.feedback,
            "auto": False
        }

メモリー管理

長い会話やコンテキストに対処します：

class AgentMemory:
    def __init__(self, max_tokens=8000):
        self.max_tokens = max_tokens
        self.messages = []
        self.summaries = []

    def add(self, message):
        self.messages.append(message)

        if self.token_count() > self.max_tokens:
            self.compress()

    def compress(self):
        # Summarize older messages
        old_messages = self.messages[:-5]  # Keep recent
        summary = summarize(old_messages)

        self.summaries.append(summary)
        self.messages = self.messages[-5:]

    def get_context(self):
        return {
            "summaries": self.summaries,
            "recent_messages": self.messages
        }

避けるべきよくある落とし穴

• 単一のエージェントで十分なのに、マルチエージェントシステムを構築する
• 安全上の制約なしに、エージェントに過度の自律性を与える
• ツールの失敗やエッジケースに対処しない
• LLM がツール呼び出しを幻覚する可能性を忘れる
• エージェントが行き詰まったときに無限ループに陥る
• エージェントの挙動をデバッグするためのログが不足する
• エージェントが指示に完全に従うと思い込む
• エージェントのループにおけるコスト（トークン使用量）を無視する