От переводчика: Temporal — один из лидеров нового рынка оркестрации, который предлагает иной подход к этой задаче, чем BPM-движки Camunda, Flowable и прочие. Лозунг Temporal — «workflow-как-код». Хм, исполняемая модель BPMN — тоже по факту код. И в чем разница? — Temporal предлагает более инженерный подход, и это многим нравится, в нем нет вот этого менеджерского флёра, как в BPMN. С другой стороны, нет такой наглядной и строгой визуализации процессов.

Ну, что ж, пусть расцветают все цветы, как говорил великий кормчий Мао Цзэдун. А может и не говорил. Короче, слово автору статьи:

Многоагентная архитектура открывает несколько мощных паттернов. Здесь я начну с основ и опишу, как с помощью Temporal сделать создание многоагентных систем простым, надёжным и увлекательным.

Почему стоит использовать многоагентные архитектуры?

ИИ-агенты мощны, но их можно перегрузить слишком большим количеством инструментов или слишком большим объёмом размышлений при ограниченном окне контекста. Возможно, вы захотите ограничить доступ конкретного агента или использовать для него подходящую модель. Кроме того, универсальные «делай-всё» агенты сложнее для оценки, чем специализированные агенты, хорошо выполняющие одну задачу.

Объединять агентов в единый процесс удобнее всего при помощи Temporal: его оркестрация надёжно управляет множеством задач как частями одного общего процесса.

В этом посте мы кратко обсудим оркестрацию с агентами-маршрутизаторами в Temporal, а затем глубже рассмотрим делегирование задач.

Определения терминов агентных систем

Вот несколько определений, которые помогут разобраться.

Агенты-маршрутизаторы

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

В этом примере наш пользователь пытается спланировать поездку. Он общается с маршрутизирующим агентом (Routing Agent), который направляет его к агенту бронирования авиабилетов (Flights Booking agent), чтобы найти мероприятия и забронировать рейсы.

Затем, после бронирования рейсов, система перенаправляет пользователя к агенту PTO (Paid Time Off), чтобы проверить его доступные дни отпуска и забронировать время отпуска.

Вы можете подробнее изучить маршрутизацию агентов в демо Temporal Agent здесь, а ниже есть видео с демонстрацией работы: https://youtu.be/8Dc_0dC14yY

Делегирование задач

Более мощная модель управления сложностью при большом числе агентов — делегировать обязанности агентам по задачам.

Думайте об этом как о команде специалистов, где каждый агент хорошо справляется со своей частью проблемы. Мы называем это «делегированием задач» (Task Delegation), и это отражает то, как команды работают в реальной жизни. Вспомните команду, в которой вы работаете в своей компании. У каждого есть своя роль, определённые зоны ответственности и протоколы взаимодействия. Ваши агенты могут использовать аналогичную структуру.

Если хотите сразу перейти к финалу и посмотреть результат в действии, посмотрите это видео: https://youtu.be/cnpmpzu1xTw

Простой пример: обнаружение и устранение проблемы с аналитическим агентом

Допустим, я — волшебник во вселенной Гарри Поттера и работаю в Scribbulus Supplies, магазине бумаги и магических товаров в Косом переулке. Ученики и преподаватели Хогвартса получают у меня все свои перья, пергамент, школьные принадлежности и т. п.

В последнее время я замечаю разные проблемы с заказами: не хватает запасов на складе, заказы требуют одобрения и т. д. Я не хочу решать эти проблемы вручную — я хочу построить систему, которая магически выполняет исправления заказов, чтобы мне не пришлось о них беспокоиться (в этом сезоне так много матчей по квиддичу).

К счастью, я точно знаю, как это исправить. Вот как.

Процесс исправления

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

@activity.defn
async def analyze_order_problems(input: dict) -> dict:
   #<snip> 
   # Define the messages for the LLM completion
    context_instructions = "You are a helpful assistant that detects and analyzes problems in orders. " \
    "Your task is to analyze the provided orders and identify any issues or anomalies. " \
    "You will receive a list of orders in JSON format, " \
    "each containing an 'order_id', 'order_date', 'status', 'items', and 'quantities'. " \
    "Look for common problems such as orders needing approval, orders stuck or delayed for various reasons for more than two weeks, " \
    "or other anomalies. " \
    "Ensure your response is valid JSON and does not contain any markdown formatting. " \
    "The response should be a JSON object with a key 'issues' that contains a list of detected issues, " \
    "each with an order_id, item with key 'issue' that describes the issue, " \
    "the customer_name the order is for, and  " \
    "a confidence_score of how sure you are there is a problem. " \
    "Feel free to include additional notes in 'additional_notes' if necessary. " \
    "If there are no issues, note that in additional_notes. " \
    "The list of orders to analyze is as follows: " \
    + json.dumps(orders_to_detect_json, indent=2)
    messages = [
        {
            "role": "system",
            "content": context_instructions
            + ". The current date is "
            + DATE_FOR_ANALYSIS.strftime("%B %d, %Y"),
        },
    ]

    try:
        completion_kwargs = {
            "model": llm_model,
            "messages": messages,
            "api_key": llm_key,
        }

        response = completion(**completion_kwargs)

        response_content = response.choices[0].message.content

    except Exception as e:
        activity.logger.error(f"Error in LLM completion: {str(e)}")
        raise

