AIエージェント入門｜メモリ・無限ループ・調整の3つの課題と対策

ひとことで言うと

AIエージェント（生成AIに自動判断させる仕組み）は、データを覚えておく「メモリ」、複数のエージェント間での「調整」、自動実行時の「無限ループ」という3つの課題があります。特に初心者の多くが「メモリが足りない」と思い込みますが、実は「エージェント同士がうまく協力できていない」という別の問題であることが多いです。この記事では、それぞれの課題の原因と対策を、実装に役立つレベルで解説します。

AIエージェントとは何か

生成AI（大規模言語モデルのこと）に「自動で判断と実行をさせる」技術がAIエージェントです。例えば、「ユーザーからの質問に対し、必要なデータを探して、それを分析して、最終的な答えを返す」という一連の流れを、人間が一段階ずつ指示するのではなく、AIが自分で判断して実行する形です。

エージェントは以下の流れで動きます。

1. ユーザーからの指示を受け取る
2. 現在の状況を確認する（どんなデータがあるか、何をしたか、何をするべきか）
3. 次のステップを決める（検索する、計算する、ほかのツールを呼び出す、など）
4. ステップを実行する
5. 結果を確認して、ステップ3に戻る（または完了）

この「3→4→5→3…」のサイクルが、自動で何度も繰り返される仕組みです。

課題1：メモリ問題は実は「調整問題」かもしれない

多くの開発者が最初に悩むのは「エージェントが前の会話を忘れている」「過去のデータを参照できない」というメモリ不足の問題だと思うことです。しかし、実際の根本原因は異なります。

複数のエージェントが動く場合（例：検索エージェント、分析エージェント、回答エージェントが別々に動く）、それぞれのエージェントが「誰が何をしたのか」「今どんな状態なのか」を正しく把握できていないことが多いです。つまり、メモリ容量の問題ではなく、エージェント間の「調整」の問題なのです。

エージェント間の調整が失敗する原因

• 指示が重複する：複数のエージェントが同じ作業をしてしまう
• 指示が矛盾する：あるエージェントの指示が別のエージェントと食い違っている
• 結果の受け渡しがうまくいかない：前のエージェントの結果が、次のエージェントに正しく渡されない
• 実行順序が不明確：どのエージェントがいつ動くべきか、事前に定まっていない

課題2：無限ループを防ぐ SafeAgent の実装

エージェントが自動的に実行を繰り返すため、うまくいかない場合に「終わらない」という状況が発生します。

# safe_agent.py
# 無限ループ・コスト暴走を防ぐ安全なエージェント基盤

import anthropic
import json
import hashlib
from pathlib import Path
from datetime import datetime
from dataclasses import dataclass, field

client = anthropic.Anthropic()
LOG_FILE = Path("agent_run.jsonl")

@dataclass
class StepLog:
    step: int
    input_hash: str    # 前回と同じ入力なら進展なし → 停止判定
    output_summary: str
    cost_usd: float
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat()[:19])

class SafeAgent:
    """
    無限ループ対策:
    1. max_steps: 最大実行回数の制限
    2. 進展チェック: 前回と同じ出力なら停止
    3. 予算制限: 累計コストが上限を超えたら停止
    """
    
    def __init__(
        self,
        model: str = "claude-haiku-4-5-20251001",
        max_steps: int = 10,
        budget_usd: float = 0.10,
    ):
        self.model = model
        self.max_steps = max_steps
        self.budget_usd = budget_usd
        self.total_cost = 0.0
        self.step_logs: list[StepLog] = []
        self.seen_hashes: set[str] = set()
        
        # モデル別料金
        self.pricing = {
            "claude-haiku-4-5-20251001": (0.80, 4.00),
            "claude-sonnet-4-5": (3.00, 15.00),
            "claude-opus-5": (15.00, 75.00),
        }
    
    def _hash_text(self, text: str) -> str:
        return hashlib.md5(text.encode()).hexdigest()[:8]
    
    def _calc_cost(self, in_tokens: int, out_tokens: int) -> float:
        in_price, out_price = self.pricing.get(self.model, (3.00, 15.00))
        return (in_tokens * in_price + out_tokens * out_price) / 1_000_000
    
    def _call_llm(self, messages: list[dict], system: str = "") -> tuple[str, float]:
        kwargs = {"model": self.model, "max_tokens": 512, "messages": messages}
        if system:
            kwargs["system"] = system
        
        resp = client.messages.create(**kwargs)
        text = resp.content[0].text
        cost = self._calc_cost(resp.usage.input_tokens, resp.usage.output_tokens)
        return text, cost
    
    def run(self, task: str) -> str:
        system = """タスクを段階的に実行してください。
各ステップで:
1. 現在の判断と実行内容を説明する
2. タスクが完了したら「DONE: [最終結果]」と出力する
3. 完了できない場合は「BLOCKED: [理由]」と出力する"""
        
        messages = [{"role": "user", "content": f"タスク: {task}"}]
        print(f"エージェント開始 (最大{self.max_steps}ステップ, 予算${self.budget_usd:.2f})")
        
        for step in range(1, self.max_steps + 1):
            # 予算チェック
            if self.total_cost >= self.budget_usd:
                print(f"予算上限到達: ${self.total_cost:.4f} >= ${self.budget_usd:.2f}")
                return f"予算超過で停止 (${self.total_cost:.4f}使用)"
            
            output, cost = self._call_llm(messages, system)
            self.total_cost += cost
            
            # 進展チェック
            output_hash = self._hash_text(output)
            if output_hash in self.seen_hashes:
                print(f"Step {step}: 同じ出力を繰り返しています → ループ検出、停止")
                return f"ループ検出で停止 (Step {step})"
            self.seen_hashes.add(output_hash)
            
            # ログ記録
            log = StepLog(
                step=step,
                input_hash=self._hash_text(messages[-1]["content"]),
                output_summary=output[:100],
                cost_usd=round(cost, 5)
            )
            self.step_logs.append(log)
            with open(LOG_FILE, "a") as f:
                f.write(json.dumps({"step": log.step, "cost": log.cost_usd, "summary": log.output_summary}) + "\n")
            
            print(f"Step {step}/{self.max_steps} (${self.total_cost:.4f}累計): {output[:80]}...")
            
            # 完了・ブロック判定
            if output.startswith("DONE:"):
                print(f"\n完了 ({step}ステップ, ${self.total_cost:.4f})")
                return output[5:].strip()
            elif output.startswith("BLOCKED:"):
                print(f"\nブロック: {output[8:]}")
                return output
            
            messages.append({"role": "assistant", "content": output})
            messages.append({"role": "user", "content": "続けてください"})
        
        print(f"\n最大ステップ数({self.max_steps})に到達")
        return f"ステップ上限で停止 (${self.total_cost:.4f}使用)"

