API функционального рабочего процесса

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

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

API функционального рабочего процесса позволяет создавать рабочие процессы как простые Python асинхронные функции. Вместо определения классов исполнителя, соединения рёбер и использования , вы декорируете функцию с помощью и применяете нативный поток управления Python — , циклы , — для выражения вашей логики.

Параллельное сравнение с API графа см. в разделе API рабочих процессов в обзоре рабочих процессов.

@workflow Декоратор

Примените async к функции @workflow, чтобы создать FunctionalWorkflowDefinition без состояния:

from agent_framework import workflow

@workflow
async def text_pipeline(text: str) -> str:
    upper = await to_upper_case(text)
    return await reverse_text(upper)

Декоратор @workflow поддерживает параметризованную форму с необязательными аргументами:

from agent_framework import workflow

@workflow(name="my_pipeline", description="Uppercase then reverse")
async def text_pipeline(text: str) -> str:
    ...

параметры @workflow

Parameter Type Описание
name str | None Отображаемое имя рабочего процесса. По умолчанию значение функции равно __name__.
description str | None Необязательное описание, доступное для чтения человеком.

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

Вызовите .build() для определения, чтобы создать FunctionalWorkflow с состоянием. Каждый созданный рабочий процесс представляет один логический вызывающий объект или сеанс. Создайте отдельные экземпляры для независимых вызывающих, чтобы изолировать их состояние выполнения и воспроизведения.

Передайте хранилище контрольных точек в .build(), когда рабочий процесс должен сохранять результаты этапов и состояние:

from agent_framework import InMemoryCheckpointStorage

storage = InMemoryCheckpointStorage()
workflow_instance = text_pipeline.build(checkpoint_storage=storage)
Parameter Type Описание
checkpoint_storage CheckpointStorage | None Хранилище по умолчанию для хранения результатов шага и состояния между запусками.

Сигнатура функции рабочего процесса

Первый параметр функции рабочего процесса получает входные данные, .run() переданные в встроенный рабочий процесс. ctx: RunContext Добавьте параметр только в том случае, если требуется HITL, состояние ключа и значения или пользовательские события— это необязательно в противном случае:

# No ctx needed — just a plain pipeline
@workflow
async def simple_pipeline(data: str) -> str:
    result = await process(data)
    return result

# ctx needed for HITL, state, or custom events
@workflow
async def hitl_pipeline(data: str, ctx: RunContext) -> str:
    feedback = await ctx.request_info({"draft": data}, response_type=str)
    return feedback

RunContext сначала определяется по аннотации типа, затем по имени параметра ctx, поэтому работают и параметр ctx: RunContext, и голый параметр ctx.

Запуск рабочего процесса

Вызовите .run() на объекте FunctionalWorkflow, возвращаемом .build():

workflow_instance = text_pipeline.build()

# .run() wraps the result in a WorkflowRunResult with events and state
result = await workflow_instance.run("hello world")
print(result.text)                        # first output as a string
print(result.get_outputs())               # list of terminal outputs
print(result.get_intermediate_outputs())  # list of intermediate outputs
print(result.get_final_state())           # WorkflowRunState.IDLE

параметры run()

Parameter Type Описание
message Any | None Входные данные, переданные функции рабочего процесса в качестве первого аргумента.
stream bool Если True, возвращает ResponseStream, который возвращает WorkflowEvent объекты. По умолчанию — False.
responses dict[str, Any] | None Ответы HITL, для которых ключом является request_id. Используется для возобновления приостановленного рабочего процесса.
checkpoint_id str | None Контрольная точка для восстановления. checkpoint_storage должен быть задан во время сборки или для этого запуска.
checkpoint_storage CheckpointStorage | None Переопределяет хранилище для контрольных точек, используемое при сборке, для данного запуска.
include_status_events bool Включите события изменения состояния в результат, отличный от потоковой передачи.

Укажите один входной режим для каждого вызова: message, responsesили checkpoint_id. Исключение составляет возобновление из контрольной точки с внешним вводом, при котором checkpoint_id и responses могут быть переданы вместе.

WorkflowRunResult

