Flux de travail Microsoft Agent Framework - Utilisation de flux de travail en tant qu’agents

Spécifications

Pour utiliser un flux de travail en tant qu’agent, l’exécuteur de démarrage du flux de travail doit être en mesure de gérer IEnumerable<ChatMessage> comme entrée. Cela est automatiquement satisfait lors de l’utilisation d’exécuteurs basés sur agent créés avec AsAIAgent.

Créer un agent de flux de travail

Utilisez la méthode d’extension AsAIAgent() pour convertir n’importe quel flux de travail compatible en agent :

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

Paramètres AsAIAgent

Utilisation d’agents de flux de travail

Création d’une session

Chaque conversation avec un agent de workflow nécessite une session pour gérer l’état :

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

Exécution sans diffusion en continu

Pour les cas d’usage simples où vous souhaitez obtenir la réponse complète :

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

Exécution de streaming

Pour les mises à jour en temps réel à mesure que le flux de travail s’exécute :

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

Gestion des demandes d’entrée externes

Lorsqu’un flux de travail contient des exécuteurs qui demandent une entrée externe (à l’aide RequestInfoExecutor), ces requêtes sont exposées en tant qu’appels de fonction dans la réponse de l’agent :

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
        }
    }
}

Sérialisation et reprise de session

Les sessions de l’agent de flux de travail peuvent être sérialisées pour la persistance et reprise ultérieurement :

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

Spécifications

Pour utiliser un flux de travail en tant qu’agent, l’exécuteur de démarrage du flux de travail doit être en mesure de gérer l’entrée de message. Cela est automatiquement satisfait lors de l’utilisation de Agent ou d'exécuteurs basés sur un agent.

Création d’un agent de flux de travail

Appelez as_agent() tout workflow compatible pour le convertir en agent :

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

Paramètres as_agent

Utilisation d’agents de flux de travail

Création d’une session

Vous pouvez éventuellement créer une session pour gérer l’état de conversation à plusieurs tours :

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

Exécution sans diffusion en continu

Pour les cas d’usage simples où vous souhaitez obtenir la réponse complète :

# 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}")

Exécution de streaming

Pour les mises à jour en temps réel à mesure que le flux de travail s’exécute :

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

Gestion des demandes d’entrée externes

Lorsqu’un flux de travail contient des exécuteurs qui demandent une entrée externe (à l’aide request_info), ces requêtes sont exposées en tant qu’appels de fonction dans la réponse de l’agent. L’appel de fonction utilise le nom 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

Fourniture de réponses aux demandes en attente

Pour poursuivre l’exécution du flux de travail après une demande d’entrée externe, créez un résultat de fonction et renvoyez-le :

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

Exemple complet

Voici un exemple complet illustrant un agent de flux de travail avec une sortie de 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())

Présentation de la conversion d’événements

Lorsqu’un flux de travail s’exécute en tant qu’agent, les événements de flux de travail sont convertis en réponses de l’agent. Le type de réponse dépend de la façon dont vous appelez run():

• run(): retourne un AgentResponse résultat contenant le résultat complet une fois le flux de travail terminé
• run(..., stream=True): retourne un itérable asynchrone d'objets AgentResponseUpdate à mesure que le workflow s’exécute, fournissant des mises à jour en temps réel.

Pendant l’exécution, les événements de flux de travail internes sont mappés aux réponses de l’agent comme suit :

Cette conversion vous permet d’utiliser l’interface de l’agent standard tout en ayant accès aux informations détaillées du flux de travail si nécessaire.