Агент выдаёт результат примерно в таком виде:

{
	"issues": [
		{
			"customer_name": "Rubeus Hagrid",
			"order_id": "ORD-004-RHG",
			"issue": "Order contains restricted item requiring approval.",
			"confidence_score": 0.9,
			"additional_notes": "Order includes 'Dragon Egg - Norwegian Ridgeback' which requires Ministry approval."
		}
]
}

Итак, с помощью ИИ мы можем определить, что заказ Хагрида требует одобрения перед упаковкой и отправкой. Я могу запускать эту Activity из Temporal Workflow, и она будет надёжно выполняться, оставаясь устойчивой к недоступности базы данных, медленным ответам или временным проблемам LLM. Агент-аналитик сообщает, что на 90% уверен в своём выводе о том, что заказ застрял из-за необходимости одобрения.

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

Подключение исправляющего агента

Давайте добавим в наш workflow ещё два шага: создание плана и выполнение плана.

Создание плана объединяет найденные проблемы, данные о запасах и заказах, а также инструменты, которые мы предоставляем агенту для исправления заказов, и формирует план их исправления.

Вот как выглядит план для заказа Хагрида:

{
	"proposed_tools": {
		"ORD-004-RHG": [
			{
				"tool_name": "request_approval_tool"
"confidence_score": 0.9,
				"tool_arguments": {
					"approval_request_contents": "Request to Approve Order",
					"approver": "approve-orders@diagonalley.co.uk",
					"order_id": "ORD-004-RHG"
				}				
			}
		]
	}
}

Выполнение плана — это шаг, на котором мы запускаем инструменты в соответствии с планом. Надёжность Temporal обеспечивает успешное выполнение инструментами своих задач.

Мы разделяем эти шаги, чтобы человек мог проверить план и утвердить его или отклонить.

Это требует, чтобы наш процесс умел ждать одобрения человеком, а также принимать ввод от человека. Temporal Workflows упрощают это. Вот как теперь выглядит наш Workflow:

@workflow.run
    async def run(self, inputs: dict) -> str:
        #execute the analysis agent
        await self.analyze_problems(self.context)

        # Execute the planning for this agent
        await self.create_plan(self.context)

        # Wait for the approval or reject signal
        await workflow.wait_condition(
            lambda: self.approved is not False or self.rejected is not False,
            timeout=timedelta(hours=12),
        )

        if self.rejected:
            return f"Repair REJECTED by user {self.context.get('rejected_by', 'unknown')}"

        # Proceed with the repair
        await self.execute_repair()

Добавление диалоговых возможностей с MCP

До этого момента я строил агентов, работающих в фоновом режиме. Давайте подключим их к диалоговому агенту через Model Context Protocol (MCP). В этом примере я использую goose, созданный нашими друзьями из Block.

@mcp.tool(description="Get the proposed tools for the repair workflow.",
          #tags={"repair", "order management", "workflow", "proposed tools"},
          )
async def get_proposed_tools(workflow_id: str, run_id: str) -> Dict[str, str]:
    """Return the proposed tools for the repair workflow. This is the result of the planning step. 
    This should not be confused with the tools that are actually executed.
    This won't have results before the planning step is complete."""
    load_dotenv(override=True)
    user = os.environ.get("USER_NAME", "Harry.Potter") 
    client = await get_temporal_client()
    handle = client.get_workflow_handle(workflow_id=workflow_id, run_id=run_id)

    try:
        planning_result: dict = await handle.query("GetRepairPlanningResult")
        proposed_tools_for_all_orders: dict = planning_result.get("proposed_tools", [])
        additional_notes = planning_result.get("additional_notes", "")
    except Exception as e:
        print(f"Error querying repair planning result: {e}")
        proposed_tools_for_all_orders = "No tools proposed yet."

    return {
        "proposed_tools": proposed_tools_for_all_orders,
        "additional_notes": additional_notes
    }

Я создал запрос (Query) в своём workflow, чтобы задействовать этот инструмент MCP:

@workflow.query
    async def GetRepairPlanningResult(self) -> dict:
        if "planning_result" not in self.context:
            raise ApplicationError("Planning result is not available yet.")
        return self.context["planning_result"]

Здесь вы видите, как я общаюсь с goose:

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

Мне не обязательно использовать goose — я могу подключаться к этим инструментам через чат в Slack, интегрируясь с Claude, интегрируясь с VSCode или с другими диалоговыми агентами, которые умеют работать с MCP.

Эти шаги образуют общий паттерн для агентных приложений:

Создание проактивного агента

