Microsoft Agent Framework-arbetsflöden – Använda arbetsflöden som agenter

Det här dokumentet innehåller en översikt över hur du använder arbetsflöden som agenter i Microsoft Agent Framework.

Översikt

Ibland har du skapat ett avancerat arbetsflöde med flera agenter, anpassade utförare och komplex logik – men du vill använda det precis som andra agenter. Det är precis vad arbetsflödesagenter låter dig göra. Genom att omsluta arbetsflödet som en Agentkan du interagera med det via samma välbekanta API som du skulle använda för en enkel chattagent.

Viktiga fördelar

• Enhetligt gränssnitt: Interagera med komplexa arbetsflöden med samma API som enkla agenter
• API-kompatibilitet: Integrera arbetsflöden med befintliga system som stöder agentgränssnittet
• Sammansättning: Använd arbetsflödesagenter som byggstenar i större agentsystem eller i andra arbetsflöden
• Sessionshantering: Utnyttja agentsessioner för konversationstillstånd, kontrollpunkter och återupptagande
• Stöd för direktuppspelning: Hämta realtidsuppdateringar när arbetsflödet körs

Så här fungerar det

När du konverterar ett arbetsflöde till en agent:

1. Arbetsflödet verifieras för att säkerställa att startexekutor kan acceptera chattmeddelanden
2. En session skapas för att hantera konversationstillstånd och kontrollpunkter
3. Indatameddelanden dirigeras till arbetsflödets startexekutor
4. Arbetsflödeshändelser konverteras till agentsvarsuppdateringar
5. Externa indatabegäranden (från RequestInfoExecutor) visas som funktionsanrop

Kravspecifikation

Om du vill använda ett arbetsflöde som agent måste arbetsflödets startexekutor kunna hantera IEnumerable<ChatMessage> som indata. Detta uppfylls automatiskt när du använder ChatClientAgent eller andra agentbaserade utförare.

Skapa en arbetsflödesagent

AsAgent() Använd tilläggsmetoden för att konvertera alla kompatibla arbetsflöden till en agent:

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

// First, build your workflow
var workflow = AgentWorkflowBuilder
    .CreateSequentialPipeline(researchAgent, writerAgent, reviewerAgent)
    .Build();

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

AsAgent-parametrar

Parameter	Typ	Description
id	string?	Valfri unik identifierare för agenten. Genereras automatiskt om det inte tillhandahålls.
name	string?	Valfritt visningsnamn för agenten.
description	string?	Valfri beskrivning av agentens syfte.
checkpointManager	CheckpointManager?	Valfri kontrollpunktshanterare för beständighet mellan sessioner.
executionEnvironment	IWorkflowExecutionEnvironment?	Valfri körningsmiljö. Standardvärdet är InProcessExecution.OffThread eller InProcessExecution.Concurrent som baseras på arbetsflödeskonfiguration.

Använda arbetsflödesagenter

Skapa en session

Varje konversation med en arbetsflödesagent kräver en session för att hantera tillstånd:

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

Körning utan direktuppspelning

För enkla användningsfall där du vill ha ett fullständigt svar:

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

Körning av direktuppspelning

För realtidsuppdateringar när arbetsflödet körs:

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

Hantera externa indatabegäranden

När ett arbetsflöde innehåller utförare som begär externa indata (med RequestInfoExecutor), visas dessa begäranden som funktionsanrop i agentsvaret:

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

Seriellisering och återupptagande av sessioner

Arbetsflödesagentsessioner kan serialiseras för beständighet och återupptas senare:

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

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

Kontrollpunktering med arbetsflödesagenter

Aktivera kontrollpunkter för att bevara arbetsflödestillståndet mellan processomstarter:

// Create a checkpoint manager with your storage backend
var checkpointManager = new CheckpointManager(new FileCheckpointStorage("./checkpoints"));

// Create workflow agent with checkpointing enabled
AIAgent workflowAgent = workflow.AsAgent(
    id: "persistent-workflow",
    name: "Persistent Workflow Agent",
    checkpointManager: checkpointManager
);

Kravspecifikation

Om du vill använda ett arbetsflöde som agent måste arbetsflödets startexekutor kunna hantera list[ChatMessage] som indata. Detta uppfylls automatiskt när du använder ChatAgent eller AgentExecutor.

Skapa en arbetsflödesagent

Anropa as_agent() alla kompatibla arbetsflöden för att konvertera det till en agent:

from agent_framework import WorkflowBuilder, ChatAgent
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

# Create your chat client and agents
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())

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

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

# Build your workflow
workflow = (
    WorkflowBuilder()
    .set_start_executor(researcher)
    .add_edge(researcher, writer)
    .build()
)

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

as_agent parametrar

Parameter	Typ	Description
name	str | None	Valfritt visningsnamn för agenten. Genereras automatiskt om det inte tillhandahålls.

Använda arbetsflödesagenter

Skapa en tråd

Varje konversation med en arbetsflödesagent kräver en tråd för att hantera tillstånd:

# Create a new thread for the conversation
thread = workflow_agent.get_new_thread()

Körning utan direktuppspelning

