Deep Research リポジトリに PR を出しました
先日、同僚の Junpei Tsuchida さんが Microsoftが公開しているDeep Researchのサンプルコードを理解してみる（ついでにC＃ライブラリ化も） という記事を公開しました。
そのリポジトリに追加してもらって以下の Pull Request を作成しました。
Azure.AI.OpenAI を Microsoft.Extensions.AI に変更 ＋ C#の新文法に対応
この Pull Request は Deep Research のロジック内で Azure.AI.OpenAI の ChatClient クラスを使用している部分を Microsoft.Extensions.AI の IChatClient インターフェースに変更しました。ついでに自前実装していた JSON Schema 生成のロジックが Microsoft.Extensions.AI にあったので、それを使うようにしました。
変更前
変更後
これで Deep Research のロジックは外部サービスへの依存を減らして ISearchClient と IChatClient を実装したクラスを DI で好きに差し替え可能になりました。
本当にやりたかったこと
この Pull Request を出した目的は Deep Research のロジックを Durable Functions で動かしたかったからです。Deep Research のロジックは、そこそこの時間がかかるので長時間実行が可能な Durable Functions で動かすのが適していると考えました。
Durable Functions は以下の 2 種類のトリガーの関数があって、それぞれ以下のような特徴があります。
- Orchestration Trigger
- Durable Functions のワークフローを定義する関数
- 長時間実行が可能で、状態を保存して再開できる
- Activity Trigger の関数を呼び出すことができる
- 長時間実行が可能だが、外部サービスへの IO などの処理が出来ないという制限がある
- Activity Trigger
- Durable Functions のワークフローから呼び出される関数
- 短時間実行が前提で、通常の関数と同じように外部サービスへの IO などの処理が可能
- Orchestration Trigger から呼び出される
Orchestration Trigger の関数は状態が保存されて再開できるような特殊な動作をする関係上、以下のような制約があります：
- Single-threading での実行: 単一のディスパッチャースレッドで実行されるため、効率的なコードが必要
- 決定論的な動作が必要: 同じ入力に対して常に同じ結果を返す必要がある
-
ランダムな値や現在時刻の直接取得禁止:
Random.Next()やDateTime.Nowなどの非決定的な値の使用は禁止
これらの制約により、外部サービスへの HTTP リクエストや、データベースのクエリなど、ランダムな挙動や外部ステータスに依存して動作が変わるような処理は Orchestration Trigger の関数内では実行できません。そのような処理を行うには Activity Trigger の関数を呼び出して、その中で実行する必要があります。
そのため、元々の実装だと Deep Research のロジック内で直接 Azure.AI.OpenAI の ChatClient を使用して AI を呼び出していたため、Durable Functions の Orchestration Trigger の関数内で直接呼び出すことができませんでした。そこを Microsoft.Extensions.AI の IChatClient インターフェースに変更することで、実装を差し替え可能にして IChatClient の実装を Activity Trigger の関数で呼び出すことができるようにしました。
以下は、この Deep Research の処理の中核を担う DeepResearchService の基本的なワークフローのシーケンス図です。(AI に書いてもらったのでちょっと違うかもですが、基本的な流れは合っています)
全ての外部サービスの呼び出しをインターフェース経由で行うようになったので Durable Functions の Orchestration Trigger の関数で呼び出すための下準備が整いました。これにより、Deep Research のロジックを Durable Functions の Orchestration Trigger の関数で実行できるようになりました。次にやったのは IChatClient と ISearchClient の実装クラスを作って Durable Functions の Activity Trigger の関数を呼び出すようにすることでした。
例えば IChatClient の実装クラスは以下のように Microsoft.Extensions.AI の IChatClient インターフェースを実装して、Durable Functions の Activity Trigger の関数を呼び出すようにしました。
using Microsoft.DurableTask;
using Microsoft.Extensions.AI;
namespace LongRunningDeepResearch.ChatClient;
public class OrchestratorChatClient(TaskOrchestrationContext taskOrchestrationContext) : IChatClient
{
public void Dispose() { }
public async TaskChatResponse> GetResponseAsync(IEnumerableChatMessage> messages, ChatOptions? options = null, CancellationToken cancellationToken = default)
{
return await taskOrchestrationContext.CallActivityAsyncChatResponse>(
nameof(GetResponseActivity),
new GetResponseArguments(messages, options));
}
public object? GetService(Type serviceType, object? serviceKey = null)
{
if (serviceType == typeof(IChatClient)) return this;
return null;
}
public IAsyncEnumerableChatResponseUpdate> GetStreamingResponseAsync(IEnumerableChatMessage> messages, ChatOptions? options = null, CancellationToken cancellationToken = default)
{
throw new NotSupportedException();
}
}
public record GetResponseArguments(IEnumerableChatMessage> Messages, ChatOptions? Options);
そして Activity Trigger の関数は以下のように IChatClient を DI で受け取って、実際の処理を呼び出します。
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.DependencyInjection;
namespace LongRunningDeepResearch.ChatClient;
public class GetResponseActivity([FromKeyedServices(GetResponseActivity.ChatClientKey)]IChatClient chatClient)
{
public const string ChatClientKey = $"{nameof(GetResponseActivity)}_{nameof(ChatClientKey)}";
[Function(nameof(GetResponseActivity))]
public async TaskChatResponse> GetResponseAsync(
[ActivityTrigger]GetResponseArguments args,
CancellationToken cancellationToken = default) =>
await chatClient.GetResponseAsync(args.Messages, args.Options, cancellationToken);
}
同じ要領で ISearchClient の実装クラスも作成しました。そして Durable Functions の Orchestration Trigger の関数内で、以下のようにして DeepResearchService を呼び出すようにしました。
using DeepResearch.Core;
using LongRunningDeepResearch.ChatClient;
using LongRunningDeepResearch.SearchClient;
using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;
namespace LongRunningDeepResearch;
public static class DeepResearchOrchestrator
{
[Function(nameof(DeepResearchOrchestrator))]
public static async TaskResearchResult> RunOrchestrator(
[OrchestrationTrigger] TaskOrchestrationContext context)
{
ILogger logger = context.CreateReplaySafeLogger(nameof(DeepResearchOrchestrator));
var topic = context.GetInputstring>();
if (string.IsNullOrWhiteSpace(topic))
{
throw new ArgumentException("Topic cannot be null or empty.", nameof(topic));
}
var deepResearchService = new DeepResearchService(
new OrchestratorChatClient(context),
new OrchestratorSearchClient(context));
return await deepResearchService.RunResearchAsync(topic,
new DeepResearchOptions
{
MaxResearchLoops = 3,
}));
}
}
これで Durable Functions の Orchestration Trigger の関数で Deep Research のロジックを実行できるようになりました。実際に MaxResearchLoops を 50 とかにして実行してみたのですが、何十分もの間動き続けていました。Azure の PaaS 系サービスは HTTP 通信のタイムアウトが 230 秒という制限があるので Deep Research 的な処理をやる場合は、HTTP リクエストの処理内でやるのではなく、メッセージキューとかを使って非同期に処理をするのがおすすめです。仮に HTTP リクエストのタイムアウトがなかったとしても、一般的にはあまり長時間の HTTP リクエストは避けるべきです。今回のように Durable Functions で動くようにしておけば裏でゆっくり時間をかけて処理を実行できるので、HTTP リクエストのタイムアウトを気にせずに済みます。
実際のコードは、lab/durable-functions ブランチにあります。興味のあるかたは見てみてください。
まとめ
AI を扱う処理に限った話ではないですがコアのロジックをインターフェースを使って外部の世界への依存を減らすことで、外部世界の変化の都合に振り回されないようにすることが出来ます。今回の Deep Research のロジックも IChatClient と ISearchClient のインターフェースを使うことで、Durable Functions の Orchestration Trigger の関数で実行できるようにしました。もちろん IChatClient と ISearchClient の実装を普通のものにしておけば Durable Functions じゃないところでも動きます。
土田さんのリポジトリのコードを見てから、こういうことが出来そうだなと思っていたのですが、実際にやってみてちゃんと動いたので個人的には満足しています。
Views: 0
