Partilhar via

Fluxos de trabalho do Microsoft Agent Framework - Trabalhando com agentes

Esta página fornece uma visão geral de como usar agentes nos fluxos de trabalho do Microsoft Agent Framework.

Visão geral

Para adicionar inteligência aos seus fluxos de trabalho, você pode aproveitar os agentes de IA como parte da execução do seu fluxo de trabalho. Os agentes de IA podem ser facilmente integrados em fluxos de trabalho, permitindo que você crie soluções complexas e inteligentes que antes eram difíceis de alcançar.

Adicionar um agente diretamente a um fluxo de trabalho

Você pode adicionar agentes ao seu fluxo de trabalho por meio de bordas:

using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
using Microsoft.Agents.AI;

// Create the agents first
AIAgent agentA = new ChatClientAgent(chatClient, instructions);
AIAgent agentB = new ChatClientAgent(chatClient, instructions);

// Build a workflow with the agents
WorkflowBuilder builder = new(agentA);
builder.AddEdge(agentA, agentB);
Workflow<ChatMessage> workflow = builder.Build<ChatMessage>();

Executando o fluxo de trabalho

Dentro do fluxo de trabalho criado acima, os agentes são realmente envolvidos dentro de um executor que lida com a comunicação do agente com outras partes do fluxo de trabalho. O executor pode lidar com três tipos de mensagem:

  • ChatMessage: Uma única mensagem de chat
  • List<ChatMessage>: Uma lista de mensagens de chat
  • TurnToken: Um token de turno que sinaliza o início de uma nova rodada

O executor não aciona o agente para responder até que receba um TurnToken. Todas as mensagens recebidas antes do TurnToken são armazenadas em buffer e enviadas ao agente quando o TurnToken é recebido.

StreamingRun run = await InProcessExecution.StreamAsync(workflow, new ChatMessage(ChatRole.User, "Hello World!"));
// Must send the turn token to trigger the agents. The agents are wrapped as executors.
// When they receive messages, they will cache the messages and only start processing
// when they receive a TurnToken. The turn token will be passed from one agent to the next.
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
{
    // The agents will run in streaming mode and an AgentRunUpdateEvent
    // will be emitted as new chunks are generated.
    if (evt is AgentRunUpdateEvent agentRunUpdate)
    {
        Console.WriteLine($"{agentRunUpdate.ExecutorId}: {agentRunUpdate.Data}");
    }
}

Usando o executor de agente incorporado

Você pode adicionar agentes ao seu fluxo de trabalho por meio de bordas:

from agent_framework import WorkflowBuilder
from agent_framework.azure import AzureChatClient
from azure.identity import AzureCliCredential

# Create the agents first
chat_client = AzureChatClient(credential=AzureCliCredential())
writer_agent: ChatAgent = chat_client.create_agent(
    instructions=(
        "You are an excellent content writer. You create new content and edit contents based on the feedback."
    ),
    name="writer_agent",
)
reviewer_agent = chat_client.create_agent(
    instructions=(
        "You are an excellent content reviewer."
        "Provide actionable feedback to the writer about the provided content."
        "Provide the feedback in the most concise manner possible."
    ),
    name="reviewer_agent",
)

# Build a workflow with the agents
builder = WorkflowBuilder()
builder.set_start_executor(writer_agent)
builder.add_edge(writer_agent, reviewer_agent)
workflow = builder.build()

Executando o fluxo de trabalho

Dentro do fluxo de trabalho criado acima, os agentes são realmente envolvidos dentro de um executor que lida com a comunicação do agente com outras partes do fluxo de trabalho. O executor pode lidar com três tipos de mensagem:

  • str: Uma única mensagem de chat em formato string
  • ChatMessage: Uma única mensagem de chat
  • List<ChatMessage>: Uma lista de mensagens de chat

Sempre que o executor receber uma mensagem de um desses tipos, ele acionará o agente para responder, e o tipo de resposta será um AgentExecutorResponse objeto. Esta classe contém informações úteis sobre a resposta do agente, incluindo:

  • executor_id: A ID do executor que produziu esta resposta
  • agent_run_response: A resposta completa do agente
  • full_conversation: O histórico completo da conversa até este ponto

Dois possíveis tipos de eventos relacionados às respostas dos agentes podem ser emitidos ao executar o fluxo de trabalho:

  • AgentRunUpdateEvent contendo partes da resposta do agente à medida que são geradas no modo de streaming.
  • AgentRunEvent contendo a resposta completa do agente em modo não streaming.

Por padrão, os agentes são encapsulados em executores que são executados no modo de streaming. Você pode personalizar esse comportamento criando um executor personalizado. Consulte a próxima seção para obter mais detalhes.

last_executor_id = None
async for event in workflow.run_streaming("Write a short blog post about AI agents."):
    if isinstance(event, AgentRunUpdateEvent):
        if event.executor_id != last_executor_id:
            if last_executor_id is not None:
                print()
            print(f"{event.executor_id}:", end=" ", flush=True)
            last_executor_id = event.executor_id
        print(event.data, end="", flush=True)

Usando um executor de agente personalizado

Às vezes, você pode querer personalizar como os agentes de IA são integrados em um fluxo de trabalho. Você pode conseguir isso criando um executor personalizado. Isto permite-lhe controlar:

  • A invocação do agente: fluxo contínuo ou não-contínuo
  • Os tipos de mensagem que o agente irá manipular, incluindo tipos de mensagem personalizados
  • O ciclo de vida do agente, incluindo inicialização e limpeza
  • A utilização de threads de agente e outros recursos
  • Eventos adicionais emitidos durante a execução do agente, incluindo eventos personalizados
  • Integração com outros recursos de fluxo de trabalho, como estados compartilhados e solicitações/respostas
internal sealed class CustomAgentExecutor : Executor<CustomInput, CustomOutput>("CustomAgentExecutor")
{
    private readonly AIAgent _agent;

    /// <summary>
    /// Creates a new instance of the <see cref="CustomAgentExecutor"/> class.
    /// </summary>
    /// <param name="agent">The AI agent used for custom processing</param>
    public CustomAgentExecutor(AIAgent agent) : base("CustomAgentExecutor")
    {
        this._agent = agent;
    }

    public async ValueTask<CustomOutput> HandleAsync(CustomInput message, IWorkflowContext context)
    {
        // Retrieve any shared states if needed
        var sharedState = await context.ReadStateAsync<SharedStateType>("sharedStateId", scopeName: "SharedStateScope");

        // Render the input for the agent
        var agentInput = RenderInput(message, sharedState);

        // Invoke the agent
        // Assume the agent is configured with structured outputs with type `CustomOutput`
        var response = await this._agent.RunAsync(agentInput);
        var customOutput = JsonSerializer.Deserialize<CustomOutput>(response.Text);

        return customOutput;
    }
}

from agent_framework import (
    ChatAgent,
    ChatMessage,
    Executor,
    WorkflowContext,
    handler
)

class Writer(Executor):

    agent: ChatAgent

    def __init__(self, chat_client: AzureChatClient, id: str = "writer"):
        # Create a domain specific agent using your configured AzureChatClient.
        agent = chat_client.create_agent(
            instructions=(
                "You are an excellent content writer. You create new content and edit contents based on the feedback."
            ),
        )
        # Associate the agent with this executor node. The base Executor stores it on self.agent.
        super().__init__(agent=agent, id=id)

    @handler
    async def handle(self, message: ChatMessage, ctx: WorkflowContext[list[ChatMessage]]) -> None:
        """Handles a single chat message and forwards the accumulated messages to the next executor in the workflow."""
        # Invoke the agent with the incoming message and get the response
        messages: list[ChatMessage] = [message]
        response = await self.agent.run(messages)
        # Accumulate messages and send them to the next executor in the workflow.
        messages.extend(response.messages)
        await ctx.send_message(messages)

Próximas Etapas

  • Last updated on