Microsoft Agent Framework-Workflows – Verwenden von Workflows als Agents

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

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

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:

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