セマンティック カーネル エージェント オーケストレーションの高度なトピック

開発者は、後で結果オブジェクトで GetValueAsync メソッドを呼び出すことによって、オーケストレーション呼び出しの結果を取得できます。 アプリケーションが結果を処理する準備が整ったとき、呼び出しは完了しているかもしれませんし、していないかもしれません。 そのため、開発者は必要に応じて、 GetValueAsync メソッドのタイムアウトを指定できます。 指定したタイムアウト内にオーケストレーションが完了しない場合は、タイムアウト例外がスローされます。

string output = await result.GetValueAsync(TimeSpan.FromSeconds(60));

指定したタイムアウト内にオーケストレーションが完了しない場合は、タイムアウト例外がスローされます。

開発者は、後で結果オブジェクトで get メソッドを呼び出すことによって、オーケストレーション呼び出しの結果を取得できます。 アプリケーションが結果を処理する準備が整ったとき、呼び出しは完了しているかもしれませんし、していないかもしれません。 そのため、開発者は必要に応じて、 get メソッドのタイムアウトを指定できます。 指定したタイムアウト内にオーケストレーションが完了しない場合は、タイムアウト例外がスローされます。

value = await orchestration_result.get(timeout=60)

指定したタイムアウト内にオーケストレーションが完了しない場合は、タイムアウト例外が発生します。

呼び出し内でエージェントの応答を表示するために、開発者はオーケストレーションに ResponseCallback を提供できます。 これにより、開発者はオーケストレーション プロセス中に各エージェントからの応答を確認できます。 開発者は、UI の更新、ログ記録、またはその他の目的でこのコールバックを使用できます。

public ValueTask ResponseCallback(ChatMessageContent response)
{
    Console.WriteLine($"# {response.AuthorName}\n{response.Content}");
    return ValueTask.CompletedTask;
}

SequentialOrchestration orchestration = new SequentialOrchestration(
    analystAgent, writerAgent, editorAgent)
{
    ResponseCallback = ResponseCallback,
};

呼び出し内でエージェントの応答を表示するために、開発者はオーケストレーションに agent_response_callback を提供できます。 これにより、開発者はオーケストレーション プロセス中に各エージェントからの応答を確認できます。 開発者は、UI の更新、ログ記録、またはその他の目的でこのコールバックを使用できます。

def agent_response_callback(message: ChatMessageContent) -> None:
    print(f"# {message.name}\n{message.content}")

sequential_orchestration = SequentialOrchestration(
    members=agents,
    agent_response_callback=agent_response_callback,
)

ユーザー入力 (ハンドオフやグループ チャットなど) をサポートするオーケストレーションの場合は、ユーザーからInteractiveCallbackを返すChatMessageContentを指定します。 このコールバックを使用すると、開発者は、UI プロンプトの表示や他のシステムとの統合など、ユーザー入力を収集するカスタム ロジックを実装できます。

HandoffOrchestration orchestration = new(...)
{
    InteractiveCallback = () =>
    {
        Console.Write("User: ");
        string input = Console.ReadLine();
        return new ChatMessageContent(AuthorRole.User, input);
    }
};

ユーザー入力 (ハンドオフやグループ チャットなど) をサポートするオーケストレーションの場合は、ユーザーからhuman_response_functionを返すChatMessageContentを指定します。 このコールバックを使用すると、開発者は、UI プロンプトの表示や他のシステムとの統合など、ユーザー入力を収集するカスタム ロジックを実装できます。

def human_response_function() -> ChatMessageContent:
    user_input = input("User: ")
    return ChatMessageContent(role=AuthorRole.USER, content=user_input)

handoff_orchestration = HandoffOrchestration(
    ...,
    agent_response_callback=agent_response_callback,
)

開発者は、厳密に型指定された入力クラスを使用し、オーケストレーションのジェネリック パラメーターとして指定することで、構造化データを入力としてオーケストレーションに渡すことができます。 これにより、型の安全性と、複雑なデータ構造を処理するためのオーケストレーションの柔軟性が向上します。 たとえば、GitHub の問題をトリアージするには、構造化された入力のクラスを定義します。

public sealed class GithubIssue
{
    public string Id { get; set; } = string.Empty;
    public string Title { get; set; } = string.Empty;
    public string Body { get; set; } = string.Empty;
    public string[] Labels { get; set; } = [];
}

開発者は、この型をジェネリック パラメーターとして指定することで、オーケストレーションへの入力として使用できます。

HandoffOrchestration<GithubIssue, string> orchestration =
    new(...);

GithubIssue input = new GithubIssue { ... };
var result = await orchestration.InvokeAsync(input, runtime);

カスタム入力変換

既定では、オーケストレーションは組み込みの入力変換を使用します。この変換により、オブジェクトが JSON にシリアル化され、 ChatMessageContentでラップされます。 構造化入力を基になるメッセージ型に変換する方法をカスタマイズする場合は、 InputTransform プロパティを使用して独自の入力変換関数を指定できます。

HandoffOrchestration<GithubIssue, string> orchestration =
    new(...)
    {
        InputTransform = (issue, cancellationToken) =>
        {
            // For example, create a chat message with a custom format
            var message = new ChatMessageContent(AuthorRole.User, $"[{issue.Id}] {issue.Title}\n{issue.Body}");
            return ValueTask.FromResult<IEnumerable<ChatMessageContent>>([message]);
        },
    };

これにより、型指定された入力をエージェントに表示する方法を正確に制御でき、ユーザー設定の書式設定、フィールドの選択、複数メッセージ入力などの高度なシナリオが可能になります。

開発者は、オーケストレーションのジェネリック パラメーターとして Pydantic モデル (または任意のカスタム クラス) を指定することで、構造化データをオーケストレーションへの入力として渡すことができます。 これにより、型の安全性が有効になり、オーケストレーションが複雑なデータ構造を処理できるようになります。

たとえば、GitHub の問題をトリアージするには、構造化された入力の Pydantic モデルを定義します。

from pydantic import BaseModel

class GithubIssue(BaseModel):
    id: str
    title: str
    body: str
    labels: list[str] = []

その後、この型をジェネリック パラメーターとして指定することで、オーケストレーションへの入力として使用できます。

from semantic_kernel.agents import HandoffOrchestration

def custom_input_transform(input_message: GithubIssue) -> ChatMessageContent:
    return ChatMessageContent(role=AuthorRole.USER, content=f"[{input_message.id}] {input_message.title}\n{input_message.body}")


handoff_orchestration = HandoffOrchestration[GithubIssue, ChatMessageContent](
    ...,
    input_transform=custom_input_transform,
)

GithubIssueSample = GithubIssue(
    id="12345",
    title="Bug: ...",
    body="Describe the bug...",
    labels=[],
)

orchestration_result = await handoff_orchestration.invoke(
    task=GithubIssueSample,
    runtime=runtime,
)

エージェントとオーケストレーションは、厳密に型指定された出力クラスをオーケストレーションのジェネリック パラメーターとして指定することで、構造化された出力を返すことができます。 これにより、プレーンテキストではなく、アプリケーションで豊富な構造化された結果を操作できます。

たとえば、記事を分析し、テーマ、センチメント、エンティティを抽出するとします。 構造化された出力のクラスを定義します。

public sealed class Analysis
{
    public IList<string> Themes { get; set; } = [];
    public IList<string> Sentiments { get; set; } = [];
    public IList<string> Entities { get; set; } = [];
}

その後、この型をジェネリック パラメーターとして指定することで、オーケストレーションの出力として使用できます。

ConcurrentOrchestration<string, Analysis> orchestration =
    new(agent1, agent2, agent3)
    {
        ResultTransform = outputTransform.TransformAsync, // see below
    };

// ...
OrchestrationResult<Analysis> result = await orchestration.InvokeAsync(input, runtime);
Analysis output = await result.GetValueAsync(TimeSpan.FromSeconds(60));

カスタム出力変換

既定では、オーケストレーションは組み込みの出力変換を使用します。この変換は、エージェントの応答コンテンツを出力の種類に逆シリアル化しようとします。 より高度なシナリオでは、カスタム出力変換 (たとえば、一部のモデルによる構造化された出力) を提供できます。

StructuredOutputTransform<Analysis> outputTransform =
    new(chatCompletionService, new OpenAIPromptExecutionSettings { ResponseFormat = typeof(Analysis) });

ConcurrentOrchestration<string, Analysis> orchestration =
    new(agent1, agent2, agent3)
    {
        ResultTransform = outputTransform.TransformAsync,
    };

この方法を使用すると、オーケストレーションから構造化データを直接受信して処理できるため、高度なワークフローと統合を簡単に構築できます。

エージェントとオーケストレーションは、オーケストレーションのジェネリック出力の種類として Pydantic モデル (または任意のカスタム クラス) を指定することで、構造化された出力を返すことができます。 これにより、プレーンテキストではなく、アプリケーションで豊富な構造化された結果を操作できます。

たとえば、記事を分析し、テーマ、センチメント、エンティティを抽出するとします。 構造化出力の Pydantic モデルを定義します。

from pydantic import BaseModel

class ArticleAnalysis(BaseModel):
    themes: list[str]
    sentiments: list[str]
    entities: list[str]

その後、この型をジェネリック パラメーターとして指定し、出力変換を指定することで、オーケストレーションの出力としてこの型を使用できます。

from semantic_kernel.agents import ConcurrentOrchestration
from semantic_kernel.agents.orchestration.tools import structured_outputs_transform
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion

# `structured_outputs_transform` is a built-in transform that uses the structured output

concurrent_orchestration = ConcurrentOrchestration[str, ArticleAnalysis](
    members=agents,
    output_transform=structured_outputs_transform(ArticleAnalysis, AzureChatCompletion()),
)

...

orchestration_result = await concurrent_orchestration.invoke(
    task="article text",
    runtime=runtime,
)

value = await orchestration_result.get(timeout=20)

# `value` is now an instance of ArticleAnalysis

この方法を使用すると、オーケストレーションから構造化データを直接受信して処理できるため、高度なワークフローと統合を簡単に構築できます。

結果ハンドラーで Cancel メソッドを呼び出すことで、オーケストレーションを取り消すことができます。 これにより、すべてのエージェントにシグナルを伝達することでオーケストレーションが停止され、それ以降のメッセージの処理が停止されます。

var resultTask = orchestration.InvokeAsync(input, runtime);
resultTask.Cancel();

開発者は、結果ハンドラーで cancel メソッドを呼び出すことによってオーケストレーションを取り消すことができます。 これにより、すべてのエージェントにシグナルを伝達することでオーケストレーションが停止され、それ以降のメッセージの処理が停止されます。

orchestration_result = await orchestration.invoke(task=task, runtime=runtime)
orchestration_result.cancel()