시맨틱 커널 에이전트 오케스트레이션 고급 주제

개발자는 나중에 결과 개체에서 메서드를 호출하여 오케스트레이션 호출의 GetValueAsync 결과를 가져올 수 있습니다. 애플리케이션이 결과를 처리할 준비가 되면 호출이 완료되었을 수도 있고 완료되지 않았을 수도 있습니다. 따라서 개발자는 필요에 따라 메서드에 대한 GetValueAsync 시간 제한을 지정할 수 있습니다. 지정된 시간 제한 내에 오케스트레이션이 완료되지 않으면 시간 제한 예외가 throw됩니다.

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

지정된 시간 제한 내에 오케스트레이션이 완료되지 않으면 시간 제한 예외가 throw됩니다.

개발자는 나중에 결과 개체에서 메서드를 호출하여 오케스트레이션 호출의 get 결과를 가져올 수 있습니다. 애플리케이션이 결과를 처리할 준비가 되면 호출이 완료되었을 수도 있고 완료되지 않았을 수도 있습니다. 따라서 개발자는 필요에 따라 메서드에 대한 get 시간 제한을 지정할 수 있습니다. 지정된 시간 제한 내에 오케스트레이션이 완료되지 않으면 시간 제한 예외가 throw됩니다.

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()