Рабочие процессы Microsoft Agent Framework — использование рабочих процессов в качестве агентов

В этом документе представлен обзор использования рабочих процессов в качестве агентов в Microsoft Agent Framework.

Обзор

Иногда вы создали сложный рабочий процесс с несколькими агентами, пользовательскими исполнителями и сложной логикой, но вы хотите использовать его так же, как и любой другой агент. Это именно то, что позволяют вам делать агенты рабочих процессов. Упаковав рабочий процесс в качестве Agentобъекта, вы можете взаимодействовать с ним с помощью того же знакомого API, который вы будете использовать для простого агента чата.

Ключевые преимущества

  • Унифицированный интерфейс. Взаимодействие со сложными рабочими процессами с помощью того же API, что и простые агенты
  • Совместимость API: интеграция рабочих процессов с существующими системами, поддерживающими интерфейс агента
  • Компоновка: Использование агентов рабочих процессов в качестве строительных блоков в более крупных системах агентов или других рабочих процессах.
  • Управление сеансами: эффективное использование сеансов агента для поддержания состояния и возобновления беседы
  • Поддержка потоковой передачи: получение обновлений в режиме реального времени при выполнении рабочего процесса

Принцип работы

При преобразовании рабочего процесса в агент:

  1. Рабочий процесс проверяется, чтобы его начальный исполнитель может принимать необходимые типы входных данных.
  2. Сеанс создается для управления состоянием беседы
  3. Входные сообщения направляются в начальный исполнитель рабочего процесса
  4. События рабочего процесса преобразуются в обновления ответа агента
  5. Внешние входные запросы (из RequestInfoExecutor) отображаются в виде вызовов функций

Требования

Чтобы использовать рабочий процесс в качестве агента, начальный исполнитель рабочего процесса должен иметь возможность обрабатывать IEnumerable<ChatMessage> как входные данные. Это автоматически удовлетворяется при использовании агентных исполнителей, созданных с помощью AsAIAgent.

Создание агента рабочего процесса

AsAIAgent() Используйте метод расширения для преобразования любого совместимого рабочего процесса в агент:

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

Параметр Тип Description
id string? Необязательный уникальный идентификатор агента. Автоматически генерируется, если не указано.
name string? Необязательное отображаемое имя агента.
description string? Необязательное описание назначения агента.
executionEnvironment IWorkflowExecutionEnvironment? Необязательная среда выполнения кода. По умолчанию используется InProcessExecution.OffThread или InProcessExecution.Concurrent в зависимости от конфигурации рабочего процесса.
includeExceptionDetails bool Если true включает сообщения об исключениях в содержимое ошибки. По умолчанию — false.
includeWorkflowOutputsInResponse bool Если true, преобразует исходящие выходные данные рабочего процесса в содержимое в ответах агента. По умолчанию — false.

Использование агентов рабочих процессов

Создание сеанса

Для каждой беседы с агентом рабочего процесса требуется сеанс для управления состоянием:

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

Операция без потокового режима

Для простых вариантов использования, когда требуется полный ответ:

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

Выполнение потоковой передачи

Для обновлений в режиме реального времени при выполнении рабочего процесса:

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

Обработка внешних запросов ввода

Если рабочий процесс содержит исполнителей, запрашивающих внешние входные данные (используя RequestInfoExecutor), эти запросы отображаются как вызовы функций в ответе агента:

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

Сериализация сеансов и возобновление

Сеансы агента рабочего процесса можно сериализовать для сохраняемости и возобновить позже:

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

Требования

Чтобы использовать рабочий процесс в качестве агента, начальный исполнитель рабочего процесса должен иметь возможность обрабатывать входные данные сообщения. Это автоматически удовлетворяется при использовании Agent или исполнителях на основе агента.

Создание агента рабочего процесса

Вызвать as_agent() на любом совместимом рабочем процессе для его преобразования в агента.

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

