Events

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

Встроенные типы событий

// Workflow lifecycle events
WorkflowStartedEvent     // Workflow execution begins
WorkflowOutputEvent      // Workflow outputs data
WorkflowErrorEvent       // Workflow encounters an error
WorkflowWarningEvent     // Workflow encountered a warning

// Executor events
ExecutorInvokedEvent     // Executor starts processing
ExecutorCompletedEvent   // Executor finishes processing
ExecutorFailedEvent      // Executor encounters an error
AgentResponseEvent       // An agent run produces output
AgentResponseUpdateEvent // An agent run produces a streaming update

// Superstep events
SuperStepStartedEvent    // Superstep begins
SuperStepCompletedEvent  // Superstep completes

// Request events
RequestInfoEvent         // A request is issued

Замечание

Когда агенты используют инструменты, требующие утверждения, RequestInfoEvent обычно несет ToolApprovalRequestContent нагрузку для вызовов инструментов, требующих одобрения человека. Подробнее об обработке этих событий см. в разделе Human-in-the-Loop.

# All events use the unified WorkflowEvent class with a type discriminator:

# Workflow lifecycle events
WorkflowEvent.type == "started"             # Workflow execution begins
WorkflowEvent.type == "status"              # Workflow state changed (use .state)
WorkflowEvent.type == "output"              # Workflow produces a terminal (final) output
WorkflowEvent.type == "intermediate"        # Workflow produces an intermediate (observational) output
WorkflowEvent.type == "failed"              # Workflow terminated with error (use .details)
WorkflowEvent.type == "error"               # Non-fatal error from user code
WorkflowEvent.type == "warning"             # Workflow encountered a warning

# Executor events
WorkflowEvent.type == "executor_invoked"    # Executor starts processing
WorkflowEvent.type == "executor_completed"  # Executor finishes processing
WorkflowEvent.type == "executor_failed"     # Executor encounters an error
WorkflowEvent.type == "data"                # Deprecated alias for "intermediate"

# Superstep events
WorkflowEvent.type == "superstep_started"   # Superstep begins
WorkflowEvent.type == "superstep_completed" # Superstep completes

# Request events
WorkflowEvent.type == "request_info"        # A request is issued

Замечание

Когда агенты используют инструменты, требующие утверждения, request_info события обычно несут полезную Content нагрузку для type == "function_approval_request" вызовов инструментов, требующих одобрения человека. Подробнее об обработке этих событий см. в разделе Human-in-the-Loop.

Замечание

"output" и "intermediate" — это два выходных дискриминатора. Исполнитель, назначенный источником вывода терминала, генерирует события , обрабатываемые "output". Компонент, обозначенный как источник промежуточного вывода, генерирует события "intermediate", которые потребляются WorkflowRunResult.get_intermediate_outputs(). Тип "data" является устаревшим псевдонимом для "intermediate" и будет удалён в одном из будущих выпусков; в новом коде рекомендуется использовать фильтрацию по "intermediate".

События Go реализуют workflow.Event и предоставляют доступ к своим полезным данным через Data(). Используйте переключатель типов для проверки их типизированных полей:

Тип события	Полезная нагрузка и назначение
`workflow.StartedEvent`	Запускает цикл обработки входных данных. `Message` содержит входные данные.
`workflow.OutputEvent`	Назначенный исполняющий модуль возвращает значение в `Output`. `Tags` обозначает промежуточные выходные данные.
`workflow.ErrorEvent`	Сообщает об ошибке в `Error`. `SubWorkflowID` указывает на вложенный рабочий процесс при установке.
`workflow.WorkflowWarningEvent`	Выводит предупреждение в `Message`.
`workflow.ExecutorInvokedEvent`	Исполнительный модуль начинает обработку. `ExecutorID` идентифицирует его и `Message` содержит входные данные.
`workflow.ExecutorCompletedEvent`	Исполнитель завершает обработку. `Result` содержит свой результат.
`workflow.ExecutorFailedEvent`	Исполнитель завершается с ошибкой. `ExecutorID` и `Error` определяют сбой.
`workflow.SuperStepStartedEvent`	Начинается суперстеп. `StepNumber` и `StartInfo` описывают его.
`workflow.SuperStepCompletedEvent`	Суперстеп завершается. `CompletionInfo` описывает незавершенную работу, запросы и любую из контрольных точек.
`workflow.RequestInfoEvent`	`Request` содержит внешний запрос, на который узел может ответить.
`workflow.RequestHaltEvent`	Исполнитель процесса запрашивает остановку. `Result` содержит полезные данные, если таковые есть.

Go использует OutputEvent как для терминального, так и для промежуточного вывода. Вызовите IsIntermediate(), чтобы различать их; он не использует отдельные "output" и "intermediate" дискриминаторы Python.

Размещенные агенты также используют OutputEvent с полезной нагрузкой *agent.ResponseUpdate или *agent.Response при включении соответствующих событий. Go не определяет отдельные AgentResponseEvent и AgentResponseUpdateEvent типы. См. Исполнители агента для настройки.

