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

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

Overview

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

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

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

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

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

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

Requirements

Чтобы использовать рабочий процесс в качестве агента, начальный исполнитель рабочего процесса должен иметь возможность обрабатывать 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

Parameter	Type	Описание
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);
}

Это важно

Сериализованный сеанс агента рабочего процесса содержит внутреннюю контрольную точку рабочего процесса. Если приложение восстанавливает рабочий процесс перед десериализацией или запуском сеанса, каждый внутренний агент должен быть повторно создан с тем же ChatClientAgentOptions.Id (и, если задано значение Name, с тем же Name).

id, переданный в workflow.AsAIAgent(...), идентифицирует только внешний агент рабочего процесса. Это не стабилизирует идентификаторы исполнителей агентов в рамках рабочего процесса. Инструкции по настройке см. в разделе Восстановление из контрольных точек.

Requirements

Чтобы использовать рабочий процесс в качестве агента, начальный исполнитель рабочего процесса должен иметь возможность обрабатывать входные данные сообщения. Это автоматически удовлетворяется при использовании 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

Parameter	Type	Описание
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.

Go оформляет рабочие процессы как агентов с workflow/agentworkflow. Это позволяет вызывающим пользователям использовать API-интерфейсы обычного агента, а поставщик выполняет рабочий процесс за кулисами.

Requirements

Исполнитель запуска рабочего процесса должен принять []*message.Message. Размещенные исполнители агента и исполнители, настроенные с помощью messageworkflow.Configure, удовлетворяют этому требованию.

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

Используется agentworkflow.New для упаковки любого совместимого рабочего процесса в качестве агента:

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

Параметры agentworkflow.AgentConfig

Parameter	Type	Описание
Config	agent.Config	Внедренная конфигурация агента, включая имя, описание, ПО промежуточного слоя, средства и параметры выполнения.
Environment	*inproc.ExecutionEnvironment	Необязательная среда выполнения кода. По умолчанию используется inproc.OffThread или inproc.Concurrent, если рабочий процесс допускает параллельное выполнение.
IncludeErrorDetails	bool	Если trueсодержит подробные сообщения об ошибках рабочего процесса в ответах агента. По умолчанию — false.
IncludeOutputsInResponse	bool	Если true, преобразует выходные данные исходящих сообщений рабочего процесса в контент ответов агента. По умолчанию — false.

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

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

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

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

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

Используйте RunText или Run и соберите ответ при непотоковом выполнении:

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

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

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

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

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

Внешние запросы из рабочего процесса отображаются в виде содержимого вызова функции в ответе агента. Проверьте сообщения с ответами на наличие содержимого запроса и отправьте соответствующее сообщение с ответом при последующем запуске.

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

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

Чтобы продолжить выполнение рабочего процесса, верните соответствующее содержимое ответа агенту рабочего процесса:

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
}

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

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

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

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

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

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

Предупреждение

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

Tip

Просмотрите рабочий процесс как пример агента для полного запуска.

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

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

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

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

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

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

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

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

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

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

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

• Узнайте, как обрабатывать запросы и ответы в рабочих процессах
• Узнайте, как управлять состоянием в рабочих процессах
• Узнайте, как создавать контрольные точки и возобновлять работу с них
• Узнайте, как отслеживать рабочие процессы
• Узнайте об изоляции состояния в рабочих процессах
• Узнайте, как визуализировать рабочие процессы