run() (непотоковый) возвращает WorkflowRunResult. Ключевые методы:

Метод / свойство Возвраты Описание
.text str Первый вывод в виде строки. Пустая строка, если строковые выходные данные отсутствуют.
.get_outputs() list[Any] Все выходные данные терминала, создаваемые рабочим процессом (события с type == "output").
.get_intermediate_outputs() list[Any] Все промежуточные выходные данные, создаваемые рабочим процессом (события с type == "intermediate").
.get_final_state() WorkflowRunState Конечное состояние выполнения (IDLE, , IDLE_WITH_PENDING_REQUESTSFAILED...).
.get_request_info_events() list[WorkflowEvent] Ожидающие запросы HITL при состоянии IDLE_WITH_PENDING_REQUESTS.

Streaming

``` Передайте stream=True, чтобы получать события по мере их создания: ```

from agent_framework import workflow

@workflow
async def data_pipeline(url: str) -> str:
    raw = await fetch_data(url)
    return await transform_data(raw)

workflow_instance = data_pipeline.build()

# stream=True returns a ResponseStream you iterate with async for
stream = workflow_instance.run("https://example.com/api/data", stream=True)
async for event in stream:
    if event.type == "output":
        print(f"Output: {event.data}")

# After iteration, get_final_response() returns the WorkflowRunResult
result = await stream.get_final_response()
print(f"Final state: {result.get_final_state()}")

Полный пример см. в python/samples/03-workflows/functional/basic_streaming_pipeline.py.

@step Декоратор

@step — это декоратор, который добавляет кэширование результатов, выбросы событий и контрольные точки на шаг для отдельных асинхронных функций:

from agent_framework import step, workflow

@step
async def fetch_data(url: str) -> dict:
    # expensive — hits a real API
    return await http_get(url)

@workflow
async def pipeline(url: str) -> str:
    raw = await fetch_data(url)
    return process(raw)

Что @step делает внутри рабочего процесса

  • Кэширует результаты — результат хранится с идентификатором воспроизведения, полученным на основе определения шага и связанных с ним аргументов. При перезапуске или восстановлении контрольной точки HITL завершенный шаг мгновенно возвращает сохраненный результат, нежели повторно выполняется.
  • Выдает события — executor_invoked / executor_completed / executor_failed генерируются для отслеживания. При попадании в кэш вместо этого выдаётся executor_bypassed.
  • Сохраняет контрольные точки — если у созданного рабочего процесса есть checkpoint_storage, контрольная точка сохраняется после завершения каждого шага.
  • Внедряет RunContext — если функция шага объявляет параметр ctx: RunContext, активный контекст автоматически внедряется.

Вне запущенного рабочего процесса @step является прозрачной — функция ведет себя идентично своей версии без украшения, что делает ее полностью тестируемой в изоляции.

Когда следует использовать @step

Используйте @step для функций, которые дорогостоящи для повторного выполнения: вызовы агента, внешние API запросы или любая операция, где повторное выполнение при возобновлении может быть дорогостоящим или иметь побочные эффекты. Обычные функции (без @step) по-прежнему работают внутри @workflow; они просто повторно выполняются при возобновлении рабочего процесса.

from agent_framework import InMemoryCheckpointStorage, step, workflow

@step  # cached — won't re-run on resume
async def call_llm(prompt: str) -> str:
    return (await agent.run(prompt)).text

# No @step — cheap, fine to re-run
async def validate(text: str) -> bool:
    return len(text) > 0

@workflow
async def pipeline(topic: str) -> str:
    draft = await call_llm(f"Write about: {topic}")
    ok = await validate(draft)
    return draft if ok else ""

storage = InMemoryCheckpointStorage()
workflow_instance = pipeline.build(checkpoint_storage=storage)

@step также принимает name параметр:

@step(name="transform")
async def transform_data(raw: dict) -> str:
    ...

Управление идентификатором повтора

Agent Framework автоматически выводит идентификатор повторного воспроизведения из канонических связанных аргументов. Одновременные вызовы с тем же автоматическим идентификатором должны быть взаимозаменяемыми. Используйте replay_key, если аргументы непрозрачны, захваченное состояние влияет на результат или иначе идентичные аргументы определяют различные логические вызовы:

