Zenn
🔨

A2Aネイティブな宣言型ワークフロービルダー

に公開
LLM
A2A
tech

A2A はマルチエージェント時代のプロトコルと捉えられがちですが、今回はそのA2Aの仕組みを使ってワークフローっぽく実装することもできるよ、というのを書いてみようと思います。

長くなったので、ざっくり3行でまとめると：

  • ワークフローでは 「実装の変更容易性」 と 「中断・再開メカニズム」 が重要
  • LangGraph や mastra など既存のワークフローではこれらをサポートする仕組みがあるが使いづらい
  • A2Aネイティブ な仕様で 宣言的 なワークフローの定義ができるa2a-artifact-graphライブラリを作った

https://github.com/ryukez/a2a-artifact-graph

ワークフローの運用における課題

まず、ワークフローを本番運用するにあたっては以下の 2 点が重要になると考えています。

柔軟性と信頼性のトレードオフ

ワークフローの実装においてはタスクを細かく分割すればするほど信頼性を担保できる一方、実装コストが上がり AI の本来の柔軟性が損なわれてしまうというジレンマが存在します。

https://x.com/hiro_gamo/status/1913050355044389312

さらに運用を開始した後には細かい要件の変更が発生する場合が多く、一度できたワークフローも随時更新していく必要があります。タスクを細かく分割する場合、1 つのタスクを変更するとその前後（悪い場合にはワークフロー全体）に影響が波及する場合があり、実装コストや品質保証のコストの増大につながります。

このトレードオフに正解はなくその時々の状況に沿って意思決定されるべきものですが、だからこそ重要となるのはそのバランスを柔軟に変更できること、実装（タスク分割）の変更容易性にあると考えています。

Human-in-the-Loop の実現

AI を用いたワークフローでは AI のアラインメント（要件の擦り合わせ、誤作動による破壊的な操作の防止）のために適切な人間の介入や承認 (Human-in-the-Loop) を設計することが必要不可欠です。
これには通常実行中のワークフローを一時的に中断して状態を永続化しておき、ユーザー入力を元に実行を中断時点から再開するというメカニズムが必要になります。

既存フレームワークはどうなの？

LangGraph や mastra といった主要な LLM フレームワークではこれらは Workflow クラスとして実装されていることが多く、ワークフローの分割や再開についてもサポートされています。

  • LangGraph: ワークフローはグラフとして表現される。状態は Checkpointer を使って永続化し、途中から再開することができる
  • mastra: ワークフローはコードレベルの通常の関数として表現され、ステップ に分割されて相互の関係が定義される。ステップの状態を永続化して途中から再開することができる

これらも試してみたのですが、個人的にはイマイチ求めるものとズレている感じがありました。
具体的には、以下の点が辛くなりそうだなと感じました。

暗黙的な相互依存性

ワークフローの各ステップは通常、先行するステップの出力結果に依存しています。これらのフレームワークでは state.step1 や step1.output といった形で前のステップの結果にアクセスすることができますが、これらの値は step1 が実際にその時点で実行された場合にしか存在しないため、実装者が「step1 がそれの出力に依存するステップよりも先に実行されること」を適切に保証する必要があります。^1
これは実装の柔軟性において大きなハードルとなります。要件の変更によってあるステップを削除したり実行順を変更した場合、そのステップの依存関係を把握し影響のあるステップの適切な修正や実行順の確認を行う必要が出てきます。そもそも実行順を意識してワークフローを設計しないといけないというのがストレスです。

根本的な原因は、これらの抽象化が元々「LLMの実行」のためのフレームワークであり、あくまで実行過程そのものに焦点を当てていることにあると思います。ここにSOLID 原則に基づいた設計原則を適用するとすれば、ワークフローにおけるインターフェースは各タスクの入出力であり、これらによりフォーカスが当たり、各ステップの実装内容は隠蔽されるような設計方式であるのが望ましいと思います。さらに言えばワークフローの実行順自体も「実装の詳細」にあたるものであると考えると、実装者が実行順すらも意識しなくて良いような設計が理想であると考えます。

フレームワークへの過度な依存

https://note.com/sangmin/n/ned1705bb6d4e

Langchain 脱却に関する意思決定を詳述した上の記事は印象的でした。特に「必要以上の抽象化」については個人的にも感じるところが多く、特に LangGraph のグラフワークフローは簡単な直列フローを実装したい場合でさえも煩雑すぎる印象がありました。

状態の永続化の話になると事態はさらに厄介です。ソフトウェアレベルであれば部分ごとに複数のフレームワークを共存させることも可能ですが、データレベルでの依存関係ができるとそうはいかず、長期的に運用されるシステムである場合には重大な意思決定となります。