# 使用例
agent = SafeAgent(model="claude-haiku-4-5-20251001", max_steps=5, budget_usd=0.05)
result = agent.run("Pythonで簡単な計算機CLIを設計してください")
print(f"\n結果: {result}")

課題3：中央管理者パターンでマルチエージェントを整理する

複数エージェントの問題は「直接通信させすぎること」です。中央管理者（オーケストレーター）パターンが解決策です。

# orchestrator.py
# 中央管理者が各エージェントを順番に調整するパターン

import anthropic
from dataclasses import dataclass

client = anthropic.Anthropic()

@dataclass
class AgentTask:
    name: str        # エージェント名
    role: str        # 役割説明（システムプロンプト）
    model: str = "claude-haiku-4-5-20251001"

# 各エージェントを定義（シンプルな役割のみ）
AGENTS = {
    "researcher": AgentTask(
        name="researcher",
        role="あなたは調査エージェントです。与えられたトピックについて事実を調べ、箇条書きでまとめてください。",
    ),
    "analyst": AgentTask(
        name="analyst",
        role="あなたは分析エージェントです。与えられた情報を分析して、重要な洞察と結論を3点以内で述べてください。",
    ),
    "writer": AgentTask(
        name="writer",
        role="あなたはライターエージェントです。与えられた分析結果を300文字以内の読みやすい文章にまとめてください。",
    ),
}

def run_agent(agent: AgentTask, input_text: str) -> tuple[str, float]:
    """1つのエージェントを実行する"""
    resp = client.messages.create(
        model=agent.model,
        max_tokens=512,
        system=agent.role,
        messages=[{"role": "user", "content": input_text}]
    )
    
    cost = (resp.usage.input_tokens * 0.80 + resp.usage.output_tokens * 4.00) / 1_000_000
    return resp.content[0].text, cost

def orchestrate(topic: str) -> dict:
    """中央管理者として各エージェントを順番に実行する"""
    print(f"オーケストレーション開始: {topic}")
    total_cost = 0.0
    pipeline = {}
    
    # Step 1: 調査
    print("  [1/3] 調査エージェント実行中...")
    research, cost = run_agent(AGENTS["researcher"], f"トピック: {topic}")
    total_cost += cost
    pipeline["research"] = research
    
    # Step 2: 分析（調査結果を入力）
    print("  [2/3] 分析エージェント実行中...")
    analysis, cost = run_agent(AGENTS["analyst"], f"調査結果:\n{research}")
    total_cost += cost
    pipeline["analysis"] = analysis
    
    # Step 3: 執筆（分析結果を入力）
    print("  [3/3] ライターエージェント実行中...")
    article, cost = run_agent(AGENTS["writer"], f"分析結果:\n{analysis}")
    total_cost += cost
    pipeline["article"] = article
    
    print(f"完了 (合計コスト: ${total_cost:.5f})")
    return {"topic": topic, "pipeline": pipeline, "total_cost_usd": round(total_cost, 5)}

# 実行例
result = orchestrate("2026年のAI規制の動向")
print(f"\n最終記事:\n{result['pipeline']['article']}")
print(f"総コスト: ${result['total_cost_usd']:.5f}")
# → 3エージェント合計でも $0.003 以下（Haiku使用時）

実装チェックリスト

AIエージェントを本番で使う前に確認すべき項目です。

✅ max_steps を設定した（10ステップ以下から始める）
✅ 予算上限を設定した（$0.10 から始める）
✅ 各ステップのログをファイルに記録している
✅ 「進展なし」の検出ロジックがある（同じ出力→停止）
✅ エージェントが直接互いに通信しない設計になっている
✅ 中央管理者がデータの受け渡しを担当している
✅ 本番実行前にdry_runモードでテストした

あわせて読みたい

• AIエージェントとAIアシスタントの違い【図解比較】
• AIエージェント暴走コスト防止ガイド
• AIエージェントが会話を忘れる原因と解決方法