Megosztás:

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
  • Szálkezelés: Ügynökszálak használata beszélgetési állapothoz, ellenőrzőpontokhoz é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 érvényesítve van annak biztosítása érdekében, hogy a kezdő végrehajtó fogadja a csevegési üzeneteket
  2. Létrejön egy szál a beszélgetés állapotának és ellenőrzőpontjainak 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 az ügynök-alapú végrehajtók, például a ChatClientAgent használatakor.

Munkafolyamat-ügynök létrehozása

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

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 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.
checkpointManager CheckpointManager? Nem kötelező ellenőrzőpont-kezelő a munkamenetek közötti adatmegőrzéshez.
executionEnvironment IWorkflowExecutionEnvironment? Nem kötelező végrehajtási környezet. Alapértelmezetten InProcessExecution.OffThread vagy InProcessExecution.Concurrent, a munkafolyamat konfigurációja alapján.

Munkafolyamat-ügynökök használata

Szál létrehozása

A munkafolyamat-ügynökkel folytatott minden beszélgetéshez egy szálra van szükség az állapot kezeléséhez:

// Create a new thread for the conversation
AgentThread thread = workflowAgent.GetNewThread();

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

AgentRunResponse response = await workflowAgent.RunAsync(messages, thread);

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

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

Szál szerializálása és újrakezdése

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

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

Ellenőrzőpont-készítés munkafolyamat-ügynökökkel

Engedélyezze az ellenőrzőpont-ellenőrzést a munkafolyamat állapotának a folyamat újraindítása során történő megőrzéséhez:

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

Requirements

A munkafolyamat ügynökként való használatához a munkafolyamat kezdő végrehajtójának képesnek kell lennie bemenetként kezelni list[ChatMessage] . Ez automatikusan teljesül, amikor a ChatAgent vagy a AgentExecutor van használatban.

Munkafolyamat-ügynök létrehozása

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

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

Szál létrehozása

A munkafolyamat-ügynökkel folytatott minden beszélgetéshez egy szálra van szükség az állapot kezeléséhez:

# Create a new thread for the conversation
thread = workflow_agent.get_new_thread()

Nem-streamelő végrehajtás

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

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

Streamelés végrehajtása

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

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)

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. A munkafolyamat-ügynök nyomon követi a függőben lévő kéréseket, és válaszokat vár a folytatás előtt:

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

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

A munkafolyamat végrehajtásának folytatása külső bemeneti kérés után:

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

Teljes példa

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

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

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 attól függ, hogy melyik metódust használja:

  • run(): A munkafolyamat befejezése után a teljes eredményt tartalmazó eredményt adja vissza AgentRunResponse .
  • run_stream(): A munkafolyamat végrehajtása során AgentRunResponseUpdate objektumokat ad ki, valós idejű frissítéseket biztosítva

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
AgentRunUpdateEvent Átadva mint AgentRunResponseUpdate (streamelés) vagy összesítve mint AgentRunResponse (nem streamelés)
RequestInfoEvent FunctionCallContent Konvertálás ésFunctionApprovalRequestContent
Egyéb események A raw_representation a megfigyelhetőség érdekében bevonva

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.

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