С Temporal легко создавать проактивных агентов, которые могут мониторить и даже действовать самостоятельно. Я настрою этого агента на запуск и поиск проблем. Если проблемы обнаружены, агенты пойдут глубже: проанализируют их и предложат решения. Я настрою уведомление о найденных проблемах и ожидание моего просмотра и утверждения.

Вот как это выглядит в истории Workflow Temporal:

Одно из больших достоинств Temporal в том, что я могу видеть каждый шаг — входные данные, выходы и ошибки для каждого. Если какой-то шаг завершается сбоем, он будет автоматически повторяться (Retry), пока не выполнится успешно. Я могу быстро диагностировать любые проблемы и при необходимости исправить баги или подсказки и развернуть заново.

Вот как сейчас выглядит мой Workflow:

@workflow.defn
class RepairAgentWorkflowProactive(RepairAgentWorkflow):
    def __init__(self) -> None:
    # <snip> initialization

    @workflow.run
    async def run(self, inputs: dict) -> str:
        while True:
            self.approved = False

            await workflow.wait_condition(
                lambda: self.stop_waiting is True,
                timeout=timedelta(days=1), # Wait a day for the next detect->analysis->repair cycle.           
            )  

            # Execute the detection agent
            self.problems_found_confidence_score = await self.perform_detection(self.context)

            self.problems_found_confidence_score = self.context["detection_result"].get("confidence_score", 0.0)
            if self.problems_found_confidence_score < 0.5:
                analysis_notes = self.context["detection_result"].get("additional_notes", "")
                workflow.logger.info(f"Low confidence score from detection: {self.problems_found_confidence_score} ({analysis_notes}). No repair needed at this time.")
                self.set_workflow_status("NO-REPAIR-NEEDED")
                continue  # Skip to the next iteration if no repair is needed

            #execute the analysis agent      
            await self.analyze_problems(self.context)

            # Execute the planning for this agent
            await self.create_plan(self.context)

            # for low planning confidence scores, 
            # we will notify the user and wait for approval
            if self.planning_confidence_score <= 0.95:
                # Notify the user about the planned repairs
                await self.execute_notification()
                # Wait for the approval or reject signal
                await workflow.wait_condition(
                    lambda: self.approved is not False or self.rejected is not False,
                    timeout=timedelta(hours=20),  # Wait for up to 24 hours for approval
                )

                if self.rejected:
                    continue # Skip to the next iteration if repair is rejected

            else:
                # If the planning confidence score is high enough, we can proceed without waiting for approval
                self.approved = True
                self.context["approved_by"] = f"Agentically Approved with confidence score {self.context['planning_result'].get('overall_confidence_score', 0.0)}"

            # Proceed with the repair
            await self.execute_repair()

            # Create the report with the report agent
            report_summary = await self.generate_report()

Вот как сейчас выглядит наша архитектурная модель:

Наша система теперь проактивна: она самостоятельно выявляет и анализирует проблемы, автоматически исправляя их, если уверена в корректности исправлений. Если уверенность ниже 95%, она уведомляет меня и ждёт моего просмотра и одобрения. Я могу взаимодействовать с ней через MCP, сигналы и запросы Temporal Workflow, а также через goose, просматривая проблемы с заказами и предложенные исправления.

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

Паттерны проектирования

Вот паттерны проектирования и решения, которые я принял при построении этой системы:

Жизненный цикл и взаимодействие (Lifecycle and interaction)

Вопрос, который часто задают: «Стоит ли реализовывать агента как Workflow или как Activity?»

Для меня ответ таков: зависит от жизненного цикла и взаимодействия:

• Диалоговые агенты: интерактивные, долгоживущие, оркестрация: Workflows

• Вызовы LLM: вызов внешней системы: Activities

• Простые агенты автоматизации: недолговечные, без взаимодействия: Activities

• Долгоживущие агенты: длительная работа, часто есть взаимодействие: Workflows

• Проактивные агенты: интерактивные, долгоживущие: Workflows

DAPER

Эта система реализует паттерн, полезный для агентов, которые решают проблемы и действуют от имени пользователя:

• Detect: обнаруживает, где есть проблемы

• Analyze: понимает природу этих проблем

• Plan: создает план по устранению проблем

• Execute: выполняет план

• Report: анализирует систему и готовит отчет

Использование этого паттерна может включать:

• Проактивное обнаружение

• Поддержку по время звонка

• Мониторинг или SRE tooling

• Исправление транзакций на лету

Пробуйте!

Temporal наделяет агентов и инструменты сверхспособностями: надёжностью, наблюдаемостью, простотой. Workflows обеспечивают взаимодействие, оркестрацию, динамическое принятие решений, неограниченную длительность и автосохранение состояния и памяти. Activities обеспечивают надёжное выполнение внешних вызовов с автоматическими повторами.

Подписывайтесь на канал BPM Developers — про бизнес-процессы: новости, гайды, полезная информация и юмор.