Delen via

Microsoft Agent Framework-werkstromen - Werkstromen gebruiken als agents

Dit document bevat een overzicht van het gebruik van werkstromen als agents in Microsoft Agent Framework.

Overzicht

Soms hebt u een geavanceerde werkstroom gemaakt met meerdere agents, aangepaste uitvoerders en complexe logica, maar u wilt deze net als elke andere agent gebruiken. Dat is precies wat werkstroomagents u laten doen. Door uw werkstroom als een Agentwerkstroom te verpakken, kunt u ermee werken via dezelfde vertrouwde API die u zou gebruiken voor een eenvoudige chatagent.

Belangrijkste voordelen

  • Unified Interface: Interactie met complexe werkstromen met dezelfde API als eenvoudige agents
  • API-compatibiliteit: werkstromen integreren met bestaande systemen die ondersteuning bieden voor de agentinterface
  • Samenstelbaarheid: Werkstroomagents gebruiken als bouwstenen in grotere agentsystemen of andere werkstromen
  • Threadbeheer: gebruik agentthreads voor gespreksstatus, controlepunten en hervatting
  • Streaming-ondersteuning: updates in real-time ontvangen wanneer de werkstroom wordt uitgevoerd

Hoe het werkt

Wanneer u een werkstroom converteert naar een agent:

  1. De werkstroom wordt gevalideerd om ervoor te zorgen dat de startuitvoerer chatberichten kan accepteren
  2. Er wordt een thread gemaakt om de gespreksstatus en controlepunten te beheren
  3. Invoerberichten worden doorgestuurd naar de startuitvoerer van de werkstroom
  4. Werkstroomgebeurtenissen worden geconverteerd naar antwoordupdates van agents
  5. Externe invoeraanvragen (van RequestInfoExecutor) worden weergegeven als functie-aanroepen

Requirements

Om een werkstroom als agent te gebruiken, moet de startuitvoerder van de werkstroom IEnumerable<ChatMessage> als invoer kunnen verwerken. Dit wordt automatisch voldaan bij het gebruik van ChatClientAgent of andere agentgebaseerde verwerkers.

Een werkstroomagent maken

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

Kenmerk Typologie Description
id string? Optionele unieke id voor de agent. Automatisch gegenereerd als deze niet is opgegeven.
name string? Optionele weergavenaam voor de agent.
description string? Optionele beschrijving van het doel van de agent.
checkpointManager CheckpointManager? Optioneel controlepuntbeheer voor persistentie tussen sessies.
executionEnvironment IWorkflowExecutionEnvironment? Optionele uitvoeringsomgeving. Standaardinstelling is InProcessExecution.OffThread of InProcessExecution.Concurrent op basis van de configuratie van de werkstroom.

Werkstroomagents gebruiken

Een thread maken

Voor elk gesprek met een werkstroomagent is een thread vereist voor het beheren van de status:

// Create a new thread for the conversation
AgentThread thread = workflowAgent.GetNewThread();

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

AgentRunResponse response = await workflowAgent.RunAsync(messages, thread);

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 (AgentRunResponseUpdate update in workflowAgent.RunStreamingAsync(messages, thread))
{
    // 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 (AgentRunResponseUpdate update in workflowAgent.RunStreamingAsync(messages, thread))
{
    // 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
        }
    }
}

Threadserialisatie en hervatting

Threads van workflow-agents kunnen worden geserialiseerd voor opslag en later hervat.

// Serialize the thread state
JsonElement serializedThread = thread.Serialize();

// Store serializedThread to your persistence layer...

// Later, resume the thread
AgentThread resumedThread = workflowAgent.DeserializeThread(serializedThread);

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

Controlepunten met werkstroomagents

Schakel controlepunten in om de werkstroomstatus bij het opnieuw opstarten van processen te behouden:

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

Requirements

Om een werkstroom als agent te gebruiken, moet de startuitvoerder van de werkstroom list[ChatMessage] als invoer kunnen verwerken. Dit wordt automatisch voldaan wanneer u deze gebruikt ChatAgent of AgentExecutor.

Een werkstroomagent maken

Roep as_agent() een compatibele werkstroom aan om deze te converteren naar een 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")

parameters voor as_agent

Kenmerk Typologie Description
name str | None Optionele weergavenaam voor de agent. Automatisch gegenereerd als deze niet is opgegeven.

Werkstroomagents gebruiken

Een thread maken

Voor elk gesprek met een werkstroomagent is een thread vereist voor het beheren van de status:

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

Uitvoering van niet-streaming

Voor eenvoudige gebruiksvoorbeelden waarin u het volledige antwoord wilt:

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

Streaming-uitvoering

Voor realtime-updates terwijl de workflow wordt uitgevoerd:

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)

Aanvragen voor externe invoer verwerken

Wanneer een werkstroom uitvoerders bevat die externe invoer aanvragen (met behulp van RequestInfoExecutor), worden deze aanvragen weergegeven als functie-aanroepen. De werkstroomagent houdt aanvragen bij die in behandeling zijn en verwacht antwoorden voordat u doorgaat:

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

Antwoorden geven op aanvragen die in behandeling zijn

Werkstroomuitvoering voortzetten na een externe invoeraanvraag:

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

Volledig voorbeeld

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

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

Begrijpen van gebeurtenisconversie

Wanneer een werkstroom wordt uitgevoerd als agent, worden werkstroomgebeurtenissen geconverteerd naar agentreacties. Het type antwoord is afhankelijk van de methode die u gebruikt:

  • run(): retourneert een AgentRunResponse met het volledige resultaat nadat de werkstroom is voltooid
  • run_stream(): Genereert AgentRunResponseUpdate objecten terwijl de werkstroom wordt uitgevoerd en biedt realtime updates.

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

Werkstroom gebeurtenis Antwoord van agent
AgentRunUpdateEvent Doorgegeven als AgentRunResponseUpdate (streaming) of samengevoegd in AgentRunResponse (niet-streaming)
RequestInfoEvent Geconverteerd naar FunctionCallContent en FunctionApprovalRequestContent
Andere gebeurtenissen Opgenomen in raw_representation voor waarneembaarheid

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

Gebruiksvoorbeelden

1. Complexe agentpijplijnen

Een werkstroom met meerdere agents verpakken als één agent voor gebruik in toepassingen:

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

2. Samenstelling van agent

Werkstroomagents gebruiken als onderdelen in grotere systemen:

  • Een werkstroomagent kan worden gebruikt als een hulpprogramma door een andere agent
  • Meerdere werkstroomagents kunnen samen worden georkestreerd
  • Werkstroomagents kunnen worden genest binnen andere werkstromen

3. API-integratie

Maak complexe werkstromen beschikbaar via API's die de standaardagentinterface verwachten, waardoor:

  • Chatinterfaces die gebruikmaken van geavanceerde back-endwerkstromen
  • Integratie met bestaande op agents gebaseerde systemen
  • Geleidelijke migratie van eenvoudige agents naar complexe werkstromen

Volgende stappen

  • Last updated on