AI Agent のオーケストレーションの観点で再度注目を集めている Azure Functions の Durable Functions について、特に fan in / fan Out の観点で動作を確認した。

Overview

Microsoft の大規模なカンファレンスである 2024 Ignite において、特に話題になったのが TOYOTA の O-Beya の事例である。複数の専門知識を持つ Agent 同士が協調してユーザーに対して回答を出すことから、マルチエージェントの最新事例として注目が集まっていた。その中で、具体的な AI Agent の制御に関しては、Azure Functions の Durable Functions を利用していることが発表された。

How Toyota uses Azure Cosmos DB to power their multi-agent AI system for enhanced productivity

以下のブログ記事でも Durable Functions と AI Agent に関して紹介がされている。AI Agent のオーケストレーションとなると Semantic Kernel, AutoGen, LangChain といったライブラリやフレームワークが話題になっているが、Durable Functions は従来からある機能のため、安定した実装が期待できると注目されている。

AI Challenge Day における ZEN のアーキテクチャ設計アプローチを解説！

Durable Functions とは

Azure Functions の拡張機能であり、ステートフル関数を記述することができる。こちらの機能によって状態や再実行がバックグラウンドで管理されるため、ユーザーはこれらの実装を意識することなく安定した実行が可能である。 Durable Functions を利用したアプリケーションパターンとして、いくつか紹介されているが、その中の一つのパターンとして fan-in / fan-out が存在する。

Durable Functions とは

上記の図にあるように、F1 がオーケストレータとして動作し、前処理を行った上で F2 が並列で実行される (fan-out)。その後、並列で実行された結果がすべて返ってくるまで待機し、得られた結果を F3 に渡す (fan-in) ことができる。 通常の関数では、fan-out はそこまで難しくなく、複数のメッセージを 1 つのキューに送信することで実現できる。しかしながら、 fan-in して戻ることが容易ではない。 通常の関数で fan-in する場合には、キューによってトリガーされた関数の終了を追跡した後、関数の出力を格納するコードを記述する必要があるためである。
Durable Functions が、サーバーレス関数の実行状態の管理や、失敗時の再試行を担うため、ユーザーがこれらのロジックを意識する必要がなくビジネスロジックの実装が可能である。クラウドアプリケーションを実装する上での再試行などの重要性については、以下の書籍で詳しく説明されており、Durable Functions、fan-in / fan-out のメリットをより理解することができた。

クラウドアプリケーション 10の設計原則　「Azureアプリケーションアーキテクチャガイド」から学ぶ普遍的な原理原則

マルチエージェントのアーキテクチャを考えると、ユーザーの質問をそれぞれの AI Agent に渡して fan-out し、すべての回答を組み合わせ fan-in することで、各専門家の知識を統合して回答するような実装ができると思われる。

以下の動画では、Durable Functions の開発者がメリットと内部動作に関して語られている。加えて、AI Agent を Durable Functions でオーケストレーションしているサンプルに関しても紹介されているため非常に勉強になった。

Hack Azure! #18 - Durable Functions 特集！

検証

基本は以下の手順でローカルでの Functions の開発から進めていき、fan-in / fan-out のサンプルコードを参考に以下のコードで検証した。最終的に、こちらを Azure Functions にデプロイし、実行させている。

クイック スタート: Python Durable Functions アプリを作成する

Durable Functions とは

以下の図にあるように、F2 が並列で実行され、それらの結果が出そろってから F3 が実行される。具体的には、F1 で定義されているリストの値がそれぞれの F2 に渡され、並列して実行される。F2 においては値の二乗が計算され、すべての F2 の実行が完了したら、それらの結果を合計して F3 に渡され文字列が返されている。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

import azure.functions as func
import azure.durable_functions as df

myApp = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

# An HTTP-triggered function with a Durable Functions client binding
@myApp.route(route="orchestrators/{functionName}")
@myApp.durable_client_input(client_name="client")
async def http_start(req: func.HttpRequest, client):
    function_name = req.route_params.get('functionName')
    instance_id = await client.start_new(function_name)
    response = client.create_check_status_response(req, instance_id)
    return response

# Orchestrator
@myApp.orchestration_trigger(context_name="context")
def orchestrator_function(context: df.DurableOrchestrationContext):
    # Get a list of N work items to process in parallel.
    work_batch = yield context.call_activity("f1", None)

    parallel_tasks = [ context.call_activity("f2", b) for b in work_batch ]

    outputs = yield context.task_all(parallel_tasks)

    # Aggregate all N outputs and send the result to F3.
    total = sum(outputs)
    result = yield context.call_activity("f3", total)
    return result

# Activity functions
@myApp.activity_trigger(input_name="input")
def f1(input: str) -> list:
    # Simulate getting a batch of work items.
    work_items = [1, 2, 3, 4, 5]
    return work_items

@myApp.activity_trigger(input_name="input")
def f2(input: int) -> int:
    # Simulate processing each work item. Here we just return the item squared.
    return input * input

@myApp.activity_trigger(input_name="input")
def f3(input: int):
    # Simulate aggregating results. Here we'll just print the total.
    return f"Total sum of outputs: {input}"

実行結果は以下であり、期待した結果になっていることが確認できる。これにより、細かい状態の管理や再試行を実装せずとも、並列実行することが確認できた。

まとめ

AI Agent の注目に伴い、改めて期待が集まっている Durable Functions に関して触れてみた。簡単な実装で並列処理を行うことができ、再試行などに関しても意識して実装することがないのは便利であると思う。ここら辺の実装がいかに重要であるかは、「クラウドアプリケーション 10の設計原則」の中で述べられている。また、独自で fan-in の実装をするとなると、手間がかかることも説明されているため、非常に参考になった。クラウドアプリケーションにて、再試行が適切に実施されておらず、問題が深刻化したケースを何度か見てきているため、意識せずともこのあたりが Durable Functions でカバーされるメリットは非常に大きいと思う。