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 agentbasierte Ausführungsumgebungen verwendet werden, die mit AsAIAgent erstellt wurden.

Erstellen eines Workflow-Agents

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

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

Verwenden von Workflow-Agents

Erstellen einer Sitzung

Jede Unterhaltung mit einem Workflow-Agent erfordert eine Sitzung zum Verwalten des Zustands:

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

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

AgentResponse response = await workflowAgent.RunAsync(messages, session);

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

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

Serialisierung und Wiederaufnahme der Sitzung

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

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

Anforderungen

Um einen Workflow als Agent zu verwenden, muss der Startausführer des Workflows in der Lage sein, die Nachrichteneingabe zu verarbeiten. Dies wird automatisch erfüllt, wenn Agent oder agentenbasierte Executor verwendet werden.

Erstellen eines Workflow-Agents

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

from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential

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

as_agent Parameter

Verwenden von Workflow-Agents

Erstellen einer Sitzung

Sie können optional eine Sitzung erstellen, um den Gesprächszustand über mehrere Dialogrunden hinweg zu verwalten.

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

Nicht-Streaming-Ausführung

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

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

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

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

Behandeln externer Eingabeanforderungen

Wenn ein Workflow Executoren enthält, die externe Eingabe anfordern (mithilfe request_info), werden diese Anforderungen als Funktionsaufrufe in der Agentantwort angezeigt. Der Funktionsaufruf verwendet den Namen 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

Bereitstellen von Antworten auf ausstehende Anforderungen

Um die Workflowausführung nach einer externen Eingabeanforderung fortzusetzen, erstellen Sie ein Funktionsergebnis, und senden Sie ihn zurück:

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

Vollständiges Beispiel

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

import asyncio
import os

from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential


async def main():
    # Set up the chat client
    client = AzureOpenAIResponsesClient(
        project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
        deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
        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())

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, wie Sie anrufen run():

• run(): Gibt ein AgentResponse Ergebnis zurück, das das vollständige Ergebnis enthält, nachdem der Workflow abgeschlossen wurde.
• run(..., stream=True): Gibt eine asynchrone Iterable von AgentResponseUpdate Objekten zurück, während der Workflow ausgeführt wird und Echtzeitaktualisierungen bereitgestellt werden.

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.