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

Tento dokument obsahuje přehled použití pracovních postupů jako agentů v rozhraní Microsoft Agent Framework.

Přehled

Někdy jste vytvořili sofistikovaný pracovní postup s několika agenty, vlastními exekutory a složitou logikou , ale chcete ho použít stejně jako jakýkoli jiný agent. Přesně to vám agenti pracovního postupu umožňují. Když pracovní postup zabalíte jako nějaký Agent, můžete s ním pracovat pomocí stejného známého rozhraní API, které byste použili pro jednoduchého chatovacího agenta.

Klíčové výhody

  • Sjednocené rozhraní: Interakce se složitými pracovními postupy pomocí stejného rozhraní API jako jednoduchých agentů
  • Kompatibilita rozhraní API: Integrace pracovních postupů se stávajícími systémy, které podporují rozhraní agenta
  • Kompozičnost: Použití pracovních postupových agentů jako základních stavebních prvků ve větších systémech agentů nebo v jiných pracovních postupech.
  • Správa relací: Využijte relace agenta pro stav a obnovení konverzace
  • Podpora streamování: Získání aktualizací v reálném čase při provádění pracovního postupu

Jak to funguje

Při převodu pracovního postupu na agenta:

  1. Pracovní postup se validuje, aby jeho spouštěč mohl přijmout požadované vstupní typy.
  2. Vytvoří se session pro správu stavu konverzace.
  3. Vstupní zprávy se směrují do počátečního vykonavatele pracovního postupu.
  4. Události pracovního postupu se převedou na aktualizace odpovědí agenta.
  5. Žádosti o externí vstup (z RequestInfoExecutor) se zobrazují jako volání funkcí.

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

Parameter Typ Description
id string? Volitelný jedinečný identifikátor agenta. Automaticky vygenerováno, pokud není k dispozici.
name string? Volitelný zobrazovaný název agenta.
description string? Volitelný popis účelu agenta.
executionEnvironment IWorkflowExecutionEnvironment? Volitelné spouštěcí prostředí. Výchozí hodnota je buď InProcessExecution.OffThread nebo InProcessExecution.Concurrent, v závislosti na konfiguraci pracovního postupu.
includeExceptionDetails bool Pokud true obsahuje zprávy o výjimkách v chybovém obsahu. Výchozí hodnota je false.
includeWorkflowOutputsInResponse bool Pokud truetransformuje výstupy odchozího pracovního postupu na obsah v odpovědích agenta. Výchozí hodnota je false.

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

Parameter Typ Description
name str | None Volitelný zobrazovaný název agenta. Automaticky vygenerováno, pokud není k dispozici.

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

Poznámka:

Relace jsou volitelné. Pokud nepředáte něco do session a run(), agent interně zpracovává stav. Pokud workflow.as_agent() je vytvořena bez context_providers, architektura přidá InMemoryHistoryProvider() ve výchozím nastavení tak, aby historie s více turny fungovala. Pokud seznam context_providers předáte explicitně, použije se v původní podobě.

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.

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

Událost pracovního postupu Odpověď agenta
event.type == "output" Předávané jako AgentResponseUpdate (streamování) nebo agregované do AgentResponse (bez streamování)
event.type == "request_info" Převedeno na obsah volání funkce pomocí WorkflowAgent.REQUEST_INFO_FUNCTION_NAME
Další události Ignorováno (pouze interní pracovní postup)

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.

Případy použití

1. Komplexní kanály agentů

Zabalte pracovní postup s více agenty jako jednoho agenta pro použití v aplikacích:

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

2. Složení agenta

Použití agentů pracovních postupů jako komponent ve větších systémech:

  • Agent pracovního postupu může použít jako nástroj jiného agenta.
  • Dohromady je možné orchestrovat více agentů pracovního postupu.
  • Agenty pracovního postupu je možné vnořit do jiných pracovních postupů.

3. Integrace rozhraní API

Zpřístupnění složitých pracovních postupů prostřednictvím API, které očekává standardní rozhraní Agenta, čímž umožňuje:

  • Rozhraní chatu, která používají sofistikované back-endové pracovní postupy
  • Integrace se stávajícími systémy založenými na agentech
  • Postupné migrace z jednoduchých agentů na složité pracovní postupy

Další kroky

  • Last updated on