Использование событий

using Microsoft.Agents.AI.Workflows;

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    switch (evt)
    {
        case ExecutorInvokedEvent invoke:
            Console.WriteLine($"Starting {invoke.ExecutorId}");
            break;

        case ExecutorCompletedEvent complete:
            Console.WriteLine($"Completed {complete.ExecutorId}: {complete.Data}");
            break;

        case WorkflowOutputEvent output:
            Console.WriteLine($"Workflow output: {output.Data}");
            return;

        case WorkflowErrorEvent error:
            Console.WriteLine($"Workflow error: {error.Exception}");
            return;
    }
}

from agent_framework import WorkflowEvent

async for event in workflow.run(input_message, stream=True):
    if event.type == "executor_invoked":
        print(f"Starting {event.executor_id}")
    elif event.type == "executor_completed":
        print(f"Completed {event.executor_id}: {event.data}")
    elif event.type == "intermediate":
        print(f"Intermediate output from {event.executor_id}: {event.data}")
    elif event.type == "output":
        print(f"Terminal output: {event.data}")
        return
    elif event.type == "error":
        print(f"Workflow error: {event.data}")
        return

Используйте switch по типу для событий рабочего процесса Go. Обрабатывайте ошибки итератора и события сбоя по отдельности: получение выходных данных не гарантирует успешность выполнения остального рабочего процесса.

Наблюдение за событиями

inproc.Default.Run перемещается к следующей остановке и накапливает события. Выполнение может быть приостановлено в ожидании внешнего ввода; остановка не обязательно означает завершение.

Следующая функция выводит действие исполнителя и выходные данные, возвращает ошибки и закрывает выполнение:

import (
    "context"
    "errors"
    "fmt"

    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func observeEvents(
    ctx context.Context, wf *workflow.Workflow, input any,
) (err error) {
    run, err := inproc.Default.Run(ctx, wf, input)
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
    }()

    for event := range run.NewEvents() {
        switch event := event.(type) {
        case workflow.ExecutorInvokedEvent:
            fmt.Printf("Starting %s\n", event.ExecutorID)
        case workflow.ExecutorCompletedEvent:
            fmt.Printf("Completed %s: %v\n",
                event.ExecutorID, event.Result)
        case workflow.OutputEvent:
            kind := "Terminal"
            if event.IsIntermediate() {
                kind = "Intermediate"
            }
            fmt.Printf("%s output from %s: %v\n",
                kind, event.ExecutorID, event.Output)
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }
    return nil
}

NewEvents() возвращает события, накопленные с момента предыдущего вызова. Его метка перемещается при вызове, даже если вы перестаете итерировать рано. Используйте OutgoingEvents() для воспроизведения всех накопленных событий вместо повторного вызова NewEvents() .

Стриминговые мероприятия

Используется RunStreaming для наблюдения за событиями по мере их поступления. Эта функция отслеживает рабочий процесс, который не требует внешних входных данных:

import (
    "context"
    "errors"
    "fmt"

    "github.com/microsoft/agent-framework-go/workflow"
    "github.com/microsoft/agent-framework-go/workflow/inproc"
)

func streamEvents(
    ctx context.Context, wf *workflow.Workflow, input any,
) (err error) {
    run, err := inproc.Default.RunStreaming(ctx, wf, input)
    if err != nil {
        return err
    }
    defer func() {
        err = errors.Join(err, run.Close(context.WithoutCancel(ctx)))
    }()

    for event, streamErr := range run.WatchStream(ctx) {
        if streamErr != nil {
            return streamErr
        }
        switch event := event.(type) {
        case workflow.OutputEvent:
            kind := "Terminal"
            if event.IsIntermediate() {
                kind = "Intermediate"
            }
            fmt.Printf("%s output from %s: %v\n",
                kind, event.ExecutorID, event.Output)
        case workflow.ErrorEvent:
            return event.Error
        case workflow.ExecutorFailedEvent:
            return fmt.Errorf("executor %q failed: %w",
                event.ExecutorID, event.Error)
        }
    }
    return nil
}

WatchStream остается в состоянии ожидания, когда у рабочего процесса есть ожидающие внешние запросы. Обрабатывайте RequestInfoEvent и отправляйте соответствующие ответы через SendResponse, когда используете это для рабочих процессов, управляемых запросами.

Используйте WatchUntilHalt вместо этого, когда потребитель должен возобновить работу после паузы, не ожидая ответа. Только один итератор событий может одновременно обрабатывать потоковый запуск. См. Рабочие процессы с участием человека для обработки и возобновления запросов.

Пользовательские события

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

Отслеживание хода выполнения — отображайте промежуточные шаги, чтобы звонящие могли показывать обновления состояния.
Вывод диагностики — выводит предупреждения, метрики или отладочные сведения, не изменяя выходные данные рабочего процесса.
Данные домена ретранслятора — отправка структурированных полезных данных (например, записи баз данных, вызовы инструментов) прослушивателям в режиме реального времени.

Определение пользовательских событий

Определите настраиваемое событие созданием подкласса WorkflowEvent. Базовый конструктор принимает необязательную нагрузку object? data, которая становится доступной через свойство Data.

using Microsoft.Agents.AI.Workflows;

// Simple event with a string payload
internal sealed class ProgressEvent(string step) : WorkflowEvent(step) { }

// Event with a structured payload
internal sealed class MetricsEvent(MetricsData metrics) : WorkflowEvent(metrics) { }

В Python, создайте настраиваемые события, используя класс WorkflowEvent непосредственно с строкой-дискриминатором пользовательского типа. type Параметры data содержат все сведения.

from agent_framework import WorkflowEvent

# Create a custom event with a custom type string and payload
event = WorkflowEvent(type="progress", data="Step 1 complete")

# Custom event with a structured payload
event = WorkflowEvent(type="metrics", data={"latency_ms": 42, "tokens": 128})

Замечание

Типы событий "started", "status" и "failed" зарезервированы для уведомлений жизненного цикла фреймворка. Если исполнитель пытается выпустить один из этих типов, событие игнорируется и регистрируется предупреждение.

Определите настраиваемое событие, создав тип, реализующий workflow.Event интерфейс. Метод Data возвращает полезные данные события.

import "github.com/microsoft/agent-framework-go/workflow"

type ProgressEvent struct {
    Step string
}

func (e ProgressEvent) Data() any {
    return e.Step
}

var _ workflow.Event = ProgressEvent{}

Генерация пользовательских событий

Вызовите кастомные события из обработчика сообщений исполнителя с помощью функции AddEventAsync на IWorkflowContext.

using Microsoft.Agents.AI.Workflows;

internal sealed class ProgressEvent(string step) : WorkflowEvent(step) { }

internal sealed partial class CustomExecutor() : Executor("CustomExecutor")
{
    [MessageHandler]
    private async ValueTask HandleAsync(string message, IWorkflowContext context)
    {
        await context.AddEventAsync(new ProgressEvent("Validating input"));

        // Executor logic...

        await context.AddEventAsync(new ProgressEvent("Processing complete"));
    }
}

Передайте пользовательские события из обработчика событий, вызвав add_event на WorkflowContext.

from agent_framework import (
    handler,
    Executor,
    WorkflowContext,
    WorkflowEvent,
)

class CustomExecutor(Executor):

    @handler
    async def handle(self, message: str, ctx: WorkflowContext[str]) -> None:
        await ctx.add_event(WorkflowEvent(type="progress", data="Validating input"))

        # Executor logic...

        await ctx.add_event(WorkflowEvent(type="progress", data="Processing complete"))

Генерируйте пользовательские события из обработчика исполнителя, вызвав AddEvent для workflow.Context:

import "github.com/microsoft/agent-framework-go/workflow"

customExecutor := workflow.NewExecutor("CustomExecutor", func(ctx *workflow.Context, message string) error {
    if err := ctx.AddEvent(ProgressEvent{Step: "Validating input"}); err != nil {
        return err
    }

    // Executor logic...

    return ctx.AddEvent(ProgressEvent{Step: "Processing complete"})
}).Bind()

Для пользовательских событий не требуется WithOutputFrom: AddEvent выдает их напрямую. Используйте YieldOutput и назначьте источник вывода, если вам также нужны обычные выходные данные рабочего процесса.

Использование пользовательских событий

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

await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    switch (evt)
    {
        case ProgressEvent progress:
            Console.WriteLine($"Progress: {progress.Data}");
            break;

        case WorkflowOutputEvent output:
            Console.WriteLine($"Done: {output.Data}");
            return;
    }
}

Фильтрация по строке дискриминатора пользовательского типа:

async for event in workflow.run(input_message, stream=True):
    if event.type == "progress":
        print(f"Progress: {event.data}")
    elif event.type == "output":
        print(f"Done: {event.data}")
        return

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

for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }

    switch e := evt.(type) {
    case ProgressEvent:
        fmt.Printf("Progress: %v\n", e.Data())
    case workflow.OutputEvent:
        fmt.Printf("Done: %v\n", e.Output)
    case workflow.ErrorEvent:
        return e.Error
    case workflow.ExecutorFailedEvent:
        return fmt.Errorf("executor %q failed: %w", e.ExecutorID, e.Error)
    }
}

Дальнейшие действия

Построитель рабочих процессов и выполнение

См. также:

Обратная связь

Были ли сведения на этой странице полезными?

Последнее обновление: 2026-10-29

Events

Встроенные типы событий

Использование событий

Наблюдение за событиями

Стриминговые мероприятия

Пользовательские события

Определение пользовательских событий

Генерация пользовательских событий

Использование пользовательских событий

Дальнейшие действия

Обратная связь

Дополнительные ресурсы