Pracovní postupy rozhraní Microsoft Agent Framework – Použití pracovních postupů jako agentů

Požadavky

Pokud chcete jako agenta použít pracovní postup, musí být spouštěcí exekutor pracovního postupu schopný zpracovat IEnumerable<ChatMessage> jako vstup. To je automaticky splněno při použití exekutorů založených na agentech vytvořených pomocí AsAIAgent.

Vytvoření agenta pracovního postupu

AsAIAgent() Pomocí metody rozšíření můžete převést jakýkoli kompatibilní pracovní postup na agenta:

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

Parametry AsAIAgent

Použití agentů pracovního postupu

Vytvoření relace

Každá konverzace s agentem pracovního postupu vyžaduje relaci pro správu stavu:

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

Provádění bez streamování

V případě jednoduchých případů použití, kdy chcete úplnou odpověď:

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

Spouštění streamování

Aktualizace v reálném čase při provádění pracovního postupu:

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

Zpracování externích vstupních požadavků

Pokud pracovní postup obsahuje exekutory, které vyžadují externí vstup (pomocí RequestInfoExecutor), zobrazí se tyto požadavky jako volání funkce v odpovědi agenta:

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

Serializace a obnovení relace

Relace agenta pracovního postupu lze serializovat pro trvalost a obnovit později:

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

Požadavky

Pokud chcete jako agenta použít pracovní postup, musí spouštěcí exekutor pracovního postupu zpracovávat vstup zpráv. To se automaticky splní při použití Agent exekutorů nebo exekutorů založených na agentech.

Vytvoření agenta pracovního postupu

Zavolejte as_agent() na libovolný kompatibilní pracovní postup pro jeho převedení na agenta:

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

parametry as_agent

Použití agentů pracovního postupu

Vytvoření relace

Můžete podle potřeby vytvořit relaci pro správu stavu konverzace napříč několika interakcemi:

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

Provádění bez streamování

V případě jednoduchých případů použití, kdy chcete úplnou odpověď:

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

Spouštění streamování

Aktualizace v reálném čase při provádění pracovního postupu:

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

Zpracování externích vstupních požadavků

Pokud pracovní postup obsahuje exekutory, které vyžadují externí vstup (pomocí request_info), zobrazí se tyto požadavky jako volání funkce v odpovědi agenta. Volání funkce používá název 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

Poskytování odpovědí na čekající žádosti

Pokud chcete pokračovat v provádění pracovního postupu po externím vstupním požadavku, vytvořte výsledek funkce a odešlete ho zpět:

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

Kompletní příklad

Zde je úplný příklad agenta pracovního postupu s streamovaným výstupem:

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

Principy převodu událostí

Když pracovní postup běží jako agent, události pracovního postupu se převedou na reakce agenta. Typ odpovědi závisí na tom, jak voláte run():

• run(): Vrátí AgentResponse po dokončení pracovního postupu celý výsledek.
• run(..., stream=True): Vrátí asynchronní iterovatelnost AgentResponseUpdate objektů při provádění pracovního postupu a poskytuje aktualizace v reálném čase.

as_agent() předává volajícímu jak "output" (terminál), tak události "intermediate". Sada předávaných typů událostí je AGENT_FORWARDED_EVENT_TYPES = {"output", "intermediate"}. Všechny ostatní interní události pracovního postupu se zahodí.

Během provádění se interní události pracovního postupu mapují na odpovědi agenta následujícím způsobem:

Tento převod umožňuje používat standardní rozhraní agenta a v případě potřeby mít přístup k podrobným informacím o pracovním postupu. Vlastnost .text u AgentResponse i AgentResponseUpdate vrací pouze konečnou odpověď ("output"); chcete-li získat přístup k průběžnému stavu, zkontrolujte položky obsahu v text_reasoning.

Go zabalí pracovní postupy jako agenty pomocí workflow/agentworkflow. Volajícím to umožní používat rozhraní API normálního agenta, zatímco poskytovatel spouští pracovní postup na pozadí.

Požadavky

Spouštěcí vykonavatel pracovního postupu musí přijmout []*message.Message. Tento požadavek splňují executory hostovaného agenta i executory nakonfigurované pomocí messageworkflow.Configure.

Vytvoření agenta pracovního postupu

Slouží agentworkflow.New k zabalení libovolného kompatibilního pracovního postupu jako agenta:

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

Parametry agentworkflow.AgentConfig

Použití agentů pracovního postupu

Vytvoření relace

Vytvořte relaci agenta, pokud chcete, aby se stav workflow zachoval mezi jednotlivými interakcemi:

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

Provádění bez streamování

Použijte RunText nebo Run a shromážděte odpověď pro provádění bez streamování:

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

Spouštění streamování

Aktualizace v reálném čase při provádění pracovního postupu:

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

Zpracování externích vstupních požadavků

Externí požadavky z pracovního postupu se zobrazují jako obsah volání funkce v odpovědi agenta. Zkontrolujte, zda zprávy odpovědí obsahují obsah požadavku, a při pozdějším spuštění odešlete odpovídající odpověď.

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

Poskytování odpovědí na čekající žádosti

Pokud chcete pokračovat v provádění pracovního postupu, vraťte odpovídající obsah odpovědi agentu pracovního postupu:

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
}

Serializace a obnovení relace

Relace agenta pracovního postupu lze serializovat pro trvalost a obnovit později:

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

Kompletní příklad

Následující příklad sestaví pracovní postup kanálu obsahu, zabalí ho jako agenta a streamuje odpovědi prostřednictvím normálního rozhraní API agenta:

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