Freigeben über

Microsoft Agent Framework-Workflows – Verwenden von Workflows als Agents

Dieses Dokument bietet eine Übersicht über die Verwendung von Workflows als Agents im Microsoft Agent Framework.

Überblick

Manchmal haben Sie einen anspruchsvollen Workflow mit mehreren Agents, benutzerdefinierten Executoren und komplexer Logik erstellt – möchten ihn jedoch wie jeden anderen Agenten verwenden. Workflow-Agenten ermöglichen es Ihnen genau das zu tun. Durch das Umschließen Ihres Workflows als Agent können Sie mit dem Workflow über dieselbe vertraute API interagieren, die Sie für einen einfachen Chat-Agenten verwenden.

Wichtige Vorteile

  • Einheitliche Schnittstelle: Interagieren mit komplexen Workflows mit derselben API wie einfache Agents
  • API-Kompatibilität: Integrieren von Workflows in vorhandene Systeme, die die Agent-Schnittstelle unterstützen
  • Komponierbarkeit: Die Verwendung von Workflow-Agenten als Bausteine in größeren Agentensystemen oder anderen Workflows
  • Threadverwaltung: Nutzen von Agentthreads für Unterhaltungszustand, Checkpoint und Wiederaufnahme
  • Streamingunterstützung: Abrufen von Echtzeitupdates während der Ausführung des Workflows

Funktionsweise

Wenn Sie einen Workflow in einen Agent konvertieren:

  1. Der Workflow wird validiert, um sicherzustellen, dass der Startprozess Chatnachrichten annehmen kann.
  2. Ein Thread wird erstellt, um den Gesprächszustand und die Checkpoints zu verwalten.
  3. Eingabemeldungen werden an den Startausführer des Workflows weitergeleitet.
  4. Workflow-Ereignisse werden in Agent-Antwort-Updates umgewandelt.
  5. Externe Eingabeanforderungen (von RequestInfoExecutor) werden als Funktionsaufrufe angezeigt.

Anforderungen

Um einen Workflow als Agent zu verwenden, muss der Startausführer des Workflows als Eingabe verarbeiten IEnumerable<ChatMessage> können. Dies wird automatisch erfüllt, wenn ChatClientAgent oder andere agentenbasierte Ausführungsprogramme verwendet werden.

Erstellen eines Workflow-Agents

Verwenden Sie die AsAgent() Erweiterungsmethode, um jeden kompatiblen Workflow in einen Agent zu konvertieren:

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

Parameter Typ Description
id string? Optionaler eindeutiger Bezeichner für den Agent. Wird automatisch generiert, falls nicht angegeben.
name string? Optionaler Anzeigename für den Agent.
description string? Optionale Beschreibung des Zwecks des Agenten.
checkpointManager CheckpointManager? Optionaler Prüfpunkt-Manager für Persistenz über Sitzungen hinweg.
executionEnvironment IWorkflowExecutionEnvironment? Optionale Ausführungsumgebung. Standardmäßig wird InProcessExecution.OffThread oder InProcessExecution.Concurrent basierend auf der Workflowkonfiguration verwendet.

Verwenden von Workflow-Agents

Erstellen eines Threads

Jede Unterhaltung mit einem Workflow-Agent erfordert einen Thread zum Verwalten des Zustands:

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

Nicht-Streaming-Ausführung

Für einfache Anwendungsfälle, in denen Sie die vollständige Antwort benötigen:

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-Ausführung

Für Echtzeitaktualisierungen während der Ausführung des Workflows:

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

Behandeln externer Eingabeanforderungen

Wenn ein Workflow Executoren enthält, die externe Eingabe anfordern (mithilfe RequestInfoExecutor), werden diese Anforderungen als Funktionsaufrufe in der Agentantwort angezeigt:

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

Thread-Serialisierung und Wiederaufnahme

Workflow-Agent-Threads können zur Persistenz serialisiert und später fortgesetzt werden:

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

Checkpointing mit Workflow-Agenten

Aktivieren Sie prüfpunkte, um den Workflowstatus über Prozessneustarts hinweg beizubehalten:

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

Anforderungen

Um einen Workflow als Agent zu verwenden, muss der Startausführer des Workflows als Eingabe verarbeiten list[ChatMessage] können. Dies wird bei der Verwendung von ChatAgent oder AgentExecutor automatisch erfüllt.

Erstellen eines Workflow-Agents

Rufen Sie as_agent() jeden kompatiblen Workflow auf, um ihn in einen Agent zu konvertieren:

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 Parameter

Parameter Typ Description
name str | None Optionaler Anzeigename für den Agent. Wird automatisch generiert, falls nicht angegeben.

Verwenden von Workflow-Agents

Erstellen eines Threads

Jede Unterhaltung mit einem Workflow-Agent erfordert einen Thread zum Verwalten des Zustands:

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

Nicht-Streaming-Ausführung

Für einfache Anwendungsfälle, in denen Sie die vollständige Antwort benötigen:

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-Ausführung

Für Echtzeitaktualisierungen während der Ausführung des Workflows:

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)

Behandeln externer Eingabeanforderungen

Wenn ein Workflow Executoren enthält, die externe Eingabe anfordern (mithilfe RequestInfoExecutor), werden diese Anforderungen als Funktionsaufrufe angezeigt. Der Workflow-Agent verfolgt ausstehende Anforderungen und erwartet Antworten, bevor sie fortfahren:

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

Bereitstellen von Antworten auf ausstehende Anforderungen

So setzen Sie die Workflowausführung nach einer externen Eingabeanforderung fort:

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

Vollständiges Beispiel

Hier ist ein vollständiges Beispiel, in dem ein Workflow-Agent mit Streamingausgabe veranschaulicht wird:

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

Verständnis der Ereigniskonvertierung

Wenn ein Workflow als Agent ausgeführt wird, werden Workflowereignisse in Agentantworten konvertiert. Die Art der Antwort hängt davon ab, welche Methode Sie verwenden:

  • run(): Gibt ein AgentRunResponse Ergebnis zurück, das das vollständige Ergebnis enthält, nachdem der Workflow abgeschlossen wurde.
  • run_stream(): Liefert AgentRunResponseUpdate Objekte während der Ausführung des Workflows und stellt Echtzeitaktualisierungen bereit.

Während der Ausführung werden interne Workflow-Ereignisse den Agentantworten wie folgt zugeordnet:

Workflow-Ereignis Antwort des Agenten
AgentRunUpdateEvent Übergeben als AgentRunResponseUpdate (Streaming) oder aggregiert in AgentRunResponse (nicht streaming)
RequestInfoEvent Konvertiert in FunctionCallContent und FunctionApprovalRequestContent
Andere Ereignisse Eingeschlossen in raw_representation zur Beobachtbarkeit

Mit dieser Konvertierung können Sie die Standard-Agent-Schnittstelle verwenden und gleichzeitig bei Bedarf auf detaillierte Workflowinformationen zugreifen.

Anwendungsfälle

1. Komplexe Agent-Pipelines

Einen Multi-Agenten-Workflow als einzelnen Agenten zusammenfassen zur Nutzung in Anwendungen.

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

2. Zusammensetzung des Agenten

Verwenden Sie Workflow-Agents als Komponenten in größeren Systemen:

  • Ein Workflow-Agent kann von einem anderen Agent als Tool verwendet werden.
  • Mehrere Workflow-Agents können zusammen orchestriert werden
  • Workflow-Agents können in anderen Workflows geschachtelt werden

3. API-Integration

Machen Sie komplexe Workflows über APIs verfügbar, die die Standard-Agent-Schnittstelle erwarten, und ermöglichen Sie Folgendes:

  • Chatschnittstellen, die komplexe Back-End-Workflows verwenden
  • Integration in vorhandene agentbasierte Systeme
  • Schrittweise Migration von einfachen Agents zu komplexen Workflows

Nächste Schritte

  • Last updated on