Flussi di lavoro di Microsoft Agent Framework - Uso di flussi di lavoro come agenti

Requisiti

Per usare un flusso di lavoro come agente, l'executor di avvio del flusso di lavoro deve essere in grado di gestire IEnumerable<ChatMessage> come input. Questa operazione viene soddisfatta automaticamente quando si usano executor basati su agente creati con AsAIAgent.

Creare un agente del flusso di lavoro

Usare il AsAIAgent() metodo di estensione per convertire qualsiasi flusso di lavoro compatibile in un agente:

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

Parametri AsAIAgent

Uso degli agenti del flusso di lavoro

Creazione di una sessione

Ogni conversazione con un agente del flusso di lavoro richiede una sessione per gestire lo stato:

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

Esecuzione non in streaming

Per casi d'uso semplici in cui si vuole ottenere la risposta completa:

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

Esecuzione in streaming

Per gli aggiornamenti in tempo reale durante l'esecuzione del flusso di lavoro:

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

Gestione delle richieste di input esterne

Quando un flusso di lavoro contiene executor che richiedono input esterno (usando RequestInfoExecutor), queste richieste vengono visualizzate come chiamate di funzione nella risposta dell'agente:

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

Serializzazione e ripresa delle sessioni

Le sessioni dell'agente del flusso di lavoro possono essere serializzate per la persistenza e riprese in un secondo momento:

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

Requisiti

Per usare un flusso di lavoro come agente, il gestore dell'avvio del flusso di lavoro deve essere in grado di gestire l'input dei messaggi. Questa condizione è soddisfatta automaticamente quando si usano esecutori basati su Agent o agenti.

Creare un agente del flusso di lavoro

Chiamare as_agent() su qualsiasi flusso di lavoro compatibile per convertirlo in un agente:

from agent_framework.foundry import FoundryChatClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential

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

Parametri as_agent

Uso degli agenti del flusso di lavoro

Creazione di una sessione

Facoltativamente, è possibile creare una sessione per gestire lo stato della conversazione tra più turni:

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

Esecuzione non in streaming

Per casi d'uso semplici in cui si vuole ottenere la risposta completa:

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

Esecuzione in streaming

Per gli aggiornamenti in tempo reale durante l'esecuzione del flusso di lavoro:

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

Gestione delle richieste di input esterne

Quando un flusso di lavoro contiene executor che richiedono input esterno (usando request_info), queste richieste vengono visualizzate come chiamate di funzione nella risposta dell'agente. La chiamata di funzione usa il nome 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

Fornire risposte alle richieste in sospeso

Per continuare l'esecuzione del flusso di lavoro dopo una richiesta di input esterna, creare un risultato della funzione e inviarlo di nuovo:

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

Esempio completo

Ecco un esempio completo che illustra un agente del flusso di lavoro con output di streaming:

import asyncio
import os

from agent_framework.foundry import FoundryChatClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential


async def main():
    # Set up the chat client
    client = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        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())

Informazioni sulla conversione degli eventi

Quando un flusso di lavoro viene eseguito come agente, gli eventi del flusso di lavoro vengono convertiti in risposte dell'agente. Il tipo di risposta dipende da come chiami run().

• run(): restituisce un oggetto AgentResponse contenente il risultato completo al termine del flusso di lavoro
• run(..., stream=True): restituisce un oggetto iterabile asincrono di oggetti AgentResponseUpdate man mano che il flusso di lavoro viene eseguito, fornendo aggiornamenti in tempo reale

as_agent() inoltra sia "output" (terminale) sia "intermediate" gli eventi al chiamante. Il set di tipi di evento inoltrati è AGENT_FORWARDED_EVENT_TYPES = {"output", "intermediate"}. Tutti gli altri eventi interni al flusso di lavoro vengono eliminati.

Durante l'esecuzione, gli eventi interni del flusso di lavoro vengono mappati alle risposte dell'agente come indicato di seguito:

Questa conversione consente di usare l'interfaccia dell'agente standard, pur avendo accesso a informazioni dettagliate sul flusso di lavoro quando necessario. La proprietà .text sia in AgentResponse sia in AgentResponseUpdate restituisce solo la risposta finale ("output"); esamina gli elementi di contenuto text_reasoning per accedere allo stato di avanzamento intermedio.

Go incapsula i flussi di lavoro come agenti con workflow/agentworkflow. In questo modo i chiamanti usano le NORMALI API di esecuzione dell'agente mentre il provider esegue il flusso di lavoro in background.

Requisiti

L'esecutore iniziale del flusso di lavoro deve accettare []*message.Message. Gli esecutori di agenti ospitati e gli esecutori configurati con messageworkflow.Configure soddisfano questo requisito.

Creare un agente del flusso di lavoro

Usare agentworkflow.New per eseguire il wrapping di qualsiasi flusso di lavoro compatibile come agente:

wfAgent, err := agentworkflow.New(wf, agentworkflow.AgentConfig{
    IncludeOutputsInResponse: true,
    Config: agent.Config{
        Name: "WorkflowAgent",
    },
})
if err != nil {
    return err
}