Параметр Тип Description
name str | None Необязательное отображаемое имя агента. Автоматически генерируется, если не указано.

Использование агентов рабочих процессов

Создание сеанса

При необходимости можно создать сеанс для управления состоянием беседы на нескольких этапах.

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

Замечание

Сеансы являются необязательными. Если вы не передаете session в run(), агент обрабатывает состояние внутренне. Если workflow.as_agent() создан без context_providers, платформа добавляет InMemoryHistoryProvider() по умолчанию, чтобы многоходовая история работала автоматически. Если вы передаете context_providers явно, этот список используется как есть.

Операция без потокового режима

Для простых вариантов использования, когда требуется полный ответ:

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

Выполнение потоковой передачи

Для обновлений в режиме реального времени при выполнении рабочего процесса:

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

Обработка внешних запросов ввода

Если рабочий процесс содержит исполнителей, запрашивающих внешние входные данные (используя request_info), эти запросы отображаются как вызовы функций в ответе агента. Вызов функции использует имя 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

Предоставление ответов на ожидающие запросы

Чтобы продолжить выполнение рабочего процесса после внешнего запроса ввода, создайте результат функции и отправьте его обратно:

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

Полный пример

Ниже приведен полный пример демонстрации рабочего агента с потоковым выводом:

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

Понимание преобразования событий

При запуске рабочего процесса в качестве агента события рабочего процесса преобразуются в ответы агента. Тип ответа зависит от способа вызова run():

  • run(): возвращает полный AgentResponse результат после завершения рабочего процесса.
  • run(..., stream=True): возвращает асинхронных итерируемых AgentResponseUpdate объектов по мере выполнения, предоставляя обновления в реальном времени.

as_agent() перенаправляет как "output" события (terminal), так и "intermediate" события вызывающей стороне. Набор пересылаемых типов событий: AGENT_FORWARDED_EVENT_TYPES = {"output", "intermediate"}. Все остальные внутренние события рабочего процесса удаляются.

Во время выполнения внутренние события рабочего процесса сопоставляются с ответами агента следующим образом:

Событие рабочего процесса Ответ агента
event.type == "output" Ответ терминала — передается как AgentResponseUpdate (потоковая передача) или агрегируется в AgentResponse (непотоковый). response.text возвращает только эти выходные данные терминала.
event.type == "intermediate" Прогресс наблюдения — отображается как содержимое text_reasoning в AgentResponseUpdate. Не включено в response.text.
event.type == "request_info" Преобразовано в содержимое вызова функции с помощью WorkflowAgent.REQUEST_INFO_FUNCTION_NAME
Другие события Игнорируется (только для внутреннего рабочего процесса)

Это преобразование позволяет использовать стандартный интерфейс агента, но при необходимости иметь доступ к подробным сведениям о рабочем процессе. Свойство .text как у AgentResponse, так и у AgentResponseUpdate возвращает только окончательный ответ ("output"); чтобы получить доступ к промежуточным этапам выполнения, проверяйте элементы содержимого text_reasoning.

Варианты использования

1. Сложные потоки агента

Объедините рабочий процесс с несколькими агентами в виде одного агента для применения в приложениях.

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

2. Состав агента

Используйте агенты рабочего процесса в качестве компонентов в крупных системах:

  • Агент рабочего процесса можно использовать в качестве средства другим агентом.
  • Несколько агентов рабочих процессов можно синхронизировать вместе
  • Агенты рабочих процессов могут быть вложены в другие рабочие процессы

3. Интеграция API

Обеспечьте доступ к сложным рабочим процессам через API, которые ожидают стандартный интерфейс агента, предоставляя:

  • Интерфейсы чата, использующие сложные внутренние рабочие процессы
  • Интеграция с существующими системами на основе агента
  • Постепенная миграция из простых агентов в сложные рабочие процессы

Дальнейшие шаги

  • Last updated on