@step(replay_key=lambda connection, record_id: record_id)
async def load_record(connection: object, record_id: str) -> dict:
    return await connection.load(record_id)

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

Существующие контрольные точки, основанные на ордерах, остаются совместимыми для последовательного воспроизведения. Agent Framework отклоняет одновременное срабатывание legacy-кэша, если он не может доказать, что сохраненный результат принадлежит одному и тому же логическому вызову. Повторно создайте контрольную точку с текущей версией рабочего процесса и добавьте replay_key, если аргументы не предоставляют стабильный идентификатор.

Полный пример см. в python/samples/03-workflows/functional/steps_and_checkpointing.py.

RunContext

RunContext — это контекст выполнения, внедренный в функции рабочего процесса и шага. Это нужно только при использовании HITL, состояния типа "ключ/значение" или пользовательских событий.

Импортируйте его из agent_framework:

from agent_framework import RunContext, workflow

ctx.request_info() — "Человек в контуре"

ctx.request_info() приостанавливает рабочий процесс, чтобы ожидать внешних входных данных:

@workflow
async def review_pipeline(topic: str, ctx: RunContext) -> str:
    draft = await write_draft(topic)
    feedback = await ctx.request_info(
        {"draft": draft, "instructions": "Please review this draft"},
        response_type=str,
        request_id="review_request",
    )
    return await revise_draft(draft, feedback)

Параметры:

Parameter Type Описание
request_data Any Полезная нагрузка, описывающая необходимые входные данные (dict, Pydantic-модель, string, ...).
response_type type Ожидаемый Python тип ответа.
request_id str | None Стабильный идентификатор для этого запроса. Если auto::<index> id не указан, он генерируется детерминированно на основе порядка вызовов.

Семантика воспроизведения: При первом выполнении request_info() вызывает внутренний сигнал, который приостанавливает рабочий процесс. Сигнал никогда не виден вашему коду. Вызывающий получает WorkflowRunResult с get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS.

Чтобы возобновить выполнение, вызовите .run(responses={request_id: value}) для того же созданного рабочего процесса. Рабочий процесс повторно выполняется из верхней части и request_info() возвращает предоставленное значение немедленно.

@step-декорированные функции, которые выполнялись до приостановки, возвращают кэшированные результаты при возобновлении вместо повторного выполнения.

Прежде чем рабочий процесс сообщает о приостановке, Agent Framework ожидает, пока активные одновременные вызовы @step завершатся, и записывает их завершенные результаты для повторного воспроизведения. При возобновлении эти ветви возвращают кэшированные результаты вместо повторения вызовов модели, вызовов инструментов, операций записи или других побочных эффектов. Оставшиеся отсоединенные задачи рабочего процесса отменяются и выводятся из очереди, поэтому работа не продолжается после возврата приостановленного выполнения. Это поведение применяется к собственным шаблонам параллелизма, таким как asyncio.gather и asyncio.TaskGroup.

Для ответа, отличного от None, request_info() приводит и проверяет предоставленное значение на соответствие response_type перед его возвратом. Сопоставление может создавать объявленную модель, а строка становится Content, когда response_type=Content. Если приведение не может вернуть значение запрошенного типа, операция resume вызывает ValueError. Явное указание None сохраняет существующее поведение: рабочий процесс записывает предупреждение в журнал и возвращает None.

Обработка ответа:

workflow_instance = review_pipeline.build()

# Phase 1 — run until the workflow pauses
result1 = await workflow_instance.run("AI Safety")
assert result1.get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS

requests = result1.get_request_info_events()
print(requests[0].request_id)  # "review_request"

# Phase 2 — resume with the human's answer
result2 = await workflow_instance.run(
    responses={"review_request": "Add more details about alignment research"}
)
print(result2.text)

Полный пример см. в python/samples/03-workflows/functional/hitl_review.py.

ctx.request_info() также поддерживается внутри @step функций.