Parametri di agentworkflow.AgentConfig

Uso degli agenti del flusso di lavoro

Creazione di una sessione

Creare una sessione dell'agente quando si vuole rendere persistente lo stato del flusso di lavoro tra turni:

session, err := wfAgent.CreateSession(ctx)
if err != nil {
    return err
}

Esecuzione non in streaming

Usare RunText o Run e raccogliere la risposta per l'esecuzione non in streaming:

response, err := wfAgent.RunText(ctx, "Analyze this", agent.WithSession(session)).Collect()
if err != nil {
    return err
}
fmt.Println(response.String())

Esecuzione in streaming

Per gli aggiornamenti in tempo reale durante l'esecuzione del flusso di lavoro:

for update, err := range wfAgent.RunText(ctx, "Analyze this", agent.WithSession(session), agent.Stream(true)) {
    if err != nil {
        return err
    }
    fmt.Print(update.String())
}

Gestione delle richieste di input esterne

Le richieste esterne dal flusso di lavoro vengono visualizzate come contenuto della chiamata di funzione nella risposta dell'agente. Esaminare i messaggi di risposta per il contenuto della richiesta e inviare la risposta corrispondente in un'esecuzione successiva.

var requestCall *message.FunctionCallContent
for content := range response.Contents() {
    if call, ok := content.(*message.FunctionCallContent); ok {
        requestCall = call
        break
    }
}

Fornire risposte alle richieste in sospeso

Per continuare l'esecuzione del flusso di lavoro, restituire il contenuto della risposta corrispondente all'agente del flusso di lavoro:

result := &message.FunctionResultContent{
    CallID: requestCall.CallID,
    Result: "approved",
}

response, err = wfAgent.Run(
    ctx,
    []*message.Message{{
        Role:     message.RoleTool,
        Contents: []message.Content{result},
    }},
    agent.WithSession(session),
).Collect()
if err != nil {
    return err
}

Serializzazione e ripresa delle sessioni

Le sessioni dell'agente del flusso di lavoro possono essere serializzate per la persistenza e riprese in un secondo momento:

// Serialize the session state.
serializedSession, err := json.Marshal(session)
if err != nil {
    return err
}

// Store serializedSession to your persistence layer...

// Later, resume the session.
var resumedSession agent.Session
if err := json.Unmarshal(serializedSession, &resumedSession); err != nil {
    return err
}

for update, err := range wfAgent.RunText(ctx, "Continue the article", agent.WithSession(&resumedSession), agent.Stream(true)) {
    if err != nil {
        return err
    }
    fmt.Print(update.String())
}

Esempio completo

L'esempio seguente crea un workflow della pipeline di contenuti, lo incapsula come agente e trasmette le risposte tramite la normale API dell'agente:

package main

import (
    "cmp"
    "context"
    "fmt"
    "log"
    "os"

    "github.com/microsoft/agent-framework-go/agent"
    "github.com/microsoft/agent-framework-go/provider/foundryprovider"
    "github.com/microsoft/agent-framework-go/workflow/agentworkflow"

    "github.com/Azure/azure-sdk-for-go/sdk/azidentity"
)

func main() {
    ctx := context.Background()
    endpoint := os.Getenv("FOUNDRY_PROJECT_ENDPOINT")
    model := cmp.Or(os.Getenv("FOUNDRY_MODEL"), "gpt-4o-mini")

    credential, err := azidentity.NewDefaultAzureCredential(nil)
    if err != nil {
        log.Fatal(err)
    }

    researcher := foundryprovider.NewAgent(endpoint, credential, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
        Instructions: "Research and gather information on the given topic.",
        Config:      agent.Config{Name: "Researcher"},
    })
    writer := foundryprovider.NewAgent(endpoint, credential, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
        Instructions: "Write clear, engaging content based on research.",
        Config:      agent.Config{Name: "Writer"},
    })
    reviewer := foundryprovider.NewAgent(endpoint, credential, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
        Instructions: "Review the content and provide a final polished version.",
        Config:      agent.Config{Name: "Reviewer"},
    })

    wf, err := agentworkflow.NewSequentialWorkflowBuilder(researcher, writer, reviewer).
        WithName("content-pipeline").
        Build()
    if err != nil {
        log.Fatal(err)
    }

    wfAgent, err := agentworkflow.New(wf, agentworkflow.AgentConfig{
        IncludeOutputsInResponse: true,
        Config: agent.Config{
            Name: "Content Pipeline Agent",
        },
    })
    if err != nil {
        log.Fatal(err)
    }

    session, err := wfAgent.CreateSession(ctx)
    if err != nil {
        log.Fatal(err)
    }

    for update, err := range wfAgent.RunText(ctx, "Write about quantum computing", agent.WithSession(session), agent.Stream(true)) {
        if err != nil {
            log.Fatal(err)
        }
        if text := update.String(); text != "" {
            fmt.Print(text)
        }
    }
}