Flussi di lavoro di Microsoft Agent Framework - Uso di flussi di lavoro come agenti

Requisiti

Per usare un flusso di lavoro come agente, l'executor di avvio del flusso di lavoro deve essere in grado di gestire IEnumerable<ChatMessage> come input. Questa operazione viene soddisfatta automaticamente quando si usano executor basati su agente creati con AsAIAgent.

Creare un agente del flusso di lavoro

Usare il AsAIAgent() metodo di estensione per convertire qualsiasi flusso di lavoro compatibile in un agente:

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

// Create agents
AIAgent researchAgent = chatClient.AsAIAgent("You are a researcher. Research and gather information on the given topic.");
AIAgent writerAgent = chatClient.AsAIAgent("You are a writer. Write clear, engaging content based on research.");
AIAgent reviewerAgent = chatClient.AsAIAgent("You are a reviewer. Review the content and provide a final polished version.");

// Build a sequential workflow
var workflow = new WorkflowBuilder(researchAgent)
    .AddEdge(researchAgent, writerAgent)
    .AddEdge(writerAgent, reviewerAgent)
    .Build();

// Convert the workflow to an agent
AIAgent workflowAgent = workflow.AsAIAgent(
    id: "content-pipeline",
    name: "Content Pipeline Agent",
    description: "A multi-agent workflow that researches, writes, and reviews content"
);

Parametri AsAIAgent

Uso degli agenti del flusso di lavoro

Creazione di una sessione

Ogni conversazione con un agente del flusso di lavoro richiede una sessione per gestire lo stato:

// Create a new session for the conversation
AgentSession session = await workflowAgent.CreateSessionAsync();

Esecuzione non in streaming

Per casi d'uso semplici in cui si vuole ottenere la risposta completa:

var messages = new List<ChatMessage>
{
    new(ChatRole.User, "Write an article about renewable energy trends in 2025")
};

AgentResponse response = await workflowAgent.RunAsync(messages, session);

foreach (ChatMessage message in response.Messages)
{
    Console.WriteLine($"{message.AuthorName}: {message.Text}");
}

Esecuzione in streaming

Per gli aggiornamenti in tempo reale durante l'esecuzione del flusso di lavoro:

var messages = new List<ChatMessage>
{
    new(ChatRole.User, "Write an article about renewable energy trends in 2025")
};

await foreach (AgentResponseUpdate update in workflowAgent.RunStreamingAsync(messages, session))
{
    // Process streaming updates from each agent in the workflow
    if (!string.IsNullOrEmpty(update.Text))
    {
        Console.Write(update.Text);
    }
}

Gestione delle richieste di input esterne

Quando un flusso di lavoro contiene executor che richiedono input esterno (usando RequestInfoExecutor), queste richieste vengono visualizzate come chiamate di funzione nella risposta dell'agente:

await foreach (AgentResponseUpdate update in workflowAgent.RunStreamingAsync(messages, session))
{
    // Check for function call requests
    foreach (AIContent content in update.Contents)
    {
        if (content is FunctionCallContent functionCall)
        {
            // Handle the external input request
            Console.WriteLine($"Workflow requests input: {functionCall.Name}");
            Console.WriteLine($"Request data: {functionCall.Arguments}");

            // Provide the response in the next message
        }
    }
}

Serializzazione e ripresa delle sessioni

Le sessioni dell'agente del flusso di lavoro possono essere serializzate per la persistenza e riprese in un secondo momento:

// Serialize the session state
JsonElement serializedSession = await workflowAgent.SerializeSessionAsync(session);

// Store serializedSession to your persistence layer...

// Later, resume the session
AgentSession resumedSession = await workflowAgent.DeserializeSessionAsync(serializedSession);

// Continue the conversation
await foreach (var update in workflowAgent.RunStreamingAsync(newMessages, resumedSession))
{
    Console.Write(update.Text);
}

Requisiti

Per usare un flusso di lavoro come agente, il gestore dell'avvio del flusso di lavoro deve essere in grado di gestire l'input dei messaggi. Questa condizione è soddisfatta automaticamente quando si usano esecutori basati su Agent o agenti.

Creazione di un agente del flusso di lavoro

Chiamare as_agent() su qualsiasi flusso di lavoro compatibile per convertirlo in un agente:

from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential

# Create your chat client and agents
client = AzureOpenAIResponsesClient(
    project_endpoint="<your-endpoint>",
    deployment_name="<your-deployment>",
    credential=AzureCliCredential(),
)

researcher = client.as_agent(
    name="Researcher",
    instructions="Research and gather information on the given topic.",
)

writer = client.as_agent(
    name="Writer",
    instructions="Write clear, engaging content based on research.",
)

# Build a sequential workflow
workflow = SequentialBuilder(participants=[researcher, writer]).build()

# Convert the workflow to an agent
workflow_agent = workflow.as_agent(name="Content Pipeline Agent")

Parametri as_agent

Uso degli agenti del flusso di lavoro

Creazione di una sessione

Facoltativamente, è possibile creare una sessione per gestire lo stato della conversazione tra più turni:

# Create a new session for the conversation
session = await workflow_agent.create_session()

Esecuzione non in streaming

Per casi d'uso semplici in cui si vuole ottenere la risposta completa:

# You can pass a plain string as input
response = await workflow_agent.run("Write an article about AI trends")

for message in response.messages:
    print(f"{message.author_name}: {message.text}")

Esecuzione in streaming

Per gli aggiornamenti in tempo reale durante l'esecuzione del flusso di lavoro:

async for update in workflow_agent.run(
    "Write an article about AI trends",
    stream=True,
):
    if update.text:
        print(update.text, end="", flush=True)

Gestione delle richieste di input esterne

Quando un flusso di lavoro contiene executor che richiedono input esterno (usando request_info), queste richieste vengono visualizzate come chiamate di funzione nella risposta dell'agente. La chiamata di funzione usa il nome WorkflowAgent.REQUEST_INFO_FUNCTION_NAME:

from agent_framework import Content, Message, WorkflowAgent

response = await workflow_agent.run("Process my request")

# Look for function calls in the response
human_review_function_call = None
for message in response.messages:
    for content in message.contents:
        if content.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME:
            human_review_function_call = content

Fornire risposte alle richieste in sospeso

Per continuare l'esecuzione del flusso di lavoro dopo una richiesta di input esterna, creare un risultato della funzione e inviarlo di nuovo:

if human_review_function_call:
    # Parse the request arguments
    request = WorkflowAgent.RequestInfoFunctionArgs.from_json(
        human_review_function_call.arguments
    )

    # Create a response (your custom response type)
    result_data = MyResponseType(approved=True, feedback="Looks good")

    # Create the function call result
    function_result = Content.from_function_result(
        call_id=human_review_function_call.call_id,
        result=result_data,
    )

    # Send the response back to continue the workflow
    response = await workflow_agent.run(Message("tool", [function_result]))

Esempio completo

Ecco un esempio completo che illustra un agente del flusso di lavoro con output di streaming:

import asyncio
import os

from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential


async def main():
    # Set up the chat client
    client = AzureOpenAIResponsesClient(
        project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
        deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
        credential=AzureCliCredential(),
    )

    # Create specialized agents
    researcher = client.as_agent(
        name="Researcher",
        instructions="Research the given topic and provide key facts.",
    )

    writer = client.as_agent(
        name="Writer",
        instructions="Write engaging content based on the research provided.",
    )

    reviewer = client.as_agent(
        name="Reviewer",
        instructions="Review the content and provide a final polished version.",
    )

    # Build a sequential workflow
    workflow = SequentialBuilder(participants=[researcher, writer, reviewer]).build()

    # Convert to a workflow agent
    workflow_agent = workflow.as_agent(name="Content Creation Pipeline")

    # Run the workflow
    print("Starting workflow...")
    print("=" * 60)

    current_author = None
    async for update in workflow_agent.run(
        "Write about quantum computing",
        stream=True,
    ):
        # Show when different agents are responding
        if update.author_name and update.author_name != current_author:
            if current_author:
                print("\n" + "-" * 40)
            print(f"\n[{update.author_name}]:")
            current_author = update.author_name

        if update.text:
            print(update.text, end="", flush=True)

    print("\n" + "=" * 60)
    print("Workflow completed!")


if __name__ == "__main__":
    asyncio.run(main())

Informazioni sulla conversione degli eventi

Quando un flusso di lavoro viene eseguito come agente, gli eventi del flusso di lavoro vengono convertiti in risposte dell'agente. Il tipo di risposta dipende da come chiami run().

• run(): restituisce un oggetto AgentResponse contenente il risultato completo al termine del flusso di lavoro
• run(..., stream=True): restituisce un oggetto iterabile asincrono di oggetti AgentResponseUpdate man mano che il flusso di lavoro viene eseguito, fornendo aggiornamenti in tempo reale

Durante l'esecuzione, gli eventi interni del flusso di lavoro vengono mappati alle risposte dell'agente come indicato di seguito:

Questa conversione consente di usare l'interfaccia dell'agente standard, pur avendo accesso a informazioni dettagliate sul flusso di lavoro quando necessario.