ctx.add_event() — пользовательские события

Используется ctx.add_event() для выдачи событий, относящихся к приложению, а также событий жизненного цикла платформы. Полные сведения и примеры см. в разделе "Создание пользовательских событий".

ctx.get_state() / ctx.set_state() — состояние ключа и значения

Используйте ctx.get_state() и ctx.set_state() для хранения значений, которые сохраняются при прерываниях HITL и включаются в контрольные точки. Полные сведения см. в разделе "Состояние рабочего процесса".

Значения состояния должны быть сериализуемыми в формате JSON при настройке хранилища контрольных точек.

ctx.is_streaming()

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

get_run_context()

Извлекает активный RunContext из любого места в работающем рабочем процессе — полезно для вспомогательных функций, которые не объявляют ctx параметры:

from agent_framework import get_run_context

async def helper():
    ctx = get_run_context()
    if ctx is not None:
        ctx.set_state("helper_ran", True)

Возвращается None при вызове вне запущенного рабочего процесса.

Параллелизм с asyncio.gather

Используйте стандартный параллелизм Python для распределения и сбора — без необходимости в примитивах платформы.

import asyncio
from agent_framework import workflow

@workflow
async def research_pipeline(topic: str) -> str:
    web, papers, news = await asyncio.gather(
        research_web(topic),
        research_papers(topic),
        research_news(topic),
    )
    return await synthesize([web, papers, news])

asyncio.gather также работает при украшении @stepфункций. Каждая ветвь кэшируется по своему идентификатору воспроизведения. Добавьте replay_key, когда непрозрачные аргументы или захваченное состояние позволяют различать одновременные вызовы к одному и тому же шагу.

Полный пример см. в python/samples/03-workflows/functional/parallel_pipeline.py.

Вызов агентов внутри рабочих процессов

Вызовы агентов выполняются как обычные вызовы функций внутри @workflow.

from agent_framework import Agent, workflow

writer = Agent(name="WriterAgent", instructions="Write a short poem.", client=client)
reviewer = Agent(name="ReviewerAgent", instructions="Review the poem.", client=client)

@workflow
async def poem_workflow(topic: str) -> str:
    poem = (await writer.run(f"Write a poem about: {topic}")).text
    review = (await reviewer.run(f"Review this poem: {poem}")).text
    return f"Poem:\n{poem}\n\nReview: {review}"

Добавьте @step к функциям вызова агента, когда хотите, чтобы их результаты кэшировались при возобновлениях HITL или восстановлении контрольных точек.

from agent_framework import step

@step
async def write_poem(topic: str) -> str:
    return (await writer.run(f"Write a poem about: {topic}")).text

Полный пример см. в python/samples/03-workflows/functional/agent_integration.py.

.as_agent() — использование рабочего процесса в качестве агента

Создайте FunctionalWorkflow, а затем оберните его в совместимый с агентом объект с помощью .as_agent():

from agent_framework import workflow

@workflow
async def poem_workflow(topic: str) -> str:
    ...

# Build the workflow and wrap it as an agent
agent = poem_workflow.build().as_agent(name="PoemAgent")

# Use with the standard agent interface
response = await agent.run("Write a poem about the ocean")
print(response.text)

# Or use in a larger workflow or orchestration

.as_agent()FunctionalWorkflowAgent возвращает FunctionalWorkflowAgent, который открывает тот же интерфейс, что и другие объекты агента, делая функциональные рабочие процессы составляемыми с любой системой, принимающей агентов.

Parameter Type Описание
name str | None Покажите имя агента. По умолчанию используется имя рабочего процесса.
description str | None Необязательное переопределение описания. По умолчанию используется описание рабочего процесса.
context_providers Sequence[Any] | None Необязательные провайдеры контекста, связанные с агентом.

Samples

Примеры запуска находятся в следующих примерах папок:

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

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

См. также:

В настоящее время API функционального рабочего процесса недоступен для C#.

В настоящее время API функционального рабочего процесса недоступен для Go. Используйте API графовых рабочих процессов в Конструкторе и выполнении рабочих процессов.

  • Последнее обновление: