Microsoft Agent Framework-munkafolyamatok – Munkafolyamatok használata ügynökökként

Ez a dokumentum áttekintést nyújt a munkafolyamatok ügynökökként való használatáról a Microsoft Agent Frameworkben.

Áttekintés

Néha több ügynökkel, egyéni végrehajtóval és összetett logikával létrehozott egy kifinomult munkafolyamatot, de ugyanúgy szeretné használni, mint bármely más ügynököt. Ez az, amit pontosan megengednek a munkafolyamat-ügynökök neked. A munkafolyamatot Agent formájában csomagolva, ugyanazon a jól ismert API-n keresztül léphet vele interakcióba, amit egy egyszerű csevegőügynök esetén használna.

Főbb előnyök

  • Egyesített felület: Összetett munkafolyamatok használata ugyanazzal az API-val, mint az egyszerű ügynökök
  • API-kompatibilitás: Munkafolyamatok integrálása az ügynökfelületet támogató meglévő rendszerekkel
  • Kompatibilitás: Munkafolyamat-ügynökök használata építőelemként nagyobb ügynökrendszerekben vagy más munkafolyamatokban
  • Munkamenet-kezelés: Agent munkamenetek kihasználása beszélgetési állapothoz és újrakezdéshez
  • Streamelési támogatás: Valós idejű frissítések lekérése a munkafolyamat végrehajtásakor

Hogyan működik?

Amikor egy munkafolyamatot ügynökké alakít át:

  1. A munkafolyamat ellenőrzése biztosítja, hogy a kezdő végrehajtó elfogadja a szükséges bemeneti típusokat
  2. Munkamenet jön létre a beszélgetés állapotának kezeléséhez
  3. A bemeneti üzenetek a munkafolyamat kezdő végrehajtójának lesznek átirányítva
  4. A munkafolyamat-események ügynökválasz-frissítésekké alakulnak
  5. A külső bemeneti kérések (forrásból RequestInfoExecutor) függvényhívásokként jelennek meg

Requirements

A munkafolyamat ügynökként való használatához a munkafolyamat kezdő végrehajtójának képesnek kell lennie bemenetként kezelni IEnumerable<ChatMessage> . Ez automatikusan teljesül, ha ügynökalapú végrehajtókat AsAIAgenthasznál.

Munkafolyamat-ügynök létrehozása

AsAIAgent() A bővítménymetódus használatával bármilyen kompatibilis munkafolyamatot átalakíthat ügynökké:

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 paraméterek

Paraméter Típus Description
id string? Az ügynök opcionális egyedi azonosítója. Automatikusan létrejön, ha nincs megadva.
name string? Az ügynök megjelenítéséhez választható név.
description string? Az ügynök céljának nem kötelező leírása.
executionEnvironment IWorkflowExecutionEnvironment? Nem kötelező végrehajtási környezet. Alapértelmezetten InProcessExecution.OffThread vagy InProcessExecution.Concurrent, a munkafolyamat konfigurációja alapján.
includeExceptionDetails bool Ha true, a hibatartalom tartalmazhat kivételüzeneteket. Alapértelmezett érték: false.
includeWorkflowOutputsInResponse bool Ha true, átalakítja a kimenő munkafolyamat kimeneteit az ügynökválaszok tartalmává. Alapértelmezett érték: false.

Munkafolyamat-ügynökök használata

Munkamenet létrehozása

A munkafolyamat-ügynökkel folytatott minden beszélgetéshez munkamenet szükséges az állapot kezeléséhez:

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

Nem-streamelő végrehajtás

Olyan egyszerű használati esetekben, amikor a teljes választ szeretné kapni:

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

Streamelés végrehajtása

Valós idejű frissítések a munkafolyamat végrehajtásakor:

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

Külső bemeneti kérések kezelése

Ha egy munkafolyamat olyan végrehajtókat tartalmaz, amelyek külső bemenetet kérnek (a használatával RequestInfoExecutor), ezek a kérések függvényhívásokként jelennek meg az ügynök válaszában:

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

Munkamenet szerializálása és újraindítása

A munkafolyamat-ügynök munkamenetei szerializálhatók az adatmegőrzéshez, és később folytathatók:

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

Important

A szerializált munkafolyamat-ügynök munkamenet tartalmazza a belső munkafolyamat-ellenőrzőpontot. Ha az alkalmazás a munkamenet deszerializálása vagy futtatása előtt rekonstruálja a munkafolyamatot, minden belső ügynököt ugyanazzal ChatClientAgentOptions.Id (és ha Name van beállítva, ugyanazzal Name) kell újra létrehozni.