ワークフローの状態保存という一般的な要件に対して特定のフレームワークに依存するという意思決定をすることは、なかなか難しいのではないかと思います。

a2a-artifact-graph

以上の課題意識を元に今回実装したのが ① A2A ネイティブで で、② Artifact を主役のインターフェースとして ③ ステップの疎結合性が担保された ワークフロー設計を持つ a2a-artifact-graph です。

https://github.com/ryukez/a2a-artifact-graph

具体的なコードを見てみましょう。前の記事 で扱ったレシピ提案エージェントのシンプル版を題材とします。

Artifact が主役といっているように、Step の実装に入る前にまず Artifact を定義します。今回は「料理の提案 -> レシピ詳細の出力」というワークフローを考え、それぞれの Artifact を定義します。

A2A の Artifact は parts: (TextPart | DataPart | FilePart)[] という表現能力の高いシグネチャになっているので、より直感的にシンプルな JSON データを扱える dataArtifact() というヘルパーメソッドも用意しています。

const RecipeNameArtifact = dataArtifact(
  "recipe_name",
  z.object({
    name: z.string(),
  })
);

const RecipeDetailArtifact = dataArtifact(
  "recipe_detail",
  z.object({
    description: z.string(),
  })
);

type Artifacts = [
  InstanceType<typeof RecipeNameArtifact>,
  InstanceType<typeof RecipeDetailArtifact>
];

次に、これらの Artifact を出力する builder を定義していきます。

const openai = new OpenAI();

const recipeNameBuilder = defineBuilder<Artifacts>()({
  name: "recipe_name",
  inputs: () => [] as const,
  outputs: () => ["recipe_name"] as const,
  async *build({ history }) {
    assert(history);

    // 2回目以降のユーザーの入力の場合、レシピ名のセレクトとして処理
    if (history.filter((m) => m.role === "user").length >= 2) {
      const message = history[history.length - 1];
      assert(message.parts[0].type === "text");

      yield RecipeNameArtifact.fromData({
        data: { name: message.parts[0].text },
      });
      return;
    }

    // 初回のユーザーの入力の場合、レシピ名の提案を行う
    assert(history && history[0].parts[0].type === "text");
    const input = history[0].parts[0].text;

    const prompt = `あなたは料理のエキスパートです。ユーザーの要望に応じて、Web検索を行い、考えられる料理名を10個提案してください。結果は改行区切りで出力し、他のテキストは含めないでください。ユーザーの要望: ${input}`;

    const response = await openai.chat.completions.create({
      model: "gpt-4o-mini",
      messages: [{ role: "user", content: prompt }],
    });

    yield {
      state: "input-required",
      message: {
        role: "agent",
        parts: [
          {
            type: "text",
            text: `次の候補の中から料理を選んでください: ${(
              response.choices[0].message.content ?? ""
            )
              .split("\n")
              .map((line) => line.trim())
              .filter((line) => line.length > 0)
              .join(", ")}`,
          },
        ],
      },
    };
  },
});

レシピの提案 -> ユーザー入力というフローを実現するために分岐が発生しているためやや複雑ですが、初回はユーザーの入力を元に複数の料理を提案し（state: "input-required"でユーザー入力待ち状態になる）、2 回目以降はユーザーの返答を入力として受け付けて Artifact として返すという実装になっています。

そしてこちらが 2 段目の「レシピ詳細を出力する」ステップになります。

const recipeDetailBuilder = defineBuilder<Artifacts>()({
  name: "recipe_detail",
  inputs: () => ["recipe_name"] as const,
  outputs: () => ["recipe_detail"] as const,
  async *build({ inputs }) {
    const recipeName = inputs.recipe_name.parsed().name;

    const prompt = `あなたは料理のエキスパートです。ユーザーの希望するレシピの詳細を以下の形式で出力してください。結果だけを出力してください。

【料理名】

【材料】（2 人分）

- 材料 1
- 材料 2
  ...

【手順】

1. 手順 1
2. 手順 2

【コツ・ポイント】

- ポイント 1
- ポイント 2

ユーザーの要望：${recipeName}`;

    const response = await openai.chat.completions.create({
      model: "gpt-4o-mini",
      messages: [{ role: "user", content: prompt }],
    });

    yield RecipeDetailArtifact.fromData({
      data: { description: response.choices[0].message.content ?? "" },
    });
  },
});

inputs と outputs で "recipe_name" を受け取り、 "recipe_detail" を返すことが表現されています。
重要なのは inputs を指定することで inputs.recipe_name.parsed().name という形で型安全にアクセスができるようになっていることです。（コードだけだとわからないですが汗）

recipeDetailBuilder は "recipe_name" Artifact のみに依存しており、それを出力する builder が誰なのか、そもそも存在しているのかは知らないし、知る必要がありません。また outputs が後続のワークフローでどう使われるのかも意識する必要がありません。inputs から outputs を計算することが単一の責務であり、ここに集中して実装することができます。

最後にグラフを宣言する部分です。ここまでで定義した artifacts と builders を使ってグラフを構成します。

const artifactGraph = new ArtifactGraph<Artifacts>(
  {
    recipe_name: (artifact) => new RecipeNameArtifact(artifact),
    recipe_detail: (artifact) => new RecipeDetailArtifact(artifact),
  },
  [recipeNameBuilder, recipeDetailBuilder]
);

export async function* recipeAgent({
  task,
  history,
}: TaskContext): AsyncGenerator<TaskYieldUpdate, schema.Task | void, unknown> {
  yield* artifactGraph.run({ task, history, verbose: true });
}

各 builder の実行順を指定していないことに注意してください。（配列で渡していますが、実行順である必要はありません）
それぞれの builder は inputs -> outputs の依存関係を規定するため、これを使って正しく実行可能な実行順を自動的に計算することができます^2。また計算不可能な artifact やループ構造があると（実行時ではなく）宣言時にエラーにすることができます。


入出力の関係から、builderの実行順序を自動で決定することができる

これにより、完全ではないもののステップの実行順についても気を使わないといけない部分がかなり減りました。また依存関係が明示されていることでグラフ構造の静的な解析による検証もしやすくなります。

これらの設計により、以下に示すようなメリットがあると考えています。

疎結合性の向上

既存のフレームワークでは artifact と step が密結合しており、結果として step 間に依存関係が生まれるようになっていました。a2a-artifact-graph では builder は インターフェースである artifact の世界だけに依存しており、実装である他の builder を意識する必要がありません。
入出力が明確に定義されていることで、各 builder のテストも独立に行いやすくなります。

A2A ネイティブなワークフロー再開メカニズム

各 builder の計算結果は artifact として出力されますが、これは A2A の仕様に従って永続化されます。これは完全に A2A の技術仕様に則っており、他フレームワークを使用するときのように独自のデータ設計が必要になりません。

ArtifactGraph は保存された artifact を参照し、全ての出力が計算済みである builder は自動でスキップするようになっています。そのため input-required やエラーでワークフローの実行が中断されたとしても、もう一度呼び出すことで中断した箇所から処理を再開することができます。

A2Aも一種のフレームワークであることは否めませんが、A2Aはオープンな外部プロトコルであって、他のライブラリの内部仕様に依存するよりはリスクが少なく、アプリケーション同士の接続性も担保しやすいはずです（もちろん、A2Aの普及状況次第ですが）。

手続き的プログラミングから宣言的プログラミングへ

従来のワークフローは「最初に A を実行し、次に B を実行し、...」という風に処理を手続き的に記述していました。
a2a-artifact-graph の特徴は 「何を計算するか」 が主役であり、 「どのように計算するか」 が副次的なものとして位置付けられていることです。

これは「柔軟性と信頼性のトレードオフ」に関する一つのアプローチといえそうです。柔軟性を最大限に高めたい場合には、最終的な成果物のみを定義し、必要なツールを与えてエージェントに自由に計算をさせることができます。一方でより信頼性に重きを置きたい場合には中間生成物をガイドとして与えてタスクを分割してあげたり、一部の処理だけを従来の固定コードで計算するといった組み替えができます。


エージェントに全任せの状態から、部分部分の処理を切り出していく

これは既存のワークフローフレームワークでのタスク分割と本質的には同じですが、思考の過程が少し変わってくるように思います。通常ワークフローの設計は「まず A を計算する必要があり、次に A を使って B を計算し...」という風に順方向に、ボトムアップに設計するのが自然ですが、最終成果物から逆算する場合はトップダウン的です。前者のボトムアップ的なアプローチをとる場合、エージェントvsワークフローの二者択一的になりがちな一方、トップダウン的に分割していく場合はまず完全なエージェントからはじめて、必要に応じて部分部分を切り出していけるため、エージェントからワークフローへとプログラムをより連続的に変化させていきやすいでしょう。

より一般的に言うなら、「人間が What（外部仕様、インターフェース）を設計し、AI が How（内部仕様、実装）を埋める」 というこの協業形態は、AI 時代のエンジニアリングにおける一つの有望なモデルになっていくのではないかと考えています。

さいごに

今回は既存のワークフローフレームワークにおける課題感から出発して A2A 上での新しいワークフロー方式を考えてみました。
今回はワークフローが完全に固定の場合のみを考えましたが、実際には Agent Routing のように途中で処理が分岐するようなケースもあり、別途考慮する必要があるでしょう。この辺りはまた別の記事で考えてみたいと思います。

Discussion