För enkla användningsfall där du vill ha ett fullständigt svar:

from agent_framework import ChatMessage, Role

messages = [ChatMessage(role=Role.USER, content="Write an article about AI trends")]

response = await workflow_agent.run(messages, thread=thread)

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

Körning av direktuppspelning

För realtidsuppdateringar när arbetsflödet körs:

messages = [ChatMessage(role=Role.USER, content="Write an article about AI trends")]

async for update in workflow_agent.run_stream(messages, thread=thread):
    # Process streaming updates from each agent in the workflow
    if update.text:
        print(update.text, end="", flush=True)

Hantera externa indatabegäranden

När ett arbetsflöde innehåller utförare som begär externa indata (med hjälp av RequestInfoExecutor) visas dessa begäranden som funktionsanrop. Arbetsflödesagenten spårar väntande begäranden och förväntar sig svar innan du fortsätter:

from agent_framework import (
    FunctionCallContent,
    FunctionApprovalRequestContent,
    FunctionApprovalResponseContent,
)

async for update in workflow_agent.run_stream(messages, thread=thread):
    for content in update.contents:
        if isinstance(content, FunctionApprovalRequestContent):
            # The workflow is requesting external input
            request_id = content.id
            function_call = content.function_call

            print(f"Workflow requests input: {function_call.name}")
            print(f"Request data: {function_call.arguments}")

            # Store the request_id to provide a response later

# Check for pending requests
if workflow_agent.pending_requests:
    print(f"Pending requests: {list(workflow_agent.pending_requests.keys())}")

Tillhandahålla svar på väntande begäranden

Så här fortsätter du arbetsflödeskörningen efter en extern indatabegäran:

# Create a response for the pending request
response_content = FunctionApprovalResponseContent(
    id=request_id,
    function_call=function_call,
    approved=True,
)

response_message = ChatMessage(
    role=Role.USER,
    contents=[response_content],
)

# Continue the workflow with the response
async for update in workflow_agent.run_stream([response_message], thread=thread):
    if update.text:
        print(update.text, end="", flush=True)

Fullständigt exempel

Här är ett fullständigt exempel som visar en arbetsflödesagent med strömmande utdata:

import asyncio
from agent_framework import (
    ChatAgent,
    ChatMessage,
    Role,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework._workflows import SequentialBuilder
from azure.identity import AzureCliCredential


async def main():
    # Set up the chat client
    chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())

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

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

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

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

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

    # Create a thread and run the workflow
    thread = workflow_agent.get_new_thread()
    messages = [ChatMessage(role=Role.USER, content="Write about quantum computing")]

    print("Starting workflow...")
    print("=" * 60)

    current_author = None
    async for update in workflow_agent.run_stream(messages, thread=thread):
        # 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())

Att förstå händelsekonvertering

När ett arbetsflöde körs som en agent konverteras arbetsflödeshändelser till agentsvar. Typen av svar beror på vilken metod du använder:

• run(): Returnerar ett AgentResponse innehållande det fullständiga resultatet när arbetsflödet har slutförts
• run_stream(): Ger AgentResponseUpdate objekt när arbetsflödet körs, vilket ger realtidsuppdateringar

Under körning mappas interna arbetsflödeshändelser till agentsvar på följande sätt:

Arbetsflödeshändelse	Agentsvar
AgentResponseUpdateEvent	Skickas som AgentResponseUpdate (direktuppspelning) eller aggregeras till AgentResponse (icke-direktuppspelning)
RequestInfoEvent	Konverteras till FunctionCallContent och FunctionApprovalRequestContent
Andra händelser	Ingår i raw_representation för observerbarhet

Med den här konverteringen kan du använda standardagentgränssnittet samtidigt som du har åtkomst till detaljerad arbetsflödesinformation när det behövs.

Användningsfall

Komplexa agent-pipelines

Omsluta ett arbetsflöde med flera agenter som en enda agent för användning i program:

User Request --> [Workflow Agent] --> Final Response
                      |
                      +-- Researcher Agent
                      +-- Writer Agent  
                      +-- Reviewer Agent

2. Agentsammansättning

Använd arbetsflödesagenter som komponenter i större system:

• En arbetsflödesagent kan användas som ett verktyg av en annan agent
• Flera arbetsflödesagenter kan samordnas tillsammans
• Arbetsflödesagenter kan kapslas i andra arbetsflöden

3. API-integrering

Exponera komplexa arbetsflöden via API:er som förväntar sig standardgränssnittet för agenten, vilket aktiverar:

• Chattgränssnitt som använder avancerade serverdelsarbetsflöden
• Integrering med befintliga agentbaserade system
• Gradvis migrering från enkla agenter till komplexa arbetsflöden

Nästa steg

• Lär dig hur du hanterar begäranden och svar i arbetsflöden
• Lär dig hur du hanterar tillstånd i arbetsflöden
• Lär dig hur du skapar kontrollpunkter och återupptar från dem
• Lär dig hur du övervakar arbetsflöden
• Lär dig mer om tillståndsisolering i arbetsflöden
• Lär dig hur du visualiserar arbetsflöden