高度なエージェントハーネスの構築

基本的なハーネスのループは正しいですが、単純です。よくできたジェット機に乗った一人のパイロットはドッグファイトに勝てるかもしれませんが、誰もそのように航空作戦を実行しません。実際の作戦では、誰かが離陸する前にどの出撃を飛ばすかを決定するミッションプランナー、並行して独立した出撃を飛ばす飛行隊、燃料が尽きる前に基地への帰還を強制する燃料予算とビンゴコール、事後にすべてのミッションを再構築可能にするフライトレコーダー、そしてミッションが実際に成功したかどうかを決定する事後レビューが追加されます。これらのどれもパイロットに取って代わるものではありません。それらはパイロットを構造で囲み、システム全体を高速、安全、デバッグ可能、測定可能に保ちます。Claude Code、Devin、Cursor、Hermes、その他のプロダクションエージェントは、基本的なループに対してまったく同じことを行います。この記事では、フレームワークの背後にあるメカニズムを隠すことなく、基本的なハーネスのすべての部分をそのプロダクション形状にアップグレードします。この演習全体を導く質問は単純です。単一のLLM呼び出しを、計画、実行、回復、そしてそれが正しいことを行ったと証明できる信頼性の高いシステムにどうやって変えるか？私たちの答えはコンポジションです。私たちは小さくテスト可能なプリミティブを構築します。型付きツール、プランDAG、階層メモリ、検証階層、予算、トレーサー、そしてそれらを意図的に薄いオーケストレーターで配線します。各プリミティブは、単純なエージェントが特定の予測可能な方法で失敗するからです。LLMは無効なツール引数を発明するので、Pydantic検証付きの型付きツールを追加します。すべてが逐次実行されるので、依存関係グラフと並列実行を追加します。コンテキストウィンドウがジャンクでいっぱいになるので、検索予算の下でマルチティアメモリを追加します。悪い出力が静かに伝播するので、検証階層を追加します。1つのプロンプトがすべてをやろうとするので、それをプランナー、ワーカー、クリティックの役割に分割します。コストが暴走するので、段階的な低下を伴う多次元予算編成を追加します。ハーネスが通常機能することを証明することは、評価スイート、検索ベンチマーク、および特殊なワーカープールで、将来の記事で完全に扱われます。実行中の例投稿全体を通して、都市比較エージェントを構築します。都市のリストが与えられると、人口、タイムゾーン、および各都市の短い物語の要約でそれらを比較するレポートを生成します。タスクはほとんど侮辱的に単純に見えますが、慎重に選択されました。各都市属性の検索は、他のどの検索とも独立しています。これは、3つの都市のリクエストが自然に9つのツール呼び出しに分解され、それらすべてを同時に実行できることを意味します。一方で、最終レポートは、すべての検索が最初に完了することに依存しているため、私たちは平坦なステップリストを超えています。要求されたすべての都市が実際にレポートに表示されることをプログラムで確認して、結果を検証できます。そして、ツールは非常に異なるコストを持っています。人口とタイムゾーンの検索はインメモリ辞書読み取りですが、都市ごとの要約と最終集計はそれぞれLLMを呼び出します。これにより、管理すべき現実的な予算の圧力がかかります。再現性のために、検索ツールは小さなモックされた辞書CITY_FACTSから読み取るため、ノートブックはネットワークアクセスなしで完全に再現可能です。LLMバックエンドのコンポーネントは、実際のAnthropicモデルまたは決定論的なモックに対して実行できます。これが最初のプリミティブです。プラグ可能な脳構築するすべてのコンポーネントは最終的にLLMを呼び出します。プランナー、サマライザー、アグリゲーター、クリティック。その呼び出しが1つのSDKにハードワイヤリングされている場合、ハーネス全体はテスト不可能になり、ベンダーロックインされます。したがって、まず、さまざまな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も実装しており、決定論的でロールを認識した応答を返します。計画を求められたら標準的な計画、要約を求められたらテンプレート化された1行の要約、判断を求められたらルールベースの合格/不合格の判定を返します。これにより、開発中に「オーケストレーションが間違っているのか？」と「モデルの計画が悪いのか？」を分離できます。そして、この記事のすべての実験がどのマシンでも再現可能である理由です。型付きツール基本的なハーネスではツール引数を手動で検証していましたが、このアプローチはすぐに破綻します。新しいツールごとに検証ロジックが重複し、LLMは正式なスキーマを見ることができず、引数の形状を推測するだけで、結果のエラーはモデルが自己修正できないアドホックな文字列になります。アップグレードは、各ツールの引数をPydanticモデルとして宣言し、1つの定義がすべてを駆動するようにすることです。@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()) このアプローチにより、実行時検証、AnthropicおよびOpenAIのツール使用APIが期待する正確な形状のJSONスキーマ、ドキュメント（各Field(…, description =…)はプランナーが読み取るカタログの一部になります）、およびcost_hintを介したコスト会計のフックが得られます。実行前に失敗することで、潜在的な副作用のある高価なツール呼び出しを回避できます。悪い計画は、検証レイヤーで、データベースクエリの奥深くではなく、速やかに失敗する必要があります。このアプローチは、LangChainツール、Anthropicツール使用、およびOpenAI関数呼び出しのようなフル機能のフレームワークが収束するものと似ています。私たちのレジストリには、3つのコストティアを持つ4つのツールがあります。get_populationとget_timezone()は基本的に無料の辞書ルックアップ（cost_hint =0.1）であり、summarize_city()は都市ごとに1回のLLM呼び出し（cost_hint =1.0）を行い、aggregate_report()は最終的なMarkdownを生成するトークンを大量に消費する合成呼び出し（cost_hint =2.0）を行います。最後の2つは内部でLLMを呼び出すツールであることに注意してください。LLMは他のツールと同じです。ワーカーは統一されたツールインターフェースを見ますが、一部のツールはサブプロンプトのラッパーであるため、ハーネスとは独立して内部モデルをキャッシュ、レート制限、またはスワップできます。プランはグラフです基本的なハーネスは1ターンに1つのアクションを実行しました。ステップが厳密に逐次的である場合は機能しますが、私たちのタスクには、単一の集計に供給する9つの独立した検索があります。whileループはこれらを1つずつ実行します。有向非巡回グラフ（DAG）は依存関係を明示的に表現し、実行可能なものはすべてすぐに並行して実行できるスケジューラを可能にします。したがって、LLMに一度に1つのアクションを要求する代わりに、事前にグラフ全体をプランナーに要求します。LLMは実行する前に構造を宣言します。プランナーはLLMであるため、構造を幻覚することもできます。存在しないノードIDへの依存関係や、決して完了できない循環依存関係などです。そのため、プランに対して最初に行うことは、壊れたプランを実行しようとしてトークンを無駄にする前に、それを検証することです。def ready_nodes(self) -> list[PlanNode]: """依存関係がすべて完了しており、それ自体が保留中のノード。""" 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 ready_nodes() はスケジューラの心臓部です。いつでも、依存関係がすべて満たされているノードのセットを返します。3つの都市の目標では、プランナーは10個のノードを出力します。依存関係リストが空の9つのフェッチ、すべて並行して実行可能であり、すべての依存関係を持つ1つのaggregate_reportキャップストーンです。