Az id átadott fájl workflow.AsAIAgent(...) csak a külső munkafolyamat-ügynököt azonosítja. Nem stabilizálja az ügynökök végrehajtói identitásait a munkafolyamaton belül. A konfigurációs útmutatásért lásd: Rehydrating from Checkpoints.

Requirements

A munkafolyamat ügynökként való használatához a munkafolyamat kezdő végrehajtójának képesnek kell lennie az üzenetbevitel kezelésére. Ez automatikusan teljesül az ügynökalapú végrehajtók használatakor Agent .

Munkafolyamat-ügynök létrehozása

Bármely kompatibilis munkafolyamat meghívása as_agent() ügynökké alakításához:

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

as_agent paraméterek

Paraméter Típus Description
name str | None Az ügynök megjelenítéséhez választható név. Automatikusan létrejön, ha nincs megadva.

Munkafolyamat-ügynökök használata

Munkamenet létrehozása

Igény szerint létrehozhat egy munkamenetet, amellyel több fordulón keresztül kezelheti a beszélgetés állapotát:

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

Megjegyzés:

A munkamenetek választhatóak. Ha nem ad át egy -t -nek, az ügynök belsőleg kezeli az állapotot. Ha workflow.as_agent() létrejön context_providers nélkül, a keretrendszer alapértelmezés szerint hozzáad egy InMemoryHistoryProvider()-t, így a többfordulós előzmények automatikusan működnek. Ha context_providers kifejezetten átadja, a lista változtatás nélkül lesz használva.

Nem-streamelő végrehajtás

Olyan egyszerű használati esetekben, amikor a teljes választ szeretné kapni:

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

Streamelés végrehajtása

Valós idejű frissítések a munkafolyamat végrehajtásakor:

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

Külső bemeneti kérések kezelése

Ha egy munkafolyamat olyan végrehajtókat tartalmaz, amelyek külső bemenetet kérnek (használnak request_info), ezek a kérések függvényhívásokként jelennek meg az ügynök válaszában. A függvényhívás a következő nevet WorkflowAgent.REQUEST_INFO_FUNCTION_NAMEhasználja:

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

Válaszok megadása függőben lévő kérelmekre

Ha egy külső bemeneti kérés után szeretné folytatni a munkafolyamat végrehajtását, hozzon létre egy függvényeredményt, és küldje vissza:

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

Teljes példa

Íme egy teljes példa egy munkafolyamat-ügynök streamelési kimenettel való bemutatására:

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

Az eseménykonvertálás ismertetése

Amikor egy munkafolyamat ügynökként fut, a munkafolyamat-események ügynökválaszokká alakulnak. A válasz típusa a hívás run()módjától függ:

  • run(): A munkafolyamat befejezése után a teljes eredményt tartalmazó eredményt adja vissza AgentResponse .
  • run(..., stream=True): Az objektumok aszinkron iterálható értékét AgentResponseUpdate adja vissza a munkafolyamat végrehajtásakor, valós idejű frissítéseket biztosítva

as_agent() továbbítja mind a "output" (terminál), mind a "intermediate" eseményeket a hívónak. A továbbított eseménytípusok halmaza a következő AGENT_FORWARDED_EVENT_TYPES = {"output", "intermediate"}: . Minden egyéb, a munkafolyamaton belüli esemény eldobásra kerül.

A végrehajtás során a belső munkafolyamat-események az alábbiak szerint vannak leképezve az ügynökválaszokra:

Munkafolyamat-esemény Ügynök válasza
event.type == "output" Végső válasz – AgentResponseUpdate formájában továbbítva (folyamatos továbbítás), vagy AgentResponse elemmé összevonva (nem folyamatos továbbítás). response.text csak ezeket a terminálkimeneteket adja vissza.
event.type == "intermediate" Megfigyelési előrehaladás — text_reasoning alatt AgentResponseUpdate tartalomként jelenik meg. Nem része a(z) response.text.
event.type == "request_info" Függvényhívási tartalommá alakítva a WorkflowAgent.REQUEST_INFO_FUNCTION_NAME használatával
Egyéb események Figyelmen kívül hagyva (csak munkafolyamat-belső)

Ez az átalakítás lehetővé teszi a szabványos ügynökfelület használatát, miközben szükség esetén továbbra is hozzáférhet a részletes munkafolyamat-információkhoz. A .text tulajdonság mind a AgentResponse, mind a AgentResponseUpdate esetén csak a végső ("output") választ adja vissza; a köztes előrehaladás eléréséhez vizsgálja meg a text_reasoning tartalomelemeit.

A Go ügynökként burkolja a munkafolyamatokat a workflow/agentworkflow. Ez lehetővé teszi, hogy a hívók a normál ügynök által futtatott API-kat használják, miközben a szolgáltató végrehajtja a munkafolyamatot a színfalak mögött.

Requirements

A munkafolyamat kezdő végrehajtójának el kell fogadnia []*message.Message. Az üzemeltetett ügynök-végrehajtók, valamint a(z) messageworkflow.Configure használatával konfigurált végrehajtók megfelelnek ennek a követelménynek.

Munkafolyamat-ügynök létrehozása

A kompatibilis munkafolyamatok ügynökként való körbefuttatására használható agentworkflow.New :

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

agentworkflow. AgentConfig-paraméterek

Paraméter Típus Description
Config agent.Config Beágyazott ügynök konfigurációja, beleértve a nevet, leírást, köztes szoftvereket, eszközöket és futtatási beállításokat.
Environment *inproc.ExecutionEnvironment Nem kötelező végrehajtási környezet. Alapértelmezés szerint inproc.OffThread, vagy inproc.Concurrent, ha a munkafolyamat lehetővé teszi az egyidejű végrehajtást.
IncludeErrorDetails bool Ha true, részletes munkafolyamat-hibaüzeneteket tartalmaz az ügynökválaszokban. Alapértelmezett érték: false.
IncludeOutputsInResponse bool Ha true, átalakítja a kimenő munkafolyamat üzenetkimeneteit az ügynökválaszok tartalmává. Alapértelmezett érték: false.

Munkafolyamat-ügynökök használata

Munkamenet létrehozása

Hozzon létre egy ügynök-munkamenetet, ha azt szeretné, hogy a munkafolyamat állapota több fordulóban is megmaradjon:

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

Nem-streamelő végrehajtás

Használja a RunText vagy a Run elemet, és gyűjtse össze a választ nem streamelő végrehajtás esetén:

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

Streamelés végrehajtása

Valós idejű frissítések a munkafolyamat végrehajtásakor:

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

Külső bemeneti kérések kezelése

A munkafolyamat külső kérései függvényhívási tartalomként jelennek meg az ügynök válaszában. Vizsgálja meg a válaszüzeneteket, és küldje el a megfelelő választ egy későbbi futtatás során.

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

Válaszok megadása függőben lévő kérelmekre

A munkafolyamat végrehajtásának folytatásához adja vissza a megfelelő választartalmat a munkafolyamat-ügynöknek:

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
}

Munkamenet szerializálása és újraindítása

A munkafolyamat-ügynök munkamenetei szerializálhatók az adatmegőrzéshez, és később folytathatók:

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

Teljes példa

Az alábbi példa létrehoz egy tartalomfolyamat-munkafolyamatot, ügynökként burkolja, és a válaszokat a normál ügynök API-val streameli:

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

Warning

azidentity.NewDefaultAzureCredential a fejlesztéshez kényelmes, de a termelési környezetben gondos megfontolást igényel. Éles környezetben fontolja meg egy konkrét hitelesítő adat, például a azidentity.NewManagedIdentityCredential használatát a késleltetési problémák, a hitelesítő adatok nem szándékos próbálgatása és a visszaesési mechanizmusokból eredő esetleges biztonsági kockázatok elkerülése érdekében.

Tip

Tekintse meg a munkafolyamatot bemutató ügynökmintát egy teljes, futtatható példáért.

Használati esetek

1. Összetett ügynökcsatornák

Többügynökből álló munkafolyamat burkolása egyetlen ügynökként az alkalmazásokban való használatra:

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

2. Ügynök összetétele

Munkafolyamat-ügynökök használata nagyobb rendszerek összetevőiként:

  • A munkafolyamat-ügynök egy másik ügynök által is használható eszközként
  • Több munkafolyamat-ügynök is vezénylhető együtt
  • A munkafolyamat-ügynökök más munkafolyamatokban is beágyazhatók

3. API-integráció

Összetett munkafolyamatok elérhetővé tétele olyan API-kon keresztül, amelyek a szabványos ügynökfelületre várnak, és lehetővé teszik a következőt:

  • Kifinomult háttérbeli munkafolyamatokat használó csevegőfelületek
  • Integráció meglévő ügynökalapú rendszerekkel
  • Fokozatos migrálás egyszerű ügynökökről összetett munkafolyamatokra

Következő lépések

  • Last updated on