Microsoft Agent Framework-werkstromen - Werkstromen gebruiken als agents

Requirements

Om een werkstroom als agent te gebruiken, moet de startuitvoerder van de werkstroom IEnumerable<ChatMessage> als invoer kunnen verwerken. Dit is automatisch voldaan wanneer je agent-gebaseerde uitvoerders gebruikt die zijn gemaakt met AsAIAgent.

Een werkstroomagent maken

Gebruik de AsAIAgent() extensiemethode om elke compatibele werkstroom te converteren naar een 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"
);

AsAIAgent-parameters

Werkstroomagents gebruiken

Een sessie maken

Voor elk gesprek met een werkstroomagent is een sessie vereist om de status te beheren:

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

Uitvoering van niet-streaming

Voor eenvoudige gebruiksvoorbeelden waarin u het volledige antwoord wilt:

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

Streaming-uitvoering

Voor realtime-updates terwijl de workflow wordt uitgevoerd:

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

Aanvragen voor externe invoer verwerken

Wanneer een werkstroom uitvoerders bevat die externe invoer aanvragen (met behulp RequestInfoExecutor), worden deze aanvragen weergegeven als functie-aanroepen in het agentantwoord:

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

Serialisatie en hervatting van sessies

Werkstroomagentsessies kunnen worden geserialiseerd voor persistentie en later hervat:

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

Requirements

Als u een werkstroom als agent wilt gebruiken, moet de startuitvoerfunctie van de werkstroom berichtinvoer kunnen verwerken. Dit wordt automatisch voldaan als u Agent of agentgebaseerde uitvoerders gebruikt.

Een werkstroomagent maken

Roep as_agent() een compatibele werkstroom aan om deze te converteren naar een agent:

from agent_framework.foundry import FoundryChatClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential

# Create your chat client and agents
client = FoundryChatClient(
    project_endpoint="<your-endpoint>",
    model="<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")

parameters voor as_agent

Werkstroomagents gebruiken

Een sessie maken

U kunt eventueel een sessie maken om de gespreksstatus over meerdere beurten te beheren.

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

Uitvoering van niet-streaming

Voor eenvoudige gebruiksvoorbeelden waarin u het volledige antwoord wilt:

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

Streaming-uitvoering

Voor realtime-updates terwijl de workflow wordt uitgevoerd:

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

Aanvragen voor externe invoer verwerken

Wanneer een werkstroom uitvoerders bevat die externe invoer aanvragen met behulp van request_info, worden deze aanvragen weergegeven als functie-aanroepen in het agentantwoord. De functie-aanroep gebruikt de naam 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

Antwoorden geven op aanvragen die in behandeling zijn

Als u de uitvoering van de werkstroom wilt voortzetten na een externe invoeraanvraag, maakt u een functieresultaat en stuurt u deze terug:

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

Volledig voorbeeld

Hier volgt een volledig voorbeeld waarin een werkstroomagent met streaming-uitvoer wordt gedemonstreerd:

import asyncio
import os

from agent_framework.foundry import FoundryChatClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential


async def main():
    # Set up the chat client
    client = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        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())

Begrijpen van gebeurtenisconversie

Wanneer een werkstroom wordt uitgevoerd als agent, worden werkstroomgebeurtenissen geconverteerd naar agentreacties. Het type antwoord is afhankelijk van hoe u aanroept run():

• run(): retourneert een AgentResponse met het volledige resultaat nadat de werkstroom is voltooid
• run(..., stream=True): retourneert een asynchrone itereerbaar object AgentResponseUpdate terwijl de werkstroom wordt uitgevoerd, waardoor realtime-updates worden geboden

Tijdens de uitvoering worden interne werkstroomgebeurtenissen als volgt toegewezen aan agentreacties:

Met deze conversie kunt u de standaardagentinterface gebruiken terwijl u nog steeds toegang hebt tot gedetailleerde werkstroomgegevens wanneer dat nodig is.