はじめに

最近、Zenn Bookで「AI実践ガイド2026：エージェントからローカルLLMまで」という技術書を執筆しました。その過程で、AIエージェント開発における多くの失敗と成功を経験しました。

本記事では、執筆を通じて得た実践的なベストプラクティスを、具体的なコード例とともに共有します。

:::message 本記事で紹介する内容は、実際のプロダクション環境で検証済みのパターンです。 :::

対象読者

LLM APIを使った開発経験がある方
AIエージェントの実装に興味がある方
プロダクション環境での運用を考えている方

1. ReActパターンの正しい実装

AIエージェントの基本パターンである「ReAct（Reasoning + Acting）」ですが、実装時によくある落とし穴があります。

❌ よくある間違い

# 無限ループのリスク
while not task_completed:
    thought = agent.think()
    action = agent.act(thought)
    observation = execute(action)
    # 終了条件が曖昧

✅ 推奨実装

from typing import Optional
import time

class SafeReActAgent:
    def __init__(self, max_iterations: int = 10, timeout: int = 60):
        self.max_iterations = max_iterations
        self.timeout = timeout
        self.start_time = None

    def run(self, task: str) -> str:
        self.start_time = time.time()

        for i in range(self.max_iterations):
            # タイムアウトチェック
            if time.time() - self.start_time > self.timeout:
                raise TimeoutError(f"Agent timed out after {self.timeout}s")

            # 思考フェーズ
            thought = self.think(task)

            # 終了判定を明確に
            if self._is_task_complete(thought):
                return thought.final_answer

            # アクションフェーズ
            action = self._decide_action(thought)
            observation = self._execute(action)

            # 次のループ用にコンテキスト更新
            task = self._update_context(task, observation)

        # イテレーション上限に達した場合
        return self._generate_partial_answer()

ポイント:

明確な終了条件（イテレーション上限、タイムアウト）
段階的な判定による無限ループ防止
エラー時のグレースフルな終了

2. メモリ管理の最適化

LLMのコンテキストウィンドウは有限です。長時間稼働するエージェントでは、メモリ管理が重要になります。

✅ 推奨パターン: ハイブリッドメモリ

from collections import deque
from typing import List, Dict

class HybridMemory:
    def __init__(
        self,
        short_term_size: int = 5,  # 直近5件
        vector_db=None  # 長期記憶用
    ):
        self.short_term = deque(maxlen=short_term_size)
        self.vector_db = vector_db

    def add(self, interaction: Dict[str, str]):
        # 短期記憶に追加
        self.short_term.append(interaction)

        # 重要度が高い場合は長期記憶にも保存
        if self._is_important(interaction):
            self.vector_db.add(interaction)

    def get_context(self, query: str) -> str:
        # 短期記憶（必ず含める）
        recent = "\n".join([
            f"Q: {m['query']}\nA: {m['response']}"
            for m in self.short_term
        ])

        # 関連する長期記憶を検索
        relevant = self.vector_db.search(query, top_k=3)

        return f"Recent:\n{recent}\n\nRelevant:\n{relevant}"

メリット:

直近の会話は常に参照可能
古い重要情報も検索で取得
トークン数の最適化

3. エラーハンドリングとリトライ戦略

API呼び出しは必ず失敗する可能性があります。本番環境では堅牢なエラーハンドリングが必須です。

✅ Exponential Backoff with Jitter

import time
import random
from functools import wraps

def retry_with_backoff(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0
):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise

                    # Exponential backoff with jitter
                    delay = min(
                        base_delay * (2 ** attempt) + random.uniform(0, 1),
                        max_delay
                    )

                    print(f"Retry {attempt + 1}/{max_retries} after {delay:.2f}s")
                    time.sleep(delay)

        return wrapper
    return decorator

# 使用例
@retry_with_backoff(max_retries=3, base_delay=2.0)
def call_llm(prompt: str) -> str:
    return client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )

重要な考慮点:

**Jitter（ランダム性）**の追加で、同時リトライの競合を防ぐ
指数的増加で負荷を軽減
上限設定で無限待機を防ぐ

4. マルチエージェント協調のパターン

複数のエージェントを協調させる場合、「誰が何を担当するか」を明確にする必要があります。

✅ Role-Based Collaboration

from enum import Enum
from dataclasses import dataclass

class AgentRole(Enum):
    RESEARCHER = "research"
    WRITER = "write"
    CRITIC = "review"

@dataclass
class Task:
    content: str
    assigned_to: AgentRole
    depends_on: Optional[str] = None

class MultiAgentOrchestrator:
    def __init__(self, agents: Dict[AgentRole, Agent]):
        self.agents = agents
        self.task_results = {}

    def execute_workflow(self, tasks: List[Task]) -> str:
        # 依存関係を考慮して実行
        for task in self._topological_sort(tasks):
            agent = self.agents[task.assigned_to]

            # 前のタスクの結果を注入
            context = self._build_context(task)

            result = agent.run(task.content, context=context)
            self.task_results[task.id] = result

        return self._synthesize_results()

ベストプラクティス:

各エージェントの責務を明確に
依存関係を明示的に管理
結果の統合方法を定義

5. プロンプトテンプレート管理

プロンプトはコードと同様にバージョン管理すべきです。

✅ テンプレート管理システム

from jinja2 import Template
from pathlib import Path
import yaml
import SummarySlides from "@/components/ui/SummarySlides";

class PromptManager:
    def __init__(self, template_dir: Path):
        self.templates = self._load_templates(template_dir)

    def _load_templates(self, dir: Path) -> Dict[str, Template]:
        templates = {}
        for file in dir.glob("*.yaml"):
            with open(file) as f:
                config = yaml.safe_load(f)
                templates[config['name']] = Template(config['template'])
        return templates

    def render(self, name: str, **kwargs) -> str:
        if name not in self.templates:
            raise ValueError(f"Template {name} not found")

        return self.templates[name].render(**kwargs)

# templates/code_review.yaml
"""
name: code_review
version: "1.2"
description: "コードレビュー用プロンプト"
template: |
  あなたは経験豊富な [[ language ]] エンジニアです。

  以下のコードをレビューしてください：

  ``` [[ language ]]
  [[ code ]]
  ```

以下の観点で評価してください：
[[ for criterion in criteria ]]

- [[ criterion ]]
[[ endfor ]]
"""