信頼性の高い AI エージェンツを構築するためのプロダクション・アーキテクチャ

基本ループとシステム的進化

単独のパイロットであれば良好な整備状況でも、航空作戦を実行することはありません。実際の運用では、パイロットは以下の要素によって構造的に囲まれ、システム全体として高速・安全・デバッグ可能かつ測定可能に保たれます。

• 任務選定チーム: 離陸前にどの任務（ソルティーズ）を飛行させるか決める。
• 並行運用航空団: 独立したソルティーズを飛行する。
• 燃料予算とピンゴコール: タンク枯渇前に基地への帰還を強制する。
• フライトレコーダー: 事後検証を可能にする全記録。
• 作戦後レビュー: 任務の実際の成功有無を判定する。

Claude Code、Devin、Cursor、Hermes といったプロダクション用エージェンツは、まさにこの基本ループに対し同様の構造を適用しています。

私たちのアプローチ：合成（Composition）

単一の LLM 呼び出しを、計画・実行・回復・正しさを証明する信頼性の高いシステムへと変えるための鍵は「合成」です。

• タイプ付けされたツール: Pydantic による検証。
• 計画用 DAG: 依存関係グラフと並列実行。
• 階層型メモリ: リトライ予算の下のマルチティア構成。
• 検証ヒアークジー: 不良な出力の無言伝播を防止。
• 予算管理: 機敏な劣化機構を備えた多次元的アプローチ。
• トレーサー: システム全体の可視化。

安直なエージェンツは特定の予測可能な形で失敗するため、各プリミティブが存在する理由は明確です。LLM は無効なツール引数を発明するため、タイプ付けされたツールを導入します。すべてが順次実行されるため、依存関係グラフと並列実行を追加します。

事例：都市比較エージェンツ

本記事を通じて構築する事例は、与えられた都市リストに対し、人口・タイムゾーン・各都市への短い物語的サマリーに基づいた比較レポートを生成する「都市比較エージェンツ」です。

タスクの特性と制約

このタスクは一見単純ですが、慎重に選ばれたものです。

• 並列実行構造: 各都市属性の照会は互いに独立しており、3 都市のリクエストは自然に9 回のツール呼び出しに分解されます。
• 順序依存構造: 最終報告書はすべてのルックアップが完了した後にのみ生成可能です。
• 検証可能性: プログラム的に各都市のレポート包含を検証できます。
• コスト分布:
  • 人口/タイムゾーン：辞書読み込み（安価）。
  • サマリー/集計：LLM 呼び出し（現実的な予算圧力）。

再現性の確保

• モック環境: ネットワーク接続なしの

  CITY_FACTS

  から動作。
• フレキシブル実装: Anthropic モデルか決定論的モックのどちらでも動作可能。
• 最初のプリミティブ到達: これにより基盤となる構成要素を確立します。

プラグ可能な脳（Brain）

計画担当者、サマライザー、アグリゲーター、批評家はすべて LLM に呼び出されます。特定の SDK へのハードウェアワイヤーはテスト不能かつベンダーロックの原因となります。

LLM アブストラクション

他任何事情を行う前に、LLM 呼び出し API の詳細に対する抽象化ベースクラスを定義します：

class LLMProvider:
    """共有インターフェース。サブクラスして異なるバックエンドをプラグインできます。"""

    def complete(self, system: str, user: str, role: str = "default") -> str:
        raise NotImplementedError

    async def acomplete(self, system: str, user: str, role: str = "default") -> str:
        #同期的な呼び出しをスレッドにラップ；あらゆる SDK で動作します。
        return await asyncio.to_thread(self.complete, system, user, role)

• MockProvider: 決定論的で役割に応じた応答を返します（例：計画要求→標準的計画、サマライズ→テンプレート化）。
• 目的: 「オーケストレーションが間違っているのか」かを「モデル自体の問題か」と区別可能にし、実験の再現性を確保します。

typed ツールの導入

基本ハネスではツール引を手動で検証しましたが、新規ツールごとに重複ロジックが増え、LLM はスキーマを無視してエラーを生じます。

アップグレード策: 各ツールの引数を Pydantic モデルとして宣言し、一つの定義で統制します：

@dataclass
class TypedTool:
    name: str
    description: str
    args_model: type[BaseModel]     # 引数スキーマを定義する Pydantic モデル
    fn: Callable[..., Any]
    cost_hint: float = 0.0          # 予算計算のための相対コスト

    def schema(self) -> dict:
        """Anthropic/OpenAI ツール使用 API が期待する形状"""
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": self.args_model.model_json_schema(),
        }

    def run(self, raw_args: dict) -> Any:
        args, err = self.validate_args(raw_args)
        if err is not None:
            raise ValueError(err)
        return self.fn(**args.model_dump())

メリット:

• 実行時検証による高価なツール呼び出しの回避。
• 潜在的な副作用を持つ呼び出しを早期に失敗させる。
• ドキュメント（

  Field(..., description=...)

  ）は計画担当者が読み取るカタログの一部となる。

コストティア管理

レジストリには、4 つのツールが登録されています：

cost_hint

get_population

get_timezone

0.1

summarize_city

1.0

aggregate_report

2.0

LLM も単なるツールの一種であり、内部モデルのキャッシュやレート制限の切り替えが可能です。

計画はグラフである（Plan is a Graph）

基本ハネスではターンごとに一つのアクションを実行していましたが、依存関係を明示する**有向非循環グラフ（DAG）**が優れています。LLM に単一のアクションを要求せず、全体のグラフを取得するように求めます。

グラフの検証

計画担当者（LLM）は「ハルシネーションした構造」を発明し得るため、実行前に計画を検証します：

def ready_nodes(self) -> list[PlanNode]:
    """依存関係がすべて DONE であり、かつ自らは PENDING のノード"""
    out = []
    for n in self.nodes.values():
        if n.status != NodeStatus.PENDING:
            continue
        if all(self.nodes[d].status == NodeStatus.DONE for d in n.deps):
            out.append(n)
    return out

• 役割: スケジューラーの中心。依存関係を満たしたノードセットを返します。
• 例: 3 都市の場合、10 つのノード（9 つの fetch + 1 つの

  aggregate_report

  ）を出力します。

グラフの並列実行

エグゼクテュアはレベル同期型の DAG ウォーカーです。準備セットを計算し、

asyncio.gather

MAX_CONCURRENT = 5  # 並列なツール/LLM 呼び出しの上限

async def execute_dag(dag, tools, on_step=None):
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)

    async def run_node(node):
        node.status = NodeStatus.RUNNING
        async with semaphore:
            try:
                # 同期的なツール関数はスレッドプールで実行
                node.result = await asyncio.to_thread(tools[node.tool].run, node.args)
                node.status = NodeStatus.DONE
            except Exception as exc:
                node.error = f"{type(exc).__name__}: {exc}"
                node.status = NodeStatus.FAILED

    while not dag.is_done():
        ready = dag.ready_nodes()
        if not ready:
            break  # 残りのノードは失敗した祖先に依存している
        await asyncio.gather(*(run_node(n) for n in ready))

重要な設計決断

1. スレッドプールの実行:

   asyncio.to_thread

   を使用し、ツール関数を

   async def

   に改変する必要がありません。
2. セマフォによる制限: 同時呼び出し数（例：50 ノード）を制限し、レート制限やコストスパイクを防ぎます。

適切なことを覚えている：階層型メモリ

安直なエージェンツは全てをプロンプトにダンプしますが、関連性のないテキストはモデルの性能低下を招きます。プロダクションエージェンツは認知科学に基づいた階層型メモリを使用します。

メモリの種類

コンテキストの構築（厳格な予算下）

def build_context(working: WorkingMemory, store: MemoryStore, budget_chars: int = 4000) -> str:
    """ワーキングメモリ＋取得されたメモリを組み立てる（文字数予算を尊重）"""
    pieces = [working.to_prompt()]
    used = len(pieces[0])

    # エピソード的ファースト（過去の類似タスク）、次にセマンティック（事実）
    for kind in ("episodic", "semantic"):
        for m in store.retrieve(working.goal, k=3, kind=kind):
            snippet = f"[{kind}] {m.content}"
            if used + len(snippet) + 1 > budget_chars:
                return "\n".join(pieces) + "\n(...truncated at budget...)"
            pieces.append(snippet)
            used += len(snippet) + 1

    return "\n".join(pieces)

優先順位: エピソード的メモリがセマンティック的メモリに優先されます。過去の失敗の方が一般的な事実よりも実行可能性が高いからです。コンテキストは受動的に蓄積するのではなく、能動的に組み立てられるべきです。

類似度関数の選定

• Jaccard 類似度: 無料だが、パラフレーズに対して失敗します。
• 文埋め込み (all-MiniLM-L6-v2): 384 次元ベクトル生成。パラフレーズを近傍へマッピング。
• ハイブリッドアプローチ:

  MemoryStore

  は Jaccard をバックアップとして、埋め込みファーストを試みます。

信頼と検証：ヒアークジー構造

エージェンツは自信に満ちたが間違った出力を生成します。検証なしではエラーに気づきません。すべてのチェックのコストは異なり、安価なティアから実行し、生き残ったもののみを昇進させるのが正解です。

def verify_report(report, goal, required_cities, provider) -> Verdict:
    det = deterministic_check_report(report, required_cities)
    if not det.passed:
        return det                # 安価なティアで検出 — LLM を煩わさない
    # 決定論的がパス → 主観的な品質のために LLM ジャッジへ昇進
    return llm_judge_report(report, goal, provider)

2 テアーゲートの重要性:

1. 安価なフィルターファースト: データ構造チェック（ゼロトークン）。
2. 高価なジャッジは生き残ったもののみ: LLM による主観的評価。
3. 役割の分離: Worker は生成し、Critic は評価します。

チームとの出会い：Planer, Worker, Critic

単一のプロンプトで全てを行うと目標が混乱します（例：計画制約がサマライズに滲み出す）。作業を狭いエージェントに分割します。

• Planer: 目標とツールスキーマを受け取り、検証済みの DAG JSON を返す。
• Worker: DAG を受け取り、単に実行する（予算課税付き）。
• Critic: 完了したレポートを受け取り判断を下す。

Planer のシステムプロンプト: ライブのツールカタログを組み込み、実際に存在するツールのみを参照可能です：

PLANNER_SYSTEM = """あなたは Planer エージェントです。GOAL が与えられたとき、それを実現するためのツール呼び出しの依存関係グラフを生成してください。

出力は ONLY JSON にしてください（プロースやマークダウンなし）：

{"nodes": [{"id": "...", "tool": "...", "args": {...}, "deps": [...]}, ...]}

`deps` が空または既遂足されている場合、ノードは並列で実行できます。
最終的な `aggregate_report` ノードはすべての上位フェッチ/サマリーノードに依存する必要があります。
そのクロニカルノードには"id: 'aggregate'"を推奨しますが（一意の ID も可）、利用可能なツール（名前とスキーマ）：

<<TOOLS_JSON>>
"""

燃料計と失敗モード：機敏な劣化機構

基本ハネスは単一の

max_steps

BudgetMulti

圧力（Pressure）計算

def pressure(self) -> float:
    return max(
        self.tokens_used / self.max_tokens,
        self.tool_calls_used / self.max_tool_calls,
        self.elapsed() / self.max_wall_seconds,
        self.cost_usd / self.max_cost_usd,
    )

圧力感応劣化:

• 0.7 以下: フルパイプライン実行（LLM ジャッジ含む）。
• 0.9 以上: オーケストレーターは高価な批評家をスキップ。決定論的チェックのみ使用。
• 1.0: 部分的な結果と共に実行停止。

エラー分類と回復

すべてのエラーに対して同一の対応が必要です。

classify_error()

1. 一時的エラー（レート制限等）: ジッター付き指数バックオフで再試行。
2. 検証エラー（ツールの誤用）: 構造化されたエラーをフィードし、LLM に修正させる。
3. 情報不足: 再試行は有害（幻覚化の繰り返し）。再計画が必要。
4. ポリシー違反: 致命的。即時停止。

フライトレコーダー：真の可観測性

エージェンツが失敗した際、「何が起こったか」を答えるために専用構造化イベントログが必要です。

Tracer

• アイデンティティ: ステップ ID と親 ID による親子関係。
• セマンティクス: 役割と実施されたアクション。
• 経済学: 遅延、トークン、コスト、予算圧力のスナップショット。
• 判断: 批評家の評価記録。

出力可能な形式:

• ファイルへのダンプ
• OpenTelemetry / LangSmith への送付

• matplotlib

  でのプロット
• JSON シリアライズ

ステップごとの遅延や予算圧力の増大は、並列実行の重要性や、LLM ジャッジがスキップされた理由を理解するために不可欠です。

すべてを組み合わせる：再計画ループ

プリミティブが設置されると、オーケストレータの実装はシンプルになります。コンテキスト構築 → DAG 要求 → 実行 → チェック → 保存 というフローです。

唯一の新しい振る舞い：情報に基づく再計画

for attempt in range(self._max_replans + 1):
    dag = self.planner.plan(goal) if attempt == 0 else self.planner.replan(goal, dag)
    await self.worker.execute(dag, on_step=on_step)

    if dag.any_failed():
        failed = (n for n in dag.nodes.values() if n.status == NodeStatus.FAILED)
        if any(classify_error(n.error or "") == ErrorClass.MISSING_INFO for n in failed) \
                and attempt < self._max_replans and budget.has_room():
            continue          # 情報に基づいた再計画、盲目的な再試行ではない
        return RunResult(..., status="failed_execute")
    break                     # 成功

# 圧力感応劣化：予算が厳しい場合 LLM ジャッジをスキップ
if budget.pressure() > 0.9:
    verdict = deterministic_check_report(report, required_cities)
else:
    verdict = self.critic.judge(report, goal, required_cities)

モックと実際のモデルの違いへの対応

実際のモデルは構造を保持しつつもスタイルが多様化するため、モックの想定する

aggregate

• DAG 内で ID ではなくツール名でクロニカルノードを解決。
• Planer プロンプトに明確な指示を追加（「言葉を取らない」）。

今も欠けていることと今後の課題

カスタムハネスの構築において、ほぼ全ての主要概念をカバーしました。単なるループから、7 つのプロダクション形状プリミティブへ進化させました。

合成可能性（Compositionability）

これらの部品は互いを必要としませんが、合成可能です：

• ツールを追加すると Planer が自動的にスキーマを見る。

• verify_report()

  を厳格にすると基準が更新される。
• LLM バックエンド全体を交換してもハネスは変更なし。

注意点（Production Readiness）

1. 永続化: 記憶は今現在プロセス内だが、埋め込みを Chroma / Weaviate / pgvector に保存すべき。
2. サンドボックス化: ツール出力は指示として信頼するが、プロンプトインジェクション防止のためデータとして隔離すべき。
3. 承認フロー: 不可逆的なアクションは人間の承認が必要。
4. 会計の精度: トークン推定ではなく、SDK からメタデータを取得する方が正確。

意図的な省略

単一の成功デモは機能を示しますが、大多数のケースでの安定性は「評価ハネス」の仕事であり、次回のフォローアップ記事で論じます。