Tópicos avançados de orquestração de agentes do Kernel Semântico

Os desenvolvedores podem obter o resultado de uma invocação de orquestração mais tarde chamando o GetValueAsync método no objeto de resultado. Quando o aplicativo estiver pronto para processar o resultado, a invocação poderá ou não ter sido concluída. Portanto, os desenvolvedores podem, opcionalmente, especificar um tempo limite para o GetValueAsync método. Se a orquestração não for concluída dentro do tempo limite especificado, uma exceção de tempo limite será gerada.

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

Se a orquestração não for concluída dentro do tempo limite especificado, uma exceção de tempo limite será gerada.

Os desenvolvedores podem obter o resultado de uma invocação de orquestração mais tarde chamando o get método no objeto de resultado. Quando o aplicativo estiver pronto para processar o resultado, a invocação poderá ou não ter sido concluída. Portanto, os desenvolvedores podem, opcionalmente, especificar um tempo limite para o get método. Se a orquestração não for concluída dentro do tempo limite especificado, uma exceção de tempo limite será gerada.

value = await orchestration_result.get(timeout=60)

Se a orquestração não for concluída dentro do tempo limite especificado, uma exceção de tempo limite será gerada.

Para ver as respostas do agente dentro de uma invocação, os desenvolvedores podem fornecer um ResponseCallback para a orquestração. Isso permite que os desenvolvedores observem as respostas de cada agente durante o processo de orquestração. Os desenvolvedores podem usar esse retorno de chamada para atualizações de interface do usuário, para registro ou outras finalidades.

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

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

Para ver as respostas do agente dentro de uma invocação, os desenvolvedores podem fornecer um agent_response_callback para a orquestração. Isso permite que os desenvolvedores observem as respostas de cada agente durante o processo de orquestração. Os desenvolvedores podem usar esse retorno de chamada para atualizações de interface do usuário, para registro ou outras finalidades.

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

Para as orquestrações que dão suporte à entrada do usuário (por exemplo, transferência e chat em grupo), forneça um InteractiveCallback que retorna um ChatMessageContent do usuário. Usando esse retorno de chamada, os desenvolvedores podem implementar lógica personalizada para coletar informações do usuário, como exibir uma janela de interface ou integrar com outros sistemas.

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

Para as orquestrações que dão suporte à entrada do usuário (por exemplo, transferência e chat em grupo), forneça um human_response_function que retorna um ChatMessageContent do usuário. Usando esse retorno de chamada, os desenvolvedores podem implementar lógica personalizada para coletar informações do usuário, como exibir uma janela de interface ou integrar com outros sistemas.

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

Os desenvolvedores podem passar dados estruturados como entrada para orquestrações usando uma classe de entrada fortemente tipada e especificando-a como o parâmetro genérico para a orquestração. Isso permite segurança de tipos, oferecendo mais flexibilidade para as orquestrações lidarem com estruturas de dados complexas. Por exemplo, para triagem de problemas do GitHub, defina uma classe para a entrada estruturada:

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; } = [];
}

Os desenvolvedores podem usar esse tipo como entrada para uma orquestração fornecendo-o como o parâmetro genérico:

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

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

Transformações de entrada personalizadas

Por padrão, a orquestração usará a transformação de entrada interna, que serializa o objeto para JSON e o encapsula em um ChatMessageContent. Se você quiser personalizar como sua entrada estruturada é convertida no tipo de mensagem subjacente, você pode fornecer sua própria função de transformação de entrada por meio da InputTransform propriedade:

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]);
        },
    };

Isso permite que você controle exatamente como a entrada tipada é apresentada aos agentes, permitindo cenários avançados, como formatação personalizada, seleção de campo ou entrada de várias mensagens.

Os desenvolvedores podem passar dados estruturados como entrada para orquestrações especificando um modelo Pydantic (ou qualquer classe personalizada) como o parâmetro genérico para a orquestração. Isso permite a segurança de tipo e possibilita que as orquestrações lidem com estruturas de dados complexas.

Por exemplo, para triagem de problemas do GitHub, defina um modelo Pydantic para a entrada estruturada:

from pydantic import BaseModel

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

Em seguida, você pode usar esse tipo como a entrada para sua orquestração fornecendo-o como o parâmetro genérico:

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

Agentes e orquestrações podem retornar saídas estruturadas especificando uma classe de saída com tipagem forte como parâmetro genérico da orquestração. Isso permite que você trabalhe com resultados estruturados e avançados em seu aplicativo, em vez de apenas texto sem formatação.

Por exemplo, suponha que você queira analisar um artigo e extrair temas, sentimentos e entidades. Defina uma classe para a saída estruturada:

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

Em seguida, você pode usar esse tipo como a saída para sua orquestração fornecendo-o como o parâmetro genérico:

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

Transformações de saída personalizadas

Por padrão, a orquestração usará a transformação de saída incorporada, que tenta desserializar o conteúdo de resposta do agente para o tipo de saída. Para cenários mais avançados, você pode fornecer uma transformação de saída personalizada (por exemplo, com saída estruturada por alguns modelos).

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

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

Essa abordagem permite que você receba e processe dados estruturados diretamente da orquestração, facilitando a criação de fluxos de trabalho e integrações avançadas.

Agentes e orquestrações podem retornar saídas estruturadas especificando um modelo Pydantic (ou qualquer classe personalizada) como o tipo de saída genérico para a orquestração. Isso permite que você trabalhe com resultados estruturados e avançados em seu aplicativo, em vez de apenas texto sem formatação.

Por exemplo, suponha que você queira analisar um artigo e extrair temas, sentimentos e entidades. Defina um modelo Pydantic para a saída estruturada:

from pydantic import BaseModel

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

Em seguida, você pode usar esse tipo como a saída para sua orquestração fornecendo-o como o parâmetro genérico e especificando uma transformação de saída:

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

Essa abordagem permite que você receba e processe dados estruturados diretamente da orquestração, facilitando a criação de fluxos de trabalho e integrações avançadas.

Você pode cancelar uma orquestração chamando o Cancel método no manipulador de resultados. Isso interromperá a orquestração propagando o sinal para todos os agentes e eles deixarão de processar mais mensagens.

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

Os desenvolvedores podem cancelar uma orquestração chamando o cancel método no manipulador de resultados. Isso interromperá a orquestração propagando o sinal para todos os agentes e eles deixarão de processar mais mensagens.

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