DEV Community: orkes

Automating Serialization/Deserialization Tests with Orkes Conductor and LLMs

Orkes Developers — Thu, 29 May 2025 13:33:11 +0000

Writing unit tests for serialization and deserialization can be time-consuming and repetitive work. It’s monotonous and prone to human error, but still essential for ensuring that data is correctly sent across systems (API to server, application to application, SDK clients to server).

Unlike humans, software systems don’t get tired, bored, or distracted, even if faced with such mind-numbing work of writing serialization and deserialization (SerDe) tests.

Enter Orkes Conductor, a workflow orchestration platform. Combined with large language models (LLMs), Conductor transformed a week-long grind into a streamlined automation pipeline, cutting the testing time by over half.

In this post, we’ll walk through what SerDe tests are, why they matter, how we automated their creation using orchestration, and what made Orkes Conductor the ideal orchestration tool for the job.

Quick recap: serialization/deserialization

Serialization refers to the conversion of a data object from its current state into a format that can be stored in a database or transmitted through a network. For example, a Python object into JSON, or a Java object into a stream of bytes.

Deserialization is the exact opposite of serialization: it refers to the process of reconstructing the data object from its serialized format so that it can be used again. For example, from JSON back into a Python object so that it can be used in a function.

Serialization and deserialization are key processes for transmitting data across distributed systems in a standardized format while maintaining usability.

Why serialization/deserialization tests matter

Serialization and deserialization tests are critical for data integrity. They ensure that your data objects can:

Be accurately converted to/from different formats (such as from POJOs to JSON).
Retain structure and meaning across services.
Interoperate reliably between client-side and server-side components.

Neglecting these tests can introduce subtle, high-impact bugs, especially in systems where APIs evolve or where SDKs are maintained separately from backend logic.

Yet writing these tests is typically:

Manual: You have to craft JSON input/output cases for each class.
Boilerplate-heavy: The structure of each test is nearly identical, with minor differences in class names or references.
Error-prone: A wrong class reference or mismatched field can easily slip through.

That’s exactly what our engineering team was facing at Orkes: dozens of nearly identical tests with minor variations. Instead of slogging through them manually, we built an automated solution using Orkes Conductor.

Automating SerDe tests using Conductor + LLMs

In our case at Orkes, we needed to create tests to validate the mappings between SDK POJOs and server-side POJOs to ensure that data is transmitted correctly.

Accurate serialization and deserialization are crucial for the SDK's functionality, as they ensure that workflows and tasks are correctly interpreted by the Conductor server. By including comprehensive tests, developers can confidently make changes to the SDK without introducing regressions in data handling.

Using an orchestration tool like Orkes Conductor, we can automate repetitive actions across multiple services while ensuring that intermittent failures are remediated through retries and state persistence.

Here’s the high-level breakdown of the automated process:

Retrieve test resources like mapping files and JSON templates.
Generate a test branch in the repository.
Generate the SerDe tests using LLMs and prompt engineering.
Polish and publish the test code to the branch.

Let’s explore how Conductor’s enterprise functionalities enable seamless automation at each step.

1. Initial setup and data retrieval

Fetch a mapping file that links each SDK POJO to its corresponding server-side POJO.
Pull JSON templates generated from server POJOs. These templates will serve as the base for test generation.

{
  "templates" : {
    "IndexedDoc" : {
      "content" : {
        "score" : 123.456,
        "metadata" : {
          "sample_key" : "sample_value"
        },
        "docId" : "sample_docId",
        "text" : "sample_text",
        "parentDocId" : "sample_parentDocId"
      },
      "dependencies" : [ ]
    },
// rest of the JSON template

Parse the mapping file to each extract server-to-SDK relationship. Tests will be created for each server-SDK POJO mapping later.

Using built-in system tasks like HTTP Task and JSON JQ Transform Task, these steps can be quickly assembled without needing to code everything from scratch.

2. Branch management

Retrieve the SHA of the SDK repo’s target branch.
Use the SHA to create a new feature branch specifically for test generation.

Using dynamic references like “${variable}” in the HTTP Task’s URI https://api.github.com/repos/${previousTask.output.owner}/${previousTask.output.repo}/git/refs/heads/${workflow.input.sdkbranch}, the workflow is repeatable across multiple execution runs. Going beyond, the workflow becomes reusable and can be embedded in other workflows as well.

3. Iterative test generation using LLMs

Now we arrive at the heart of the workflow. For each server-SDK POJO mapping, the workflow will:

Fetch the SDK POJO source code and file SHA for version control.
Use the class path to generate a JSON representation.
Clean the Java class (remove annotations/comments) using GPT-4o.
Generate the actual test code via prompt engineering and LLM assistance.
Encode the test, create the proper file path, and commit the file to the new branch.

The suite of pre-built LLM tasks like LLM Text Complete facilitates rapid assembly of AI-powered tasks using any LLM provider. Conductor also supports prompt management, so you can create, test, and refine your prompts. Here is a snippet of an initial prompt to generate the test code:

4. Polish and publish

With all the test code generated, it’s time to wrap up:

Run Spotless to auto-format the code for style consistency.
Create a Pull Request for all the new test files.

Conductor tracks the data flow of every single task for each workflow run, making it easy to audit what has happened. Here is the execution trace of an example execution:

{
 "ownerApp": "user@example.com",
 "createTime": 1744977613276,
 "updateTime": 1744978814626,
 "createdBy": "user@example.com",
 "status": "COMPLETED",
 "endTime": 1744978814621,
 "workflowId": "af4f073b-1c4c-11f0-8645-4e34a87470b0",
 "tasks": [
   {
     "taskType": "HTTP",
     "status": "COMPLETED",
     "inputData": {
       "asyncComplete": false,
       "http_request": {
         "headers": {
           "Authorization": "token \${workflow.secrets.githubToken}",
           "Accept": "application/json"
         },
         "method": "GET",
         "uri": "https://raw.githubusercontent.com/orkes-io/sdk-codegen/base-code/conductor-java-sdk/SdkServerpojoMappings.json"
       },
     },
     "referenceTaskName": "getMappingFile",
     "retryCount": 0,
     "seq": 1,
     "pollCount": 1,
     "taskDefName": "get_mapping_file",
     "scheduledTime": 1744977613280,
     "startTime": 1744977613299,
     "endTime": 1744977613321,
     "updateTime": 1744977613299,
     "startDelayInSeconds": 0,
     "retried": false,
     "executed": true,
     "workflowInstanceId": "af4f073b-1c4c-11f0-8645-4e34a87470b0",
     "workflowType": "SDK_POJO_ser_deser_test",
     "taskId": "af4fa37c-1c4c-11f0-8645-4e34a87470b0",
     "callbackAfterSeconds": 0,
     "workerId": "orkes-workers-deployment-6c5b9f55dc-5xknz",
     "outputData": {
       "response": {
...

Why use Orkes Conductor?

Orchestration at scale

Conductor is built for chaining together complex, fault-tolerant workflows. This automation pipeline included over a dozen steps, with multiple dependencies across GitHub, OpenAI, and other services — exactly what Conductor was designed to handle.

LLM-ready

Not every task is rigid. For creative, generative work like cleaning up Java classes or writing test code, LLMs excel even when traditional automation breaks down. Conductor made it easy to drop in LLM-powered steps wherever flexible automation was required.

Reusable components

Once we built workers for custom tasks like applying Spotless auto-formatting and generating JSON templates, they became modular, reusable components. Now, creating a new workflow for a different use is just a matter of assembling these pieces together.

Developer-friendly

With REST APIs, SDKs, and a powerful UI, Conductor meets you where you are. Whether you need full developer tooling to build integrated workflows with your systems or visual dashboards to get complete visibility into the pipeline’s execution, Conductor’s product suite makes it seamless.

Scaling work from days to minutes

The original estimate for manually writing SerDe tests for this task was more than 5 days. Using Conductor and LLMs? Just 2.5 days to build and test the workflow itself. Now that the automated system exists, regenerating or extending the tests takes less than an hour.

More than just saving time, Orkes Conductor reframed how to think about developer productivity. Using Orkes Conductor with LLMs lets you:

Automate repetitive tasks, even those with variations.
Focus your energy on high-value work, like reviewing code and refining scalable systems.
Maximize your impact by reusing components across projects.

The next time you need to ask yourself: “Can I orchestrate this instead?”

Because chances are, with Orkes Conductor, the answer is yes. Take a stab at automating repetitive work using our free Developer Playground.

Workflow-Level Resilience in Orkes Conductor: Timeouts and Failure Workflows

Orkes Developers — Mon, 12 May 2025 09:07:15 +0000

Building resilient, production-grade workflows means preparing for the unexpected—from task stalls to external service outages. While task-level timeouts catch issues in isolated steps, workflow-level resilience settings act as a safety net for your entire orchestration. They ensure your system behaves predictably under stress and provides a graceful fallback when things go wrong.

In this post, we’ll explore two key features in Orkes Conductor that help you build robust workflows:

Workflow Timeouts
Failure Workflows (a.k.a. Compensation flows)

Workflow timeouts: Don’t let things hang

A workflow timeout defines how long a workflow is allowed to run before it's forcibly marked as timed out. This is crucial when your business logic needs to meet service-level agreements (SLAs) or avoid workflows stalling indefinitely.

Workflow timeout parameters

Parameter	Description
`timeoutSeconds`	Maximum duration (in seconds) for which the workflow is allowed to run. If the workflow hasn’t reached a terminal state within this time, it is marked as TIMED_OUT. Set to 0 to disable.
`timeoutPolicy`	Action to take when a timeout occurs. Supports: TIME_OUT_WF–Terminates the workflow as TIMED_OUT. ALERT_ONLY–Logs an alert but lets the workflow continue.

Use case: E-commerce checkout with 30-minute SLA

Imagine a checkout flow involving payment, inventory locking, and order confirmation. You don’t want stale carts holding inventory hostage for hours. A 30-minute timeout ensures the workflow either completes or fails cleanly.

Here’s a simplified implementation in Python using the Conductor SDK:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) HTTP task to fetch product price (simulated with dummy URL)
    fetch_random_number_task = HttpTask(
        task_ref_name="fetch_random_number",
        http_input={
            "uri": "https://www.random.org/integers/?num=1&min=1&max=100&col=1&base=10&format=plain&rnd=new",
            "method": "GET",
            "headers": {
                "Content-Type": "application/json"
            }
        }
    )

    # 2) Set variable for base price
    set_base_price = SetVariableTask(task_ref_name='set_base_price')
    set_base_price.input_parameters.update({
        'base_price': '${fetch_random_number.output.response.body}'
    })

    # 3) Inline task to calculate final price
    calculate_price_task = InlineTask(
        task_ref_name='calculate_final_price',
        script='''
            (function() {
                let basePrice = $.base_price;
                let loyaltyDiscount = $.loyalty_discount === "gold" ? 0.2 : 0;
                let promotionDiscount = $.promotion_discount ? 0.1 : 0;
                return basePrice * (1 - loyaltyDiscount - promotionDiscount);
            })();
        ''',
        bindings={
            'base_price': '${workflow.variables.base_price}',
            'loyalty_discount': '${workflow.input.loyalty_status}',
            'promotion_discount': '${workflow.input.is_promotion_active}',
        }
    )

    # 4) Set final calculated price
    set_price_variable = SetVariableTask(task_ref_name='set_final_price_variable')
    set_price_variable.input_parameters.update({
        'final_price': '${calculate_final_price.output.result}'
    })

    # Define the workflow with a 30-minute timeout
    workflow = ConductorWorkflow(
        name='checkout_workflow',
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.description = "E-commerce checkout workflow with 30-min timeout"
    workflow.timeout_seconds(1800)  # 30 minutes
    workflow.timeout_policy(TimeoutPolicy.TIME_OUT_WORKFLOW)

    workflow.add(fetch_random_number_task)
    workflow.add(set_base_price)
    workflow.add(calculate_price_task)
    workflow.add(set_price_variable)

    # Register the workflow definition
    workflow.register(overwrite=True)
    return workflow

Check out the full sample code for the E-commerce workflow.

If the workflow exceeds 30 minutes, it is marked as TIMED_OUT automatically, allowing you to alert a team, start a cleanup flow, or retry.

E-commerce workflow with a 30-minute timeout

Failure workflows: Your fallback plan

What happens when a workflow fails unexpectedly, due to a timeout, an API error, or an unhandled edge case? That’s where failure workflows come in.

These are separate workflows that are triggered when the main workflow fails. They allow you to compensate, clean up, and notify downstream systems or users.

Failure workflow parameters

Parameter	Description
`failureWorkflow`	The name of the fallback workflow to be triggered if this one fails. The default is empty.

Use case: Hotel booking with compensation flow

Let’s say your travel booking app orchestrates a hotel reservation workflow. If the booking fails (maybe the payment went through, but the room wasn’t confirmed), you’d want to:

Trigger a refund flow, and
Notify the customer that the booking failed

Main workflow code

def register_hotel_booking_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) HTTP task to reserve a hotel (simulated with dummy URL)
    reserve_hotel_task = HttpTask(
        task_ref_name="reserve_hotel",
        http_input={
            "uri": "https://httpbin.org/post",
            "method": "POST",
            "headers": {"Content-Type": "application/json"},
            "body": {
                "hotel_id": "${workflow.input.hotel_id}",
                "checkin": "${workflow.input.checkin_date}",
                "checkout": "${workflow.input.checkout_date}",
                "customer_id": "${workflow.input.customer_id}"
            }
        }
    )

    # 2) Set variable to confirm reservation status (simulate from body)
    set_status = SetVariableTask(task_ref_name='set_reservation_status')
    set_status.input_parameters.update({
        'reservation_status': '${reserve_hotel.output.response.body.json.status}'
    })

    # 3) Inline task to check booking status
    evaluate_reservation = InlineTask(
        task_ref_name='check_booking_status',
        script='''
            (function() {
                if ($.reservation_status !== 'confirmed') {
                    throw new Error("Booking failed");
                }
                return "confirmed";
            })();
        ''',
        bindings={
            'reservation_status': '${workflow.variables.reservation_status}'
        }
    )

    workflow = ConductorWorkflow(
        name='hotel_booking_workflow',
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.description = "Hotel reservation flow with SLA and failure handling"
    workflow.timeout_seconds(900)  # 15 minutes
    workflow.timeout_policy(TimeoutPolicy.TIME_OUT_WORKFLOW)
    workflow.failure_workflow("hotel_booking_failure_handler")

    workflow.add(reserve_hotel_task)
    workflow.add(set_status)
    workflow.add(evaluate_reservation)

    workflow.register(overwrite=True)
    return workflow

Failure workflow code

def register_failure_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Notify customer (simulated with dummy URL)
    notify_customer_task = HttpTask(
        task_ref_name="notify_customer",
        http_input={
            "uri": "https://httpbin.org/post",
            "method": "POST",
            "headers": {"Content-Type": "application/json"},
            "body": {
                "customer_id": "${workflow.input.customer_id}",
                "message": "Your hotel booking could not be completed. We apologize for the inconvenience."
            }
        }
    )

    # 2) Trigger refund (simulated with dummy URL)
    refund_payment_task = HttpTask(
        task_ref_name="trigger_refund",
        http_input={
            "uri": "https://httpbin.org/post",
            "method": "POST",
            "headers": {"Content-Type": "application/json"},
            "body": {
                "payment_id": "${workflow.input.payment_id}",
                "reason": "Hotel booking failed"
            }
        }
    )

    failure_workflow = ConductorWorkflow(
        name="hotel_booking_failure_handler",
        executor=workflow_executor
    )
    failure_workflow.version = 1
    failure_workflow.description = "Handles failed hotel bookings with customer notification and refund"

    failure_workflow.add(notify_customer_task)
    failure_workflow.add(refund_payment_task)

    failure_workflow.register(overwrite=True)
    return failure_workflow

Check out the full sample code for the hotel booking workflow.

Hotel booking workflow with a failure handler workflow

Best practices

Always define timeoutSeconds at both workflow and critical task levels to prevent resource overuse.
Use failureWorkflow for any workflow that produces side effects or artifacts that need cleanup in the event of failure.

Wrap up

Building production-ready workflows in Orkes Conductor means planning for both success and failure. Timeout policies and failure workflows aren’t just safeguards—they’re essential tools for maintaining system health, meeting SLAs, and ensuring a reliable user experience. When combined thoughtfully, they allow your workflows to self-regulate, recover from disruptions, and maintain a clean system state, even when things don’t go as planned.

—

Orkes Conductor is an enterprise-grade Unified Application Platform for process automation, API and microservices orchestration, agentic workflows, and more. Check out the full set of features, or try it yourself using our free Developer Playground.

Task-Level Resilience in Orkes Conductor: Timeouts and Retries in Action

Orkes Developers — Mon, 12 May 2025 08:52:28 +0000

In distributed systems, individual task failure is not a matter of if, but when. APIs go down, services stall, and workers disappear. What matters is how your system responds. With Orkes Conductor, you don’t just handle these failures—you design for them.

Conductor provides fine-grained control over how each task behaves under failure. With customizable timeouts and retries, you can recover from transient issues without human intervention, ensure critical steps don’t hang indefinitely, and build workflows that fail gracefully instead of catastrophically.

In this blog, we’ll explore three core capabilities that enable resilient task execution in Orkes Conductor:

Task Retries
Task Timeouts
System Task Timeouts

Task retries: Recovering from flaky failures

One of the most common failure scenarios is a transient error—momentary service unavailability, network hiccups, or throttling by an external API. Conductor lets you retry failed tasks automatically, using configurable backoff strategies to avoid overwhelming downstream services.

Retry parameters

Parameter	Description
`retryCount`	The maximum number of retry attempts. Default is 3.
`retryLogic`	Retry strategy for the tasks. Supports: FIXED–Retries after a fixed interval defined by `retryDelaySeconds`. LINEAR_BACKOFF–Retries occur with a delay that increases linearly based on `retryDelaySeconds x backoffScaleFactor x attempt_number`. EXPONENTIAL_BACKOFF–Retries occur with a delay that increases exponentially based on `retryDelaySeconds x (backoffScaleFactor ^ attempt_number`.
`retryDelaySeconds`	The delay between retries. This combines with the backoff logic to calculate the actual wait time. Note: The actual duration depends on the retry policy set in `retryLogic`.
`backoffScaleFactor`	Multiplier applied to `retryDelaySeconds` to adjust how fast delays increase. Default is 1.

Use case: Flaky email provider

Imagine your email provider fails intermittently. The first request sends a 500 error, but the second or third might succeed. This is a perfect scenario for retries.

from conductor.client.configuration.configuration import Configuration
from conductor.client.http.models import TaskDef
from conductor.client.orkes_clients import OrkesClients


def main():
    api_config = Configuration()
    clients = OrkesClients(configuration=api_config)
    metadata_client = clients.get_metadata_client()

    task_def = TaskDef()
    task_def.name = 'send_email_task'
    task_def.description = 'Send an email with retry on intermittent failures'
    task_def.retry_count = 3
    task_def.retry_logic = 'EXPONENTIAL_BACKOFF'
    task_def.retry_delay_seconds = 2
    task_def.backoff_scale_factor = 2

    metadata_client.register_task_def(task_def=task_def)

    print(f'Registered the task -- view at {api_config.ui_host}/taskDef/{task_def.name}')


if __name__ == '__main__':
    main()

Check out the full sample code for the send email task.

Here, if the email task fails, it will automatically retry up to 3 times with increasing delays—2s, 4s, and 8s—allowing time for the service to recover between attempts.

Email validation workflow using a Send Email task with exponential-backoff retries

Task timeouts: Preventing workflow stalls

Retries help you recover, but timeouts prevent you from getting stuck in the first place. Whether a worker goes offline or an external service hangs, task-level timeouts ensure your workflow doesn’t wait forever.

Timeout parameters

Parameter	Description
`pollTimeoutSeconds`	The time to wait for a worker to poll this task before marking it as TIMED_OUT.
`responseTimeoutSeconds`	The time to wait for a worker to send a status update (like IN_PROGRESS) after polling.
`timeoutSeconds`	Total time allowed for the task to reach a terminal state.
`timeoutPolicy`	Action to take when a timeout occurs: RETRY–Retries the task using retry settings. TIME_OUT_WF–Marks the whole workflow as TIMED_OUT. ALERT_ONLY–Logs an alert but lets the task continue.

Use case: Slow inventory API

Say you're calling a third-party inventory API that sometimes takes too long to respond. You don't want to wait forever, but you also don’t want to fail immediately. Here's how you'd configure a balanced timeout with retries:

from conductor.client.configuration.configuration import Configuration
from conductor.client.http.models import TaskDef
from conductor.client.orkes_clients import OrkesClients


def main():
    api_config = Configuration()
    clients = OrkesClients(configuration=api_config)
    metadata_client = clients.get_metadata_client()

    task_def = TaskDef()
    task_def.name = 'inventory_check_task'
    task_def.description = 'Check inventory status with timeout and retry settings'

    # Retry settings
    task_def.retry_count = 2
    task_def.retry_logic = 'FIXED'
    task_def.retry_delay_seconds = 5

    # Timeout settings
    task_def.timeout_seconds = 30
    task_def.poll_timeout_seconds = 10
    task_def.response_timeout_seconds = 15
    task_def.timeout_policy = 'RETRY'

    metadata_client.register_task_def(task_def=task_def)

    print(f'Registered the task -- view at {api_config.ui_host}/taskDef/{task_def.name}')


if __name__ == '__main__':
    main()

Check out the full sample code for the check inventory task.

This setup gives your worker 30 seconds to complete the task. If it doesn’t respond or fails, Conductor will retry it twice, waiting 5 seconds between each attempt.

Inventory workflow with a Check Inventory task with a 30-second timeout

Together, these retry and timeout configurations help you build workflows that are not just reactive, but resilient by design.

Next, we’ll look at how the same principles apply at the workflow level, giving you end-to-end control over your system’s behavior.

System task resilience

A key part of building resilient workflows is defining how long system tasks should wait on external services or heavy computations. In Orkes Conductor, each system task has default or configurable timeout settings to control this behavior.

HTTP task resilience

In Orkes Conductor, HTTP timeouts are defined by two parameters:

Parameter	Description	Default Values
`connectionTimeout`	The maximum time (in milliseconds) to establish a TCP connection to the remote server.	30 sec
`readTimeout`	The maximum time (in milliseconds) to wait for a response after the connection is established and the request is sent.	60 sec

In Orkes Conductor, these defaults are enforced to ensure platform stability.

Orkes Conductor making an HTTP request to an external server, illustrating timeout applications.

Internal timeouts resilience

Some system tasks have implicit timeout behaviors based on internal implementation. If you're designing workflows around system tasks, it's critical to understand and respect these limits.

System Task	Connection Timeout	Read Timeout
HTTP	30 sec	60 sec
LLM	60 sec	60 sec
Opsgenie	30 sec	60 sec
Inline	n/a	4 sec (max execution time)
Business Rule	10 sec	120 sec

Wrap up

Task failures are unavoidable, but with proper retry and timeout configurations, they don’t have to break your workflows. Conductor’s task-level resilience features help you avoid cascading failures, handle transient issues gracefully, and prevent workflows from hanging indefinitely.

In the next article, we’ll scale this approach up and explore workflow-level failure handling strategies like timeout policies and compensation flows that give you end-to-end resilience.

Next up:

Workflow-Level Resilience

—

Fail Fast, Recover Smart: Timeouts, Retries, and Recovery in Orkes Conductor

Orkes Developers — Mon, 12 May 2025 08:31:39 +0000

In distributed systems, failure isn’t a possibility—it’s a certainty. APIs hang, workers crash, and networks drop. What matters isn’t avoiding failure, but recovering from it quickly and cleanly.

Orkes Conductor gives you the tools to do exactly that. Timeouts and retries aren’t just configurable—they’re core to how Conductor ensures reliability at scale. In this post, you’ll learn how to use them effectively at both the task and workflow levels, how they interact with failure workflows, and how to design resilient systems that heal themselves without manual intervention.

Let’s dive in.

Why failure handling is critical in workflow applications

Imagine you're orchestrating a payment flow. After the user places an order, your workflow triggers tasks to charge the card, update inventory, and send a confirmation email.

All good—until the inventory API goes silent.

Now you're in a mess: the card is charged, but inventory isn’t updated, and the user doesn’t get a confirmation. This is exactly the kind of situation timeouts and retries are designed to avoid.

Unresponsive HTTP task in an inventory workflow

In Orkes Conductor, timeouts and retries work together to provide resilience and control:

Timeouts prevent your workflow from hanging indefinitely.
Retries recover from transient errors without manual intervention.

Together, they help you gracefully handle:

Unresponsive services: If a downstream API is taking too long, timeout the task and retry or move on.
Crashed or unresponsive workers: If a worker crashes mid-task, timeouts ensure the task doesn’t hang forever.
Transient errors: If a task fails quickly (like a network blip or 500 error), retries can automatically reattempt the operation.
Misconfigured or under-provisioned systems: If no worker is available to pick up a task, timeouts ensure the system doesn’t just wait forever.

This is more than resilience—it’s control. Instead of writing custom error-handling logic for every edge case, you get:

At-least-once delivery
Automatic retry logic
Configurable timeouts
Self-healing workflows

Task timeouts vs task retries

As you design resilient workflows, it’s important to understand the difference between timeouts and retries for individual tasks—they solve different problems:

Aspect	Retry	Timeout
What it is	A second (or third, etc.) attempt to run the same task after it fails.	A time limit for how long a task is allowed to run (or how long you wait for it to start/progress).
Trigger	Happens after a task fails (e.g., returns FAILED).	Happens if the task takes too long to respond to or complete.
Example	API call fails with 500 Internal Server Error → Retry it after a delay.	API call hangs with no response after 30 seconds → Timeout triggers a recovery action.
Configured via	`retryCount`, `retryLogic`, `retryDelaySeconds`, `backoffScaleFactor`	`timeoutSeconds`, `pollTimeoutSeconds`, `responseTimeoutSeconds`, `timeoutPolicy`

In short:

Retries deal with tasks that fail fast.
Timeouts deal with tasks that hang slow.

Both are essential for robust, fault-tolerant tasks.

Failed worker resulting in task-timeout and retries

Workflow timeouts and failure workflows

Just like tasks, entire workflows can time out or fail, and Conductor gives you powerful tools to handle those cases gracefully.

Aspect	Workflow Timeout	Failure Workflow
What it is	A maximum time limit for the whole workflow to complete.	A backup workflow that runs automatically when the original workflow fails.
Trigger	Happens when the workflow exceeds its configured timeout.	Happens when the workflow ends in FAILED or TERMINATED status.
Example	A data pipeline takes too long (e.g., > 2 hours) → Workflow is marked TIMED_OUT.	A workflow fails due to repeated task failures → A failure workflow sends alerts and logs diagnostics.
Configured via	`timeoutSeconds`, `timeoutPolicy`	`failureWorkflow`

In short:

Workflow timeouts protect against indefinitely running flows.
Failure workflows let you define recovery logic and postmortems for failed flows.

Together, they make workflows self-aware and recoverable.

Long-running worker exceeds workflow timeout and triggers compensation workflow

Combining retries, timeouts, and failure Workflows

Retries give tasks more chances. Timeouts limit how long each attempt can run. Workflow timeouts cap the total duration of the workflow. If all else fails, failure workflows take over.

Here’s how it all fits together:

Task fails → Task Retry: it’s retried (up to retryCount).
Each retry has a timeout → Task Timeout: Limits how long you wait per attempt.
Too many failures or long delay → Workflow Timeout: Workflow fails or times out.
Workflow fails → Failure Workflow: A failure workflow runs to clean up or alert.

This layered approach gives you fault-tolerance at every level:

Default vs custom configuration: When to override

Orkes Conductor ships with sane defaults.

Task retry settings

{
  "retryCount": 3,
  "retryDelaySeconds": 60,
  "retryLogic": "FIXED",
  "backoffScaleFactor": 1
}

Task timeout settings

{
  "responseTimeoutSeconds": 600,
  "timeoutSeconds": 3600,
  "pollTimeoutSeconds": 3600,
  "timeoutPolicy": "TIME_OUT_WF"
}

Workflow settings

{
  "timeoutSeconds": 0,
  "timeoutPolicy": "ALERT_ONLY",
  "failureWorkflow": ""
}

These are a solid starting point, but not always ideal for your SLAs.

Override the defaults when:

You have strict latency goals, such as completing checkout in under 5 minutes.
You interact with unreliable third-party APIs, like a flaky partner API.
You need predictable system behavior under load or failure.

On the flip side, don’t override just for the sake of it. Defaults help reduce configuration sprawl and are good enough for many internal service workflows.

Wrap up

In distributed systems, failure is inevitable, but downtime doesn’t have to be. With Orkes Conductor, resilience is built in, not bolted on.

Timeouts help prevent indefinite hangs, and retries offer automatic recovery from transient issues. Workflow-level timeouts and failure workflows ensure your entire system can fail gracefully and bounce back, without manual intervention.

By combining these tools thoughtfully, you get more than reliability—you get confidence—confidence that your workflows will keep running, even when parts of your system don’t.

So, whether you're orchestrating high-stakes payment flows or background data pipelines, remember: design for failure, recover automatically, and keep moving forward.

Go more in-depth with task and workflow resilience with these follow-up articles:

—

Orkes Operators: Parallelism and Reusability

Orkes Developers — Mon, 28 Apr 2025 12:57:39 +0000

Workflows don’t just move forward—they branch, scale, and reuse logic. Whether you’re processing tasks in parallel, delegating work to subflows, or triggering asynchronous side-jobs, these patterns let you build smarter, more modular orchestration.

In this post, we’ll explore operators that boost concurrency and composability in Orkes Conductor:

Fork/Join for running multiple branches in parallel
Dynamic Fork for scalable, input-driven concurrency
Sub Workflow for reusing logic across workflows
Start Workflow for async, fire-and-forget execution

From concurrent notifications to reusable payment flows, these operators help you scale orchestration cleanly—without duplicating logic or overcomplicating control flow. Let’s dive in.

Fork/Join

The Fork/Join task in Orkes Conductor enables parallel execution of multiple task branches within a workflow. When a Fork task is reached, it splits the workflow into separate paths that run concurrently, allowing different sequences of tasks to execute in parallel. Once all branches are complete, a Join task is used to synchronize them before continuing with the rest of the workflow. This pattern is ideal for optimizing performance and handling tasks that can be executed independently, such as running multiple validation checks or data fetch operations simultaneously.

Example implementation

A common use case for the Fork/Join task is in a multi-channel notification workflow. Suppose a Fork task dispatches three parallel notifications—email, SMS, and HTTP. Each branch sends its respective message independently to maximize concurrency. Since HTTP responses can be retried and don’t need to block the overall flow, the Join task is configured with joinOn and is set to only the email_notification and sms_notification tasks. This allows the workflow to proceed as soon as the critical email and SMS tasks are complete, while the HTTP notification continues in the background. This setup enables partial synchronization—balancing reliability with improved efficiency.

Here’s the workflow visualized:

Workflow using a Fork/Join operator

Here’s the code snippet for creating the workflow in code:


def register_notification_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Define each notification path
    email_branch = [
        SimpleTask(
            task_def_name="process_notification_payload",
            task_reference_name="process_notification_payload_email"
        ),
        SimpleTask(
            task_def_name="email_notification",
            task_reference_name="email_notification_ref"
        )
    ]

    sms_branch = [
        SimpleTask(
            task_def_name="process_notification_payload",
            task_reference_name="process_notification_payload_sms"
        ),
        SimpleTask(
            task_def_name="sms_notification",
            task_reference_name="sms_notification_ref"
        )
    ]

    http_branch = [
        SimpleTask(
            task_def_name="process_notification_payload",
            task_reference_name="process_notification_payload_http"
        ),
        HttpTask(
            task_ref_name="http_notification_ref",
            http_input={
                "uri": "${workflow.input.http_target_url}",
                "method": "POST",
                "headers": {
                    "Content-Type": "application/json"
                },
                "body": {
                    "message": "Notification triggered"
                }
            }
        )
    ]

    # 2) Fork-Join setup (only join on email + sms)
    fork_join = ForkTask(
        task_ref_name="my_fork_join_ref",
        forked_tasks=[email_branch, sms_branch, http_branch],
        join_on=["email_notification_ref", "sms_notification_ref"]
    )

    workflow = ConductorWorkflow(
        name="notification_workflow_with_fork_join",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(fork_join)
    workflow.register(overwrite=True)

    return workflow


@worker_task(task_definition_name="process_notification_payload")
def process_notification_payload() -> dict:
    print("🛠️ Processing notification payload...")
    return {"payload_processed": True}


@worker_task(task_definition_name="email_notification")
def email_notification() -> dict:
    print("📧 Email sent to test@example.com")
    return {
        "email_sent_at": "2021-11-06T07:37:17+0000",
        "email_sent_to": "test@example.com"
    }


@worker_task(task_definition_name="sms_notification")
def sms_notification() -> dict:
    print("📱 SMS sent to +1-xxx-xxx-xxxx")
    return {
        "sms_sent_at": "2021-11-06T07:37:17+0129",
        "sms_sent_to": "+1-xxx-xxx-xxxx"
    }

Check out the full sample code for the Fork/Join Workflow.

Dynamic Fork

The Dynamic Fork task allows you to create and execute multiple parallel task branches based on dynamic input at runtime. Unlike a standard fork, which defines branches statically, Dynamic Forks generates parallel tasks on the fly using input parameters. This is particularly useful when the number of tasks isn’t known ahead of time—such as sending notifications to a dynamic list of users or processing a batch of files. Each dynamically generated task runs independently, and a corresponding Join task is used to synchronize them once all have been completed.

Example implementation

Imagine a content moderation workflow that needs to process different types of user-generated content—text, images, and videos. Each content type requires a different moderation approach, like scanning text for profanity, running image recognition for flagged visuals, or transcribing and analyzing video audio. Since the type and number of content items vary per request, a Dynamic Fork is ideal. At runtime, it dynamically generates tasks like moderate_text, moderate_image, and moderate_video, each configured with input specific to the content item. Once all moderation tasks are complete, a Join task consolidates the results for final review. This setup ensures flexibility and scalability for diverse, real-time moderation needs.

Here’s the workflow visualized:

Workflow using a Dynamic Fork operator

Here’s the code snippet for creating the workflow in code:


def register_moderation_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Define Join task
    join_task = JoinTask(
        task_ref_name="join_ref",
        join_on=[]
    )

    # 2) Define Dynamic Fork Task with inputParameters dict
    dynamic_fork = DynamicForkTask(
        task_ref_name="fork_join_dynamic_ref",
        join_task=join_task
    )
    dynamic_fork.input_parameters.update({
        "dynamicTasks": "${workflow.input.dynamicTasks}",
        "dynamicTasksInputs": "${workflow.input.dynamicTasksInputs}"
    })

    workflow = ConductorWorkflow(
        name="DynamicForkExample",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(dynamic_fork)
    workflow.register(overwrite=True)

    return workflow


@worker_task(task_definition_name="moderate_text")
def moderate_text() -> dict:
    print("📝 Moderating text...")
    return {"status": "Text moderation complete"}


@worker_task(task_definition_name="moderate_image")
def moderate_image() -> dict:
    print("🖼️ Moderating image...")
    return {"status": "Image moderation complete"}


@worker_task(task_definition_name="moderate_video")
def moderate_video() -> dict:
    print("🎥 Moderating video...")
    return {"status": "Video moderation complete"}

Check out the full sample code for the Dynamic Fork Workflow.

Sub Workflow

The Sub Workflow task in Orkes Conductor allows you to encapsulate and execute another workflow synchronously within the context of a parent workflow. This task enables modularity and reusability, letting you delegate specific parts of a workflow to be handled by a separate workflow definition. The sub-workflow can run independently and return results to the parent workflow, helping to simplify complex workflows and manage different logic in isolation. It also provides better scalability and maintainability by breaking down large workflows into smaller, reusable components.

Example implementation

In a subscription management workflow, the Sub Workflow task is used to modularize payment processing. Instead of duplicating the entire payment logic within the subscription workflow, the payment_for_subscription workflow is called as a sub-workflow whenever payment needs to be processed. This ensures that any updates made to the payment workflow are automatically reflected across all parent workflows, improving maintainability and consistency. The parent workflow can invoke the sub-workflow multiple times, allowing for seamless integration of payment logic into the larger subscription process.

Here’s the workflow visualized:

Workflow using a Sub Workflow operator

Here’s the code snippet for creating the workflow in code:


# --- Payment Workflow (Sub-Workflow for Subscription) ---

def register_payment_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Process payment (for subscription)
    process_payment = SimpleTask(
        task_def_name="process_payment",
        task_reference_name="process_payment"
    )
    process_payment.input_parameters.update({
        "amount": "${workflow.input.amount}"
    })

    workflow = ConductorWorkflow(
        name="payment_for_subscription",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(process_payment)
    workflow.register(overwrite=True)

    return workflow


# --- Subscription Workflow Definition (using Sub-Workflow) ---

def register_subscription_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Check if the user is subscribed
    check_status = SimpleTask(
        task_def_name="check_subscription_status",
        task_reference_name="check_subscription_status"
    )

    # 2) Trigger payment using sub-workflow
    trigger_payment = SubWorkflowTask(
        task_ref_name="trigger_payment",
        workflow_name="payment_for_subscription",  # The sub-workflow to invoke
        version=1
    )
    trigger_payment.input_parameters.update({
        "amount": "${workflow.input.amount}",  # Mapping input to sub-workflow
        "subscription_id": "${workflow.input.subscription_id}",
        "user_id": "${workflow.input.user_id}"
    })

    # 3) Send welcome email
    send_email = SimpleTask(
        task_def_name="send_welcome_email",
        task_reference_name="send_welcome_email"
    )

    workflow = ConductorWorkflow(
        name="subscription_management",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(check_status)
    workflow.add(trigger_payment)
    workflow.add(send_email)
    workflow.register(overwrite=True)

    return workflow


@worker_task(task_definition_name="check_subscription_status")
def check_subscription_status() -> dict:
    print("📡 Checking subscription status...")
    return {"status": "active"}


@worker_task(task_definition_name="send_welcome_email")
def send_welcome_email() -> dict:
    print("📨 Sending welcome email to subscriber...")
    return {"status": "email_sent"}


@worker_task(task_definition_name="process_payment")
def process_payment(amount) -> dict:
    print(f"💰 Charging customer: ${amount.input_data['amount']}")
    return {"status": "Payment successful"}

Check out the full sample code for the Sub Workflow Workflow.

Start Workflow

The Start Workflow task is an operator in Orkes Conductor that launches another workflow asynchronously from within the current one. Unlike the Sub Workflow task, which pauses the parent workflow until the child workflow completes, the Start Workflow task triggers the target workflow and immediately continues execution without waiting for its result. This is ideal for use cases like fire-and-forget processing, background jobs, or decoupled side operations where real-time feedback isn't needed, allowing for greater concurrency and throughput in orchestration.

Example implementation

A common use case for the Start Workflow task is submitting background jobs that don’t need to block the main workflow. For example, in an e-commerce checkout workflow, once the payment is processed, a Start Workflow task can be used to asynchronously trigger an email_receipt_workflow that sends a receipt to the customer. This ensures the checkout flow completes quickly without waiting for email processing, which can be handled independently.

Here’s the workflow visualized:

Workflow using a Start Workflow operator

Here’s the code snippet for creating the workflow in code:


# --- Email Receipt Workflow (Sub-Workflow for emails) ---

def register_email_receipt_workflow(executor: WorkflowExecutor) -> ConductorWorkflow:
    email_task = SimpleTask(
        task_def_name="send_receipt_email",
        task_reference_name="send_receipt_email_ref"
    )

    workflow = ConductorWorkflow(
        name="email_receipt_workflow",
        executor=executor
    )
    workflow.version = 1
    workflow.add(email_task)
    workflow.register(overwrite=True)
    return workflow


# --- Main Checkout Workflow Definition (using Sub-Workflow) ---

def register_checkout_workflow(executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Task to process payment
    payment_task = SimpleTask(
        task_def_name="process_payment",
        task_reference_name="process_payment_ref"
    )

    # 2) Task to asynchronously start email_receipt_workflow
    start_email_workflow_input = StartWorkflowRequest(
        name="email_receipt_workflow",
        version=1,
        input={
            "customer_email": "${workflow.input.customer_email}",
            "order_id": "${workflow.input.order_id}"
        }
    )

    start_email_workflow_task = StartWorkflowTask(
        task_ref_name="start_email_workflow_ref",
        workflow_name="email_receipt_workflow",
        start_workflow_request=start_email_workflow_input,
        version=1
    )

    workflow = ConductorWorkflow(
        name="checkout_workflow",
        executor=executor
    )
    workflow.version = 1
    workflow.add(payment_task)
    workflow.add(start_email_workflow_task)
    workflow.register(overwrite=True)
    return workflow

@worker_task(task_definition_name="process_payment")
def process_payment() -> dict:
    print("💳 Payment processed successfully.")
    return {"payment_status": "success"}


@worker_task(task_definition_name="send_receipt_email")
def send_receipt_email() -> dict:
    print("📧 Receipt email sent to customer.")
    return {"email_status": "sent"}

Check out the full sample code for the Start Workflow Workflow.

Wrap up

Parallelism and reusability aren’t just performance optimizations—they’re architectural superpowers. With operators like Fork/Join, Dynamic Fork, Sub Workflow, and Start Workflow, Orkes Conductor makes it easy to run tasks concurrently, compose modular logic, and trigger workflows asynchronously.

No threads to manage, no glue code to write—just clean, declarative orchestration that scales.

—

Orkes Operators: Loops, Waits, and Human-in-the-Loop

Orkes Developers — Mon, 28 Apr 2025 12:44:18 +0000

Not all workflows are linear. Some wait, some repeat, and some need a human to step in before moving forward.

In this post, we explore three essential operators that add patience, persistence, and people to your workflows:

Do While for flexible, condition-based iteration
Wait for clean, pause-and-resume control
Human for structured human input and approvals

Whether you're looping through review stages, waiting for external input, or routing manual approvals across teams, these operators help you build dynamic, resilient, and human-aware workflows without writing custom orchestration logic.

Do While

The Do While task is a loop operator that repeatedly executes a set of tasks as long as a specified condition remains true. Similar to a do-while loop in programming, it ensures the loop body runs at least once before evaluating the condition. This is useful for polling, retries, approval loops, or any scenario requiring repeated execution until a dynamic condition is met. The loop condition is evaluated after each iteration, and loop variables can be updated within the loop to control its execution.

Example implementation

This workflow handles a ticket that must go through multiple internal review stages: support agent, QA team, and team lead. Using a Do While Task, the workflow loops through these stages automatically, sending the ticket to each reviewer in sequence. The loop runs once for each reviewer using the itemslist and exits cleanly once all reviews are complete.

Here’s the workflow visualized:

Workflow using a Do While operator

Here’s the code snippet for creating the workflow in code:


def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Task to simulate review step with HTTP POST
    review_task = HttpTask(
        task_ref_name="log_review_comment",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/posts",
            "method": "POST",
            "headers": {
                "Content-Type": "application/json"
            },
            "body": {
                "reviewer": "${review_ticket_loop.output.item}",
                "comment": "Reviewed and updated by ${review_ticket_loop.output.item}"
            }
        }
    )

    # 2) DoWhileTask to iterate over reviewer stages
    review_loop = DoWhileTask(
        task_ref_name="review_ticket_loop",
        termination_condition="$.review_ticket_loop.iteration < $.review_ticket_loop.input.items.length",
        tasks=[review_task]
    )
    review_loop.input_parameters.update({
        "items": ["support_agent", "qa_team", "team_lead"]
    })

    workflow = ConductorWorkflow(
        name="ticket_review_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(review_loop)
    workflow.register(overwrite=True)

    return workflow

Check out the full sample code for the Do While Workflow.

Wait

The Wait task is a control flow operator used to pause workflow execution until an external signal is received. It’s ideal for scenarios that require human input, external approvals, or asynchronous events from another system. When the workflow reaches a Wait task, it halts and remains paused until it's explicitly resumed using the Conductor API or UI. This makes it easy to build workflows that involve manual steps or depend on external triggers without resorting to constant polling or custom state management.

Example implementation

In an employee onboarding workflow, after provisioning accounts and sending welcome materials, the process may require the new hire to complete a digital form or provide verification documents. A Wait task can pause the workflow at this point until the user submits the required information. Once the input is received—via a webhook or manual trigger—the workflow resumes automatically, continuing with the next steps like assigning a manager or scheduling training.

Here’s the workflow visualized:

Workflow using a Wait operator

Here’s the code snippet for creating the workflow in code:


def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Simulate account provisioning
    provision_account = HttpTask(
        task_ref_name="provision_account",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/users",
            "method": "POST",
            "headers": {
                "Content-Type": "application/json"
            },
            "body": {
                "employeeId": "${workflow.input.employeeId}",
                "email": "${workflow.input.email}"
            }
        }
    )

    # 2) Send welcome materials
    send_welcome = HttpTask(
        task_ref_name="send_welcome_materials",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/posts",
            "method": "POST",
            "headers": {
                "Content-Type": "application/json"
            },
            "body": {
                "employeeId": "${workflow.input.employeeId}",
                "message": "Welcome to the company!"
            }
        }
    )

    # 3) Wait for the new employee to submit verification (simulate with 10s wait)
    wait_for_input = WaitTask(
        task_ref_name="wait_for_employee_response",
        wait_for_seconds=10
    )

    # 4) Assign a manager
    assign_manager = SetVariableTask(task_ref_name="assign_manager")
    assign_manager.input_parameters.update({
        "manager": "alice.smith@company.com"
    })

    # 5) Schedule training
    schedule_training = SetVariableTask(task_ref_name="schedule_training")
    schedule_training.input_parameters.update({
        "training_date": "2025-04-20"
    })

    workflow = ConductorWorkflow(
        name="employee_onboarding_workflow",
        executor=workflow_executor
    )
    workflow.version = 1

    workflow.add(provision_account)
    workflow.add(send_welcome)
    workflow.add(wait_for_input)
    workflow.add(assign_manager)
    workflow.add(schedule_training)

    workflow.register(overwrite=True)
    return workflow

Check out the full sample code for the Wait Workflow.

Human

The Human task in Orkes Conductor is used to pause a workflow and wait for manual input or approval from a human before continuing. It's ideal for workflows that involve human-in-the-loop decisions, such as document reviews, access approvals, or data verification. When a Human task is reached, the workflow enters a WAITING state until a user provides the required input through the API or UI. This enables seamless integration of manual steps into automated workflows while maintaining visibility and control over the process.

Example implementation

In this example, an expense approval workflow uses two Human tasks to gather approvals from different departments. First, the HR department is prompted with a form showing the expense details in read-only fields. If HR approves, the workflow continues to the Finance department for a second round of approval using the same form template. Upon Finance’s approval, a notification workflow is triggered. If either department rejects the request, the workflow terminates early. This use case shows how Human tasks enable structured, multi-step approvals with manual input inside an otherwise automated process.

Here’s the workflow visualized:

Workflow using a Human operator

To create this workflow in Conductor, you must first define the expense_approval_form using the appropriate fields in the Orkes UI. Here’s the code snippet for creating the expense_approval_form in JSON:


{
  "createTime": 1744849075315,
  "updateTime": 1744849570917,
  "createdBy": "USER:john.doe@acme.com",
  "updatedBy": "USER:john.doe@acme.com",
  "name": "expense_approval_form",
  "version": 1,
  "jsonSchema": {
    "$schema": "http://json-schema.org/draft-07/schema",
    "properties": {
      "approvalStatus": {
        "type": "string",
        "enum": [
          "APPROVED",
          "REJECTED"
        ]
      }
    },
    "required": [
      "approvalStatus"
    ]
  },
  "templateUI": {
    "type": "VerticalLayout",
    "elements": [
      {
        "type": "Control",
        "scope": "#/properties/approvalStatus",
        "label": "Department Approval",
        "options": {}
      }
    ]
  }
}

Here’s the code snippet for creating the workflow in code:


# --- Notification Workflow ---

def register_notification_workflow(executor: WorkflowExecutor) -> ConductorWorkflow:
    notify_task = SimpleTask(
        task_def_name="notify_expense_approval",
        task_reference_name="notify_expense_approval_ref"
    )

    workflow = ConductorWorkflow(
        name="notification_workflow",
        executor=executor
    )
    workflow.version = 1
    workflow.add(notify_task)
    workflow.register(overwrite=True)
    return workflow


# --- Expense Approval Workflow ---

def register_expense_approval_workflow(executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) HR Approval Step
    hr_approval = HumanTask(
        task_ref_name="hr_approval_task",
        display_name="HR Approval",
        form_template="expense_approval_form",
        form_version=1,
        assignment_completion_strategy=AssignmentCompletionStrategy.TERMINATE
    )

    check_hr = SwitchTask(
        task_ref_name="check_hr_approval",
        case_expression="${hr_approval_task.output.approvalStatus}"
    ).switch_case("APPROVED", [])

    # 2) Finance Approval Step
    finance_approval = HumanTask(
        task_ref_name="finance_approval_task",
        display_name="Finance Approval",
        form_template="expense_approval_form",
        form_version=1,
        assignment_completion_strategy=AssignmentCompletionStrategy.TERMINATE
    )

    check_finance = SwitchTask(
        task_ref_name="check_finance_approval",
        case_expression="${finance_approval_task.output.approvalStatus}"
    ).switch_case("APPROVED", [])

    # 3) Start Notification Workflow
    start_notification = StartWorkflowTask(
        task_ref_name="start_notification_workflow",
        workflow_name="notification_workflow",
        start_workflow_request=StartWorkflowRequest(
            name="notification_workflow",
            version=1,
            input={
                "expense_id": "${workflow.input.expense_id}",
                "submitted_by": "${workflow.input.submitted_by}"
            }
        ),
        version=1
    )

    # Add the notification task inside the finance approval success branch
    check_finance.switch_case("APPROVED", [start_notification])

    # Add finance approval after HR approves
    check_hr.switch_case("APPROVED", [finance_approval, check_finance])

    workflow = ConductorWorkflow(
        name="expense_approval_workflow",
        executor=executor
    )
    workflow.version = 1
    workflow.add(hr_approval)
    workflow.add(check_hr)
    workflow.register(overwrite=True)
    return workflow


# --- Worker for notification task ---

@worker_task(task_definition_name="notify_expense_approval")
def notify_expense_approval() -> dict:
    print("🔔 Expense approval notification sent.")
    return {"notified": True}

Check out the full sample code for the Human Workflow.

Wrap up

Loops, waits, and human steps bring real-world rhythm to your workflows, allowing them to persist, pause, or involve people when needed. With tasks like Do While, Wait, and Human, Conductor makes asynchronous logic declarative, structured, and resilient. No brittle timers, no custom coordination—just clean, event-driven flow.

Next up:

Parallelism and Reusability

Orkes Operators: Branching and Conditionals

Orkes Developers — Mon, 28 Apr 2025 12:10:02 +0000

Smart workflows don’t just follow a straight line—they make decisions.

Orkes Conductor provides a set of declarative operators that let your workflows respond to input, handle conditions, and dynamically route execution without hardcoding logic into your services.

In this article, we’ll walk through four core decision-making operators:

Switch for branching logic based on runtime conditions
Dynamic for generating tasks on the fly based on data
Set Variable for setting and updating internal state during execution
Terminate for gracefully ending workflows early when needed

With these operators, you can model flexible, maintainable decision paths that react to real-world complexity—all without writing imperative glue code.

Switch

The Switch task is a control flow operator that enables conditional branching within a workflow, similar to a switch-case or if-else statement in code. It evaluates an expression at runtime and routes execution to a matching case, with support for a default path if no match is found. This makes it ideal for directing workflow logic based on dynamic values such as user input, AI-powered decisions, or status flags.

Example implementation

Let’s say you're building an order processing workflow that needs to handle different shipping methods. Based on the shippingType selected by the user (standard, express, or overnight), the workflow should branch into a different fulfillment path. A Switch task can evaluate the shippingType and route the workflow accordingly.

Here’s the workflow visualized:

Workflow using a Switch operator.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Task to fetch order details
    fetch_order_details = HttpTask(
        task_ref_name="fetch_order_details",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/posts/${workflow.input.orderId}",
            "method": "GET",
            "headers": {
                "Content-Type": "application/json"
            }
        }
    )

    # 2) SetVariable tasks for each shipping method
    standard_shipping = SetVariableTask(task_ref_name="standard_shipping_var")
    standard_shipping.input_parameters.update({
        "selected_shipping_method": "standard"
    })

    express_shipping = SetVariableTask(task_ref_name="express_shipping_var")
    express_shipping.input_parameters.update({
        "selected_shipping_method": "express"
    })

    overnight_shipping = SetVariableTask(task_ref_name="overnight_shipping_var")
    overnight_shipping.input_parameters.update({
        "selected_shipping_method": "overnight"
    })

    # 3) Define the Switch task to route based on shippingType
    switch_shipping = SwitchTask(
        task_ref_name="switch_shipping_method",
        case_expression="${workflow.input.shippingType}"
    )
    switch_shipping \
        .switch_case("standard", [standard_shipping]) \
        .switch_case("express", [express_shipping]) \
        .switch_case("overnight", [overnight_shipping]) \
        .default_case([])

    workflow = ConductorWorkflow(
        name="order_processing_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(fetch_order_details)
    workflow.add(switch_shipping)

    workflow.register(True)

    return workflow

Check out the full sample code for the Switch Workflow.

Dynamic

The Dynamic task in Orkes Conductor enables workflows to determine and execute tasks at runtime based on input data, rather than defining them statically in the workflow definition. This is especially useful for cases where the number or type of tasks isn’t known in advance—such as processing a variable-length list of items or executing a dynamic sequence of operations. The task determines a llist of task(s) to be executed, allowing workflows to adapt their structure on the fly while remaining fully managed and observable within Conductor.

Example implementation

In a shipping workflow, the courier used to fulfill an order—such as FedEx or UPS—depends on the destination address, which isn’t known until runtime. A Dynamic task allows the workflow to decide which shipping task to execute on the fly, based on the result of a previous task that determines the best courier. For instance, if the shipping_info task outputs "ship_via_fedex", the dynamic task will execute the ship_via_fedex task; if it returns "ship_via_ups", the workflow dynamically routes to ship_via_ups. This eliminates hardcoded branching and makes the workflow adaptable to real-time decision-making.

Here’s the workflow visualized:

Workflow using a Dynamic operator.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Task to simulate courier selection
    shipping_info = HttpTask(
        task_ref_name="shipping_info_ref",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/posts",
            "method": "POST",
            "headers": {
                "Content-Type": "application/json"
            },
            "body": {
                "order_id": "${workflow.input.order_id}",
                "destination": "${workflow.input.address}"
            }
        }
    )

    # 2) Dynamic task that picks the shipping method based on workflow input
    dynamic_shipping = DynamicTask(
        dynamic_task="${workflow.input.shipping_service}",
        task_reference_name="dynamic_shipping"
    )

    workflow = ConductorWorkflow(
        name="Shipping_Flow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(shipping_info)
    workflow.add(dynamic_shipping)
    workflow.register(overwrite=True)

    return workflow


# Worker for FedEx
@worker_task(task_definition_name="ship_via_fedex")
def ship_via_fedex() -> dict:
    print("📦 Shipping via FedEx")
    return {"status": "FedEx shipment created"}


# Worker for UPS
@worker_task(task_definition_name="ship_via_ups")
def ship_via_ups() -> dict:
    print("📦 Shipping via UPS")
    return {"status": "UPS shipment created"}

Check out the full sample code for the Dynamic Workflow.

Set Variable

The Set Variable task in Orkes Conductor is used to define or update variables within the scope of a workflow execution. It allows you to store values—such as computed results, flags, or counters—that can be referenced later by other tasks. This task is especially useful for tracking workflow state, managing intermediate data, or preparing inputs for conditional logic or dynamic tasks. Variables set using this task are stored in the workflow’s context and persist throughout its execution.

Example implementation

In an e-commerce order processing workflow, the Set Variable task is used to store intermediate values like the subtotal, discount, and tax during price calculation. For instance, after the calculate_subtotal task, a Set Variable task updates the subtotal variable, which is then modified by the apply_discount task and used in the final price calculation. These variables are passed to subsequent tasks like process_payment, ensuring the correct amount is charged.

Here’s the workflow visualized:

Workflow using a Set Variable operator.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Calculate subtotal
    calc_subtotal = SimpleTask(
        task_def_name="calculate_subtotal",
        task_reference_name="calculate_subtotal"
    )

    # 2) Store subtotal
    set_subtotal = SetVariableTask(task_ref_name="set_subtotal")
    set_subtotal.input_parameters.update({
        "subtotal": "${calculate_subtotal.output.subtotal}"
    })

    # 3) Apply discount
    apply_discount = SimpleTask(
        task_def_name="apply_discount",
        task_reference_name="apply_discount"
    )

    # 4) Store discounted price
    set_discounted_price = SetVariableTask(task_ref_name="set_discounted_price")
    set_discounted_price.input_parameters.update({
        "discounted_price": "${apply_discount.output.discounted_total}"
    })

    # 5) Process payment
    process_payment = SimpleTask(
        task_def_name="process_payment",
        task_reference_name="process_payment",
        input_parameters={
            "amount": "${workflow.variables.discounted_price}"
        }
    )

    workflow = ConductorWorkflow(
        name="ecommerce_order_processing",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(calc_subtotal)
    workflow.add(set_subtotal)
    workflow.add(apply_discount)
    workflow.add(set_discounted_price)
    workflow.add(process_payment)
    workflow.register(overwrite=True)

    return workflow


@worker_task(task_definition_name="calculate_subtotal")
def calculate_subtotal() -> dict:
    print("🧮 Calculating subtotal...")
    return {"subtotal": 200.0}


@worker_task(task_definition_name="apply_discount")
def apply_discount() -> dict:
    print("🏷️ Applying discount...")
    return {"discounted_total": 180.0}


@worker_task(task_definition_name="process_payment")
def process_payment(amount) -> dict:
    print(f"💰 Charging customer: {amount.input_data['amount']}")
    return {"status": "Payment successful"}

Check out the full sample code for the Set Variable Workflow.

Terminate

The Terminate task in Orkes Conductor allows you to immediately end a workflow, either successfully or with a failure, before all tasks are completed. This is useful for scenarios where certain conditions are met, such as an error or a manual intervention, that require halting the workflow early. The task can be configured with an optional message, like a termination reason, providing context. This helps ensure that workflows end gracefully, preventing unnecessary tasks from being executed when they are no longer needed.

Example implementation

In a shipping workflow, the Terminate task ends the workflow if an invalid shipping provider is provided. During the workflow, a decision is made to ship with a specific provider based on user input. If the input does not match the available options (e.g., FedEx or UPS), the workflow enters the default case of a Switch task and triggers a Terminate task, halting the workflow with a "FAILED" status and a relevant reason.

Here’s the workflow visualized:

Workflow using a Terminate operator.

Here’s the code snippet for creating the workflow in code:

def register_shipping_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Decide based on shipping provider
    switch_provider = SwitchTask(
        task_ref_name="switch_provider",
        case_expression="${workflow.input.shipping_provider}"
    )

    # 2a) FedEx path
    fedex_task = SimpleTask(
        task_def_name="ship_with_fedex",
        task_reference_name="ship_with_fedex"
    )

    # 2b) UPS path
    ups_task = SimpleTask(
        task_def_name="ship_with_ups",
        task_reference_name="ship_with_ups"
    )

    # 3) Default path — Terminate the workflow if the provider is invalid
    terminate_task = TerminateTask(
        task_ref_name="terminate_invalid_provider",
        status=WorkflowStatus.FAILED,
        termination_reason="${workflow.input.termination_reason}"
    )

    # 4) Configure Switch task cases
    switch_provider.switch_case("FEDEX", [fedex_task])
    switch_provider.switch_case("UPS", [ups_task])
    switch_provider.default_case([terminate_task])

    workflow = ConductorWorkflow(
        name="shipping_workflow_with_validation",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(switch_provider)
    workflow.register(overwrite=True)

    return workflow


@worker_task(task_definition_name="ship_with_fedex")
def ship_with_fedex() -> dict:
    print("📦 Shipping with FedEx...")
    return {"status": "Shipped via FedEx"}


@worker_task(task_definition_name="ship_with_ups")
def ship_with_ups() -> dict:
    print("📦 Shipping with UPS...")
    return {"status": "Shipped via UPS"}

Check out the full sample code for the Terminate Workflow.

Wrap up

Branching and conditionals bring workflows to life, allowing them to adapt in real time based on input, state, or external data. With tasks like Switch, Dynamic, SetVariable, and Terminate, Conductor makes decision logic clear, declarative, and maintainable. No tangled logic, no hidden conditionals—just clean, visible control flow.

Next up:

—

Control the Flow: Building Dynamic Workflows with Orkes Operators

Orkes Developers — Mon, 28 Apr 2025 12:09:24 +0000

Orchestration is a powerful design framework for coordinating distributed services and systems into automated processes. Using a workflow engine, such as Orkes Conductor, provides a myriad of benefits like durable executions, state tracking, and developer productivity.

A key part of this orchestration model is operators, which are control flow primitives that resemble programming constructs. By using operators, developers can model dynamic, intelligent workflows without writing imperative control logic, making it easier to scale, maintain, and evolve complex systems.

What are operators?

In Conductor, operators define the control flow, structure, and decision-making within a workflow. Operators are one of the three core task types used in workflows, alongside system tasks and worker tasks. While system and worker tasks focus on performing actions (like sending HTTP requests or processing data), operator tasks define the control flow.

Operators act as declarative control structures, similar to constructs in traditional programming languages. For example, a Switch operator mimics a switch-case statement, a Fork allows for parallel task execution, and a Do While enables looping logic. These constructs make it possible to express complex workflow logic (branching, looping, and conditional execution) all without writing procedural code.

Operators are essential for orchestrating agentic workflows, where conditional or parallel logic provides a mapped space for the agent to act autonomously within bounds. Similarly, using operators, human-in-the-loop flows can be dynamically controlled based on the manual input received mid-flow.

Like system tasks, operator tasks are fully managed by Conductor and do not require any external workers to run. They are evaluated and executed internally by the engine, making them lightweight and efficient building blocks for composing complex workflows.

Why use operators?

Using operators in your workflows offers a range of benefits, especially when compared to managing control flow directly in code:

Declarative logic: Operators allow you to express branching, looping, and parallelism in a clean, declarative way. No need to write imperative control logic or orchestration code.
Separation of concerns: Business logic stays inside your services, while orchestration logic lives inside your workflow. This separation makes it easier to test, reason about, and maintain your applications.
Dynamic execution paths: You can define workflows that respond to real-time data and conditions, dynamically choosing tasks or branches to execute.
Fully managed by Conductor: Operators require no external workers or custom handlers. They’re executed entirely within the Conductor engine, reducing complexity and operational overhead.

By leveraging operators, you can design robust, readable, and adaptable workflows without getting bogged down by control-flow code scattered across services.

The power of declarative logic

Declarative logic is all about describing what should happen, not how to make it happen. In traditional code, developers must spell out the exact sequence of steps and control flow — managing state, writing loops, handling conditionals, and coordinating service calls. This imperative approach can quickly become brittle and hard to maintain as workflows grow in complexity.

By contrast, declarative logic lets you define the intent of the workflow and offload the execution details to the orchestration engine. This not only simplifies development but also makes workflows easier to read, reason about, and debug.

Operators in Orkes Conductor are a perfect example of this approach. They serve as control flow declarations: you define that a workflow should loop, branch, or fork — and Conductor takes care of executing that logic under the hood. Want to repeat a task until a condition is met? Just add a Do While with the right condition. Need to split work across multiple paths and rejoin later? Use Fork and Join. There’s no need to write a single line of plumbing code.

This declarative style brings huge benefits:

Maintainability: Simply update the workflow definition to change the workflow logic. No need to rewrite service code.
Readability: Declarative workflows are self-documenting. You can inspect them in the UI or JSON and instantly understand the flow.
Modularity: You can compose and reuse smaller workflows, using operators like Sub Workflow or Start Workflow as high-level building blocks.

Declarative vs imperative logic

In this example, we demonstrate how both imperative and declarative workflows handle iterating over a list of tasks and making HTTP requests for each. The imperative workflow explicitly defines the loop, including custom error handling for each HTTP request, while the declarative workflow uses Orkes Conductor's built-in Do While operator to manage iteration, with automatic error handling provided by the engine. This showcases the difference in complexity and control between the two approaches.

Imperative logic vs declarative logic.

Declarative orchestration is especially valuable when building agentic workflows, human-in-the-loop workflows, event-driven systems, or other dynamic execution paths. Operators let you adapt to changing inputs or user actions without embedding decision logic inside services.

Overview of Conductor’s operators

Build workflows declaratively with operators.

With over a dozen operators and extensive SDK support, you can quickly build out and deploy powerful workflows. Let’s explore example workflows that make use of these most commonly used tasks:

Operator	Description
Switch	Branches the workflow based on conditions, similar to a switch-case construct. Learn more: Branching and Conditionals
Dynamic	Dynamically evaluates conditions at runtime to determine what to do. Learn more: Branching and Conditionals
Set Variable	Sets variables within the workflow, allowing for dynamic control during execution. Learn more: Branching and Conditionals
Terminate	Ends the workflow execution early, often used for terminating workflows under conditions. Learn more: Branching and Conditionals
Do While	Loops through tasks until a condition is no longer true, used for repeated actions. Learn more: Loops, Waits, and Human-in-the-Loop
Wait	Pauses execution for a set time or until a condition is met. Learn more: Loops, Waits, and Human-in-the-Loop
Human	Pauses execution until manual input is received, useful for human-in-the-loop scenarios. Learn more: Loops, Waits, and Human-in-the-Loop
Fork	Executes multiple tasks in parallel, enabling concurrent processing. Learn more: Parallelism and Reusability
Join	Merges the results from parallel tasks, allowing the workflow to proceed together. Learn more: Parallelism and Reusability
Dynamic Fork	Dynamically creates parallel tasks at runtime based on conditions or data. Learn more: Parallelism and Reusability
Sub Workflow	Triggers another workflow within the current workflow for modularity and reusability. Learn more: Parallelism and Reusability
Start Workflow	Starts another workflow asynchronously in parallel with the current workflow. Learn more: Parallelism and Reusability

Wrap up

Orkes Conductor simplifies workflow orchestration by using declarative operators to manage complex logic. With tasks like Switch, Do While, and Fork/Join, you can build flexible, scalable workflows without writing imperative code. This approach accelerates development, reduces maintenance overhead, and enables collaboration across teams. Whether through the visual diagram editor or Conductor SDKs in Java, Python, or other languages, Conductor empowers developers to easily design, manage, and scale workflows.

Mastering control flow with Conductor

With operators, you design the flow, and Conductor handles the execution logic, freeing you to focus on building the features that matter.

Check out how to use Conductor’s operators with these articles:

Automating Insurance Claims Processing with AI and Conductor

livw — Fri, 25 Apr 2025 10:12:22 +0000

One of the biggest bottlenecks for the insurance industry is the sheer number of claims to process. The manual approach is expensive and time-consuming, requiring human operations to verify the documents, cross-check against the insurance policy, and follow up with the claimant, before the claim can be approved.

These process inefficiencies translate to a poor claims experience for customers, leading to dissatisfaction and ultimately driving customers to switch insurers. What if there were a way to effectively automate all that work so that claims processing can be completed in mere days or hours, reducing customer churn?

In this guide, we will explore how to wield LLMs (large language models) and orchestration to streamline manual business processes like insurance claims. Read on to learn the benefits, technical implementation details, and a demo example that you can try out in Orkes Conductor.

The solution: AI meets orchestration

The insurance industry is not new to automation. Rules-based automation approaches like RPA involved tons of pre-planning and data preparation, which meant that once the process was set in place, new changes could not be easily made. These early automation approaches were more rigid and limited, and could not easily scale to handle complex cases or exceptions.

Automating complexity

By leveraging the natural language capabilities of LLMs, the claims processing pipeline can be more effortlessly automated without having to predefine hundreds of rules across different policy wordings. Using prompt engineering, an LLM can act as an insurance representative to evaluate whether a claim can be approved. Here is an LLM-generated evaluation, which correctly identifies potential issues without any explicit business rules set:

The claim description mentions leukemia, which is a type of cancer. The 
doctor's report indicates tumors detected in blood, which is consistent 
with leukemia. However, the stage of cancer is not explicitly mentioned, 
and the policy covers only stage 3 and 4 cancer. The required documents 
(medical bill and doctor's report) are present, but the stage information 
is missing, which creates some uncertainty. The probability [of approval] 
is relatively high but not certain due to the lack of specific staging 
information.

Automating distributed systems

Reviewing the claim details is just one step in the insurance processing pipeline. This is where orchestration comes into play. Orkes Conductor is a fully-managed orchestration platform that coordinates disparate components and systems into an automated workflow or business process. Even if the supporting documents are uploaded on a system separate from customer details, or if the payment processing service resides on a different platform from the claims portal, an orchestration engine transforms these friction points into a well-oiled flow.

These are some key highlights of Orkes Conductor:

Integration—Connect different services and data sources, including legacy systems, through a cloud-based middleware platform.
Visibility—Track the status of your processes, easily recover from failures, and gain in-depth metrics on performance.
AI capabilities—Use in-built features to natively add LLM-powered components to your processes.

Now, let’s walk through an example insurance claims flow built in Orkes Conductor.

Insurance claims processing flow

Orkes Conductor can automate any claims processing flow with stateful process orchestration and seamless AI integrations.

The claims processing flow can be distilled into three main steps:

Data extraction from multiple sources—capture the claim details, supporting documents, and relevant policy documents.
Claims assessment—review the data to determine if the claim meets the criteria for approval, using AI or human reviewers.
Outcome processing—execute the approval decision, like kicking off a payments flow and notifying the claimant.

As the orchestrator unit, Orkes Conductor acts as the execution engine that drives all the core components of the insurance claims process.

Conductor is the central orchestrator that coordinates between distributed services and systems.

1. Data extraction from multiple sources

With a suite of over 20 system tasks, Orkes Conductor supports retrieving data from many sources: through API endpoints, from data lakes, SQL databases, vector databases in a RAG system, and more. This enables you to automatically plug disparate data sources into business processes without any manual imports.

2. AI-powered claim assessment

The claim description mentions leukemia, which is a type of cancer. The 
doctor's report indicates tumors detected in blood, which is consistent 
with leukemia. However, the stage of cancer is not explicitly mentioned, 
and the policy covers only stage 3 and 4 cancer. The required documents 
(medical bill and doctor's report) are present, but the stage information 
is missing, which creates some uncertainty. The probability [of approval] 
is relatively high but not certain due to the lack of specific staging 
information.

As previously demonstrated, LLMs possess the capability to naturally parse text without requiring deep technical work. Leveraging prompt engineering, the LLM can be easily tuned to output accurate claim assessments. Even if the insurance policy changes, an LLM-powered solution will intelligently reason through any claims scenario. Here is a simplified prompt template that produced the claims assessment above:

You are an insurance claims assessor who needs to calculate the probability of approving a claim. 

This is the policy statement: "${policy-statement}".

These are the supporting docs: "${docs}".

This is the claims description: "${claims-description}".

Orkes Conductor makes it easy to scale LLM-driven processes with its AI Prompt Studio and code-free AI/LLM integrations and tasks. With the Orkes’ AI Prompt Studio, you can securely craft and test reusable prompts for the claims assessment workflow. By replacing the prompt variables like ${policy-statement} with the required information extracted in step 1, the LLM can predictably handle complex or edge cases—combining the strengths of AI-powered flexibility with guardrails.

Orkes’ suite of LLM integrations provides access to the latest AI models from all major LLM providers, like OpenAI, Amazon Bedrock, and Anthropic. With all these choices at your fingertips, Orkes’ modular workflow approach makes it easy to test and switch between any AI model to find the most effective one for your use case — in this case, an insurance claims flow.

Bring all of these together by using Orkes’ built-in LLM tasks, which provides chat completions, embedding retrievals, and more without needing to add extra code.

Orkes' suite of AI features empowers rapid development.

Human-in-the-loop

Even with guardrails for AI models, adding human review checkpoints further mitigates the risks of AI miscalculation. After the AI-driven claims assessment is completed, the process can then trigger a human review for final approval of the insurance payout. The AI-generated assessment provides a detailed summary for the human assessor to investigate, speeding up the process.

In Orkes, our human-in-the-loop features allow you to seamlessly integrate these human touchpoints with both your backend process and your frontend claims review portal. This ensures an audit trail of all actions taken, while guaranteeing that businesses don’t have to migrate to different portal just to enjoy these benefits.

3. Business decision post-actions

Finally, once the decision has been made, the workflow can route either to an early termination in the case of rejection, or to a post-processing sequence like a payout initiation and outcome notification. Orkes’ orchestration engine facilitates coordination across different microservices, internal services, and other third-party integrations like Stripe and Sendgrid.

Integration with your tech stack

Integrate the workflow into your existing portal for claims processing. Run it as a server job. Trigger the claims process from an event-driven system. Whatever your system architecture requires, Orkes Conductor brings the flexibility to fit with your existing processes rather than force you to migrate.

Try it out in Conductor

Now, here’s how you can try out a working AI-powered insurance claims processing workflow for yourself in Conductor.

Orkes’ visual workflow editor makes it intuitive to transition from a napkin-top or whiteboard-sketch idea to a production-ready workflow.

Prerequisites:

Create an account in our free Developer Playground.

Step 1: Create the insurance claim workflow

Log in to Developer Playground.
In Definitions > Workflow, select + Define workflow.

Paste the following workflow JSON into the Code tab on the right-side panel:

{
  "createTime": 1744263984129,
  "updateTime": 1744871696379,
  "name": "insurance-claim",
  "description": "Insurance claim workflow",
  "version": 1,
  "tasks": [
    {
      "name": "collate-details",
      "taskReferenceName": "collate-details_ref",
      "inputParameters": {},
      "type": "FORK_JOIN",
      "decisionCases": {},
      "defaultCase": [],
      "forkTasks": [
        [
          {
            "name": "get-customer-details",
            "taskReferenceName": "get-customer-details_ref",
            "inputParameters": {
              "uri": "https://orkes-api-tester.orkesconductor.com/api",
              "method": "GET",
              "connectionTimeOut": 3000,
              "readTimeOut": "3000",
              "accept": "application/json",
              "contentType": "application/json"
            },
            "type": "HTTP"
          },
          {
            "name": "get-claims-details",
            "taskReferenceName": "get-claims-details_ref",
            "inputParameters": {
              "expression": "// Function to randomly pick an item from the array\n(function getRandomItem(arr) {\n  const items = [\n    { id: 1, name: \"Jane\", claims: \"stage 3 stomach cancer\" },\n    { id: 2, name: \"Jen\", claims: \"stage 0 breast cancer\" },\n    { id: 3, name: \"John\", claims: \"stage 4 leukemia\" },\n    { id: 4, name: \"Jim\", claims: \"hyperthyroidism \" },\n    { id: 4, name: \"Jeff\", claims: \"leukemia \" },\n    { id: 4, name: \"June\", claims: \"stage 4 flu \" }\n  ];\n\n  const randomIndex = Math.floor(Math.random() * items.length);\n  return items[randomIndex];\n}\n)();",
              "evaluatorType": "graaljs"
            },
            "type": "INLINE"
          }
        ],
        [
          {
            "name": "get-policy-details",
            "taskReferenceName": "get-policy-details_ref",
            "inputParameters": {
              "uri": "https://orkes-api-tester.orkesconductor.com/api",
              "method": "GET",
              "connectionTimeOut": 3000,
              "readTimeOut": "3000",
              "accept": "application/json",
              "contentType": "application/json"
            },
            "type": "HTTP"
          },
          {
            "name": "get-coverage-statement",
            "taskReferenceName": "get-coverage-statement_ref",
            "inputParameters": {
              "coverageStatement": "Critical illness cover of USD 200,000.00 for stage 3 and 4 cancer. To make a successful claim, the following documents are required: (1) medical bill and (2) doctor's report. If any documents are missing, the claim cannot be processed."
            },
            "type": "SET_VARIABLE"
          }
        ],
        [
          {
            "name": "get-supporting-docs",
            "taskReferenceName": "get-supporting-docs_ref",
            "inputParameters": {
              "expression": "// Function to randomly pick an item from the array\n(function getRandomItem(arr) {\n  const items = [\n    { receipt: \"x-ray scan - USD 4000 consult USD 7000 total USD 11000\", report: \"Doctor Lee 15/03/25 Tumours detected in blood. Suspected stage 4 cancer. Patient is referred for further treatment.\" },\n    { receipt: \"x-ray scan - USD 4000 consult USD 7000 total USD 11000\", report: \"Doctor Lee 15/03/25 Tumours detected in blood. Suspected stage 1 cancer. Patient is referred for further treatment.\" },\n    { receipt: \"x-ray scan - USD 4000 consult USD 7000 total USD 11000\", report: \"Doctor Lee 15/03/25 Tumours detected in blood. Patient is referred for follow-up.\" },\n    { receipt: \"\", report: \"\" }\n  ];\n\n  const randomIndex = Math.floor(Math.random() * items.length);\n  return items[randomIndex];\n}\n)();",
              "evaluatorType": "graaljs"
            },
            "type": "INLINE"
          }
        ]
      ],
      "startDelay": 0,
      "joinOn": []
    },
    {
      "name": "join_on_collate",
      "taskReferenceName": "join_on_collate_ref",
      "inputParameters": {},
      "type": "JOIN",
      "forkTasks": [],
      "joinOn": [
        "get-customer-details_ref",
        "get-policy-details_ref"
      ],
      "optional": false
    },
    {
      "name": "get-approval-probability",
      "taskReferenceName": "get-approval-probability_ref",
      "inputParameters": {
        "llmProvider": "AnthropicClaude",
        "model": "claude-3-5-sonnet-20240620",
        "promptName": "determine-insurance-claim-probability",
        "promptVariables": {
          "claims-description": "${get-claims-details_ref.output.result.claims}",
          "policy-statement": "${workflow.variables.coverageStatement}",
          "docs": "${get-supporting-docs_ref.output.result}"
        }
      },
      "type": "LLM_TEXT_COMPLETE"
    },
    {
      "name": "ai-decider",
      "taskReferenceName": "ai-decider_ref",
      "inputParameters": {
        "probabilityValue": "${get-approval-probability_ref.output.result.probability}"
      },
      "type": "SWITCH",
      "decisionCases": {
        "0.0": [
          {
            "name": "send-refile-claim-notification",
            "taskReferenceName": "send-refile-claim-notification_ref",
            "inputParameters": {
              "uri": "https://orkes-api-tester.orkesconductor.com/api",
              "method": "GET",
              "accept": "application/json",
              "contentType": "application/json",
              "encode": true
            },
            "type": "HTTP"
          },
          {
            "name": "terminate_2",
            "taskReferenceName": "terminate_ref_2",
            "inputParameters": {
              "terminationStatus": "COMPLETED",
              "terminationReason": "Claim did not pass LLM validation check. Need to refile for claims processing."
            },
            "type": "TERMINATE"
          }
        ]
      },
      "defaultCase": [
        {
          "name": "approve-claim-payout",
          "taskReferenceName": "approve-claim-payout_ref",
          "inputParameters": {
            "__humanTaskDefinition": {
              "assignmentCompletionStrategy": "LEAVE_OPEN",
              "displayName": "Approve Claim",
              "userFormTemplate": {
                "name": "InsuranceClaims",
                "version": 1
              },
              "assignments": [
                {
                  "assignee": {
                    "user": "ACME",
                    "userType": "EXTERNAL_GROUP"
                  },
                  "slaMinutes": 0
                }
              ],
              "taskTriggers": []
            },
            "claimDescription": "${get-claims-details_ref.output.result.claims}",
            "coverageStatement": "${workflow.variables.coverageStatement}",
            "probability": "${get-approval-probability_ref.output.result.probability}",
            "aiSummary": "${get-approval-probability_ref.output.result.reason}",
            "approval": false
          },
          "type": "HUMAN"
        },
        {
          "name": "approved",
          "taskReferenceName": "approved_ref",
          "inputParameters": {
            "approval": "${approve-claim-payout_ref.output.approval}"
          },
          "type": "SWITCH",
          "decisionCases": {
            "true": [
              {
                "name": "process-payment",
                "taskReferenceName": "process-payment_ref",
                "inputParameters": {
                  "uri": "https://orkes-api-tester.orkesconductor.com/api",
                  "method": "GET",
                  "connectionTimeOut": 3000,
                  "readTimeOut": "3000",
                  "accept": "application/json",
                  "contentType": "application/json"
                },
                "type": "HTTP"
              },
              {
                "name": "send-success-notification",
                "taskReferenceName": "send-success-notification_ref",
                "inputParameters": {
                  "uri": "https://orkes-api-tester.orkesconductor.com/api",
                  "method": "GET",
                  "connectionTimeOut": 3000,
                  "readTimeOut": "3000",
                  "accept": "application/json",
                  "contentType": "application/json"
                },
                "type": "HTTP"
              }
            ]
          },
          "defaultCase": [
            {
              "name": "send-not-approved-notification",
              "taskReferenceName": "send-not-approved-notification_ref",
              "inputParameters": {
                "uri": "https://orkes-api-tester.orkesconductor.com/api",
                "method": "GET",
                "connectionTimeOut": 3000,
                "readTimeOut": "3000",
                "accept": "application/json",
                "contentType": "application/json"
              },
              "type": "HTTP"
            },
            {
              "name": "terminate_1",
              "taskReferenceName": "terminate_ref_1",
              "inputParameters": {
                "terminationStatus": "COMPLETED",
                "terminationReason": "Claim has not been approved by human agent."
              },
              "type": "TERMINATE"
            }
          ],
          "evaluatorType": "value-param",
          "expression": "approval"
        }
      ],
      "evaluatorType": "value-param",
      "expression": "probabilityValue"
    }
  ],
  "inputParameters": [
    "policyNumber",
    "customerId",
    "supportingDocs"
  ],
  "outputParameters": {},
  "failureWorkflow": "",
  "schemaVersion": 2,
  "restartable": true,
  "timeoutPolicy": "ALERT_ONLY",
  "timeoutSeconds": 0,
  "variables": {},
  "inputTemplate": {},
  "enforceSchema": true
}

Save the workflow.

Step 2: Add LLM integrations and models

Go to Integrations and select + New integration.
In the AI/LLM section, add your desired LLM provider.

Note: In the workflow that you’ve just created, the get-approval-probability task makes use of **Anthropic Claude’s Sonnet 3.5, but you can choose any LLM provider and simply modify the workflow task later.
Return to Integrations and add your LLM models to your newly-added integration.

For detailed steps on adding LLM integrations, refer to the documentation.

Step 3: Add the AI prompt

Go to Definitions > AI Prompts and select + Add AI prompt.

In the Code tab, paste the following JSON:

{
  "createTime": 1744715996941,
  "updateTime": 1744906103023,
  "name": "determine-insurance-claim-probability",
  "template": "You are an insurance claims assessor who needs to calculate the probability of approving a claim.\n\nThis is the policy statement: \"${policy-statement}\".\n\nThese are the supporting docs: \"${docs}\".\n\nThis is the claims description: \"${claims-description}\".\n\nFormat your response ONLY as a JSON object with the following structure:\n<valid>\n{\n  \"probability\": \"A numerical probability between 0.0 - 1.0. \",\n  \"reason\": \"A reason for the probability.\"\n}\n</valid>\n\nDo not wrap the JSON object in markdown. Do use quotation marks in the JSON object.\n<invalid>\n{probability=0.0, reason=The claim description mentions 'stage 5 cancer', which does not match the policy coverage for stage 3 and 4 cancer. Additionally, the provided documents confirm stage 3 cancer, further disqualifying the claim from approval.}\n</invalid>\n\n<invalid>\n```

json\n{\n  \"probability\": 0.0,\n  \"reason\": \"The claim description mentions 'stage 5 cancer', which does not match the policy coverage for stage 3 and 4 cancer. Additionally, the provided documents confirm stage 3 cancer, further disqualifying the claim from approval.\"\n}\n

```\n</invalid>",
  "description": "Determine the probability of approving a claim based on a policy statement and a claim description",
  "variables": [
    "docs",
    "claims-description",
    "policy-statement"
  ],
  "integrations": [],
  "tags": []
}

Return to the Form tab, and add any AI models you wish to use with this prompt.

Note: In the workflow that you’ve just created, the get-approval-probability task makes use of **Anthropic Claude’s Sonnet 3.5, but you can add any LLM provider here and simply modify the workflow task later.
Save the AI prompt.

Optional: Modify the AI prompt

The prompt should look something like this:

You are an insurance claims assessor who needs to calculate the probability of approving a claim.

This is the policy statement: "${policy-statement}".

These are the supporting docs: "${docs}".

This is the claims description: "${claims-description}".

Format your response ONLY as a JSON object with the following structure:
<valid>
{
  "probability": "A numerical probability between 0.0 - 1.0. ",
  "reason": "A reason for the probability."
}
</valid>

Do not wrap the JSON object in markdown. Do use quotation marks in the JSON object.
<invalid>
{probability=0.0, reason=The claim description mentions 'stage 5 cancer', which does not match the policy coverage for stage 3 and 4 cancer. Additionally, the provided documents confirm stage 3 cancer, further disqualifying the claim from approval.}
</invalid>

<invalid>
\`\`\`json
{
  "probability": 0.0,
  "reason": "The claim description mentions 'stage 5 cancer', which does not match the policy coverage for stage 3 and 4 cancer. Additionally, the provided documents confirm stage 3 cancer, further disqualifying the claim from approval."
}
\`\`\`
</invalid>

To modify the prompt, here are some guidelines to steer the LLM to respond with a well-structured JSON output:

Unambiguously establish the role of the LLM as an insurance claims assessor.
Precisely describe the expected input and output formats.
- Make sure the output has the exact fields expected by the subsequent tasks in the workflow.
- Provide clear examples of the output.
Emphasize fairness and adherence to the provided policies.
Make sure that missing inputs are properly handled.

Optional: Modify the workflow

If you wish to use other LLM providers than Anthropic Claude, make sure to modify the get-approval-probability task.

Return to your insurance claim workflow and select the get-approval-probability task in the visual diagram.
In the Task tab, modify the LLM provider and Model fields.
Save the workflow.

Step 4: Add the user form for the Human task

The user form serves as the interface for human assessors to evaluate and approve the payout. Orkes Conductor makes it easy to integrate and deploy these user forms on your own user portals through our extensive suite of APIs and SDKs.

Go to Definitions > User Forms and select + New form.

In the Code tab, paste the following JSON:

{
  "createTime": 1744816361848,
  "updateTime": 1744816643214,
  "name": "InsuranceClaims",
  "version": 1,
  "jsonSchema": {
    "$schema": "http://json-schema.org/draft-07/schema",
    "properties": {
      "claimDescription": {
        "type": "string"
      },
      "coverageStatement": {
        "type": "string"
      },
      "probability": {
        "type": "string"
      },
      "aiSummary": {
        "type": "string"
      },
      "approval": {
        "type": "boolean"
      }
    }
  },
  "templateUI": {
    "type": "VerticalLayout",
    "elements": [
      {
        "type": "Control",
        "scope": "#/properties/claimDescription",
        "label": "Claim Description",
        "options": {
          "readonly": true
        }
      },
      {
        "type": "Control",
        "scope": "#/properties/coverageStatement",
        "label": "Coverage Statement",
        "options": {
          "readonly": true
        }
      },
      {
        "type": "Control",
        "scope": "#/properties/probability",
        "label": "Probability",
        "options": {
          "readonly": true
        }
      },
      {
        "type": "Control",
        "scope": "#/properties/aiSummary",
        "label": "AI Summary",
        "options": {
          "readonly": true
        }
      },
      {
        "type": "Control",
        "scope": "#/properties/approval",
        "label": "Approve?",
        "options": {
          "default": false
        }
      }
    ]
  }
}

Save the user form.

Running the workflow

Now that all your Conductor resources are ready, let’s run the workflow to experience the power of AI in automating claims processing.

To run the workflow:

In Definitions > Workflow, select your insurance claim workflow.
Select the Run tab in the right-side panel and select Run Workflow.

Upon running the workflow, you will be redirected to the workflow execution details page. Use the visual diagram to follow the workflow progression.

In Conductor, easily track workflows as they progress through different stages.

Intelligent automation in action

In the workflow execution details page, select the get-approval-probability task and then its Output tab. The LLM will provide an approval probability along with its reasoning.

In this instance, the probability of approval is 0, since the required supporting documents are missing. With workflow orchestration, the process then automatically progresses to notify the claimant to supplement the missing documents without needing additional human review.

Let’s trigger another execution to examine how a different scenario might play out.

In this instance, the probability is non-zero since all the supporting documents are present. However, the supporting evidence is a little vague and may require further follow-up. This is where the workflow benefits greatly from human oversight. Since there is a chance for approval, the workflow routes to a human-in-the-loop task, where a human assessor can review the AI-generated assessment summary to make the final decision.

Through Conductor's Human Task APIs, these user forms can be integrated into existing portals.

Try it out yourself by going to Executions > Human Tasks and selecting the pending execution. You can claim the task, check the approve button to approve it (or leave it unchecked to reject it), and complete the task.

When you return to the workflow executions details page (in Executions > Workflow), the workflow will have progressed to the next steps depending on your choice. This is the power of Orkes Conductor, combining both orchestration and AI capabilities to build automated processes that scale.

Extending the workflow

While the data extraction, payment processing, and notification steps are mock tasks, they can be easily replaced with real data sources and services. Go a step further by switching out the mock tasks with real tasks for your own needs, or request a demo for Orkes Conductor to learn more about how orchestration can accelerate your business processes.

Summary

Slow, outdated insurance claims processes can be modernized by investing in the latest technology for AI and orchestration. Using Orkes Conductor to scale AI-automated processes offers a multitude of benefits:

AI automation and agility—Easily add and upgrade AI-driven components to your processes for intelligent automation.
Integration with existing systems—Use Orkes as the middleware platform to integrate legacy systems, third-party applications, and other components in your business ecosystem.
Enhanced governance and monitoring—Built-in monitoring and state tracking dashboards to get global visibility and auditability into processes and performance.
Human-in-the-loop oversight—Seamlessly add human oversight to processes and integrate them with frontend interfaces for your intended users.
Fast, scalable performance—Build and run processes on a performant orchestration engine that powers the biggest players across industries, like Netflix, Tesla, and American Express.

Conductor is an enterprise-grade Unified Application Platform for process automation, API and microservices orchestration, agentic workflows, and more. Check out the full set of features, or try it yourself using our free Developer Playground.

Building an Agentic Workflow: Orchestrating a Multi-Step Software Engineering Interview

Riza Farheen — Mon, 21 Apr 2025 10:46:26 +0000

As language models evolve, so does the way we build intelligent systems. One emerging pattern is the agentic workflow—a coordinated sequence of tasks in which an AI agent interacts with external tools and APIs, makes decisions, and executes complex flows autonomously.

But intelligence alone isn’t enough. Large language models excel at reasoning and response generation, yet they fall short when it comes to real-world execution. Managing state, coordinating asynchronous tasks, handling tool invocation, and ensuring reliability across multiple steps remain key challenges. That’s where orchestration comes in.

By combining LLMs with a robust workflow engine, we can design resilient, multi-step systems that bridge reasoning with action. In this guide, we’ll walk through an agentic workflow that automates a full software engineering technical interview—from candidate intake to interview questions, evaluation, and report delivery.

Example: Agentic interview app

Let’s build an application that mimics a human-led coding interview, where:

The candidate enters their details.
An LLM agent conducts the interview by generating a series of leetcode-style programming questions, which the candidate responds to in a sequence.
The agent scores answers and provides feedback.
A formatted transcript of the interview is generated.
A final report is emailed to the candidate.

This multi-step process involves reasoning (via OpenAI), external tools (Google Docs and SendGrid), and orchestration (Orkes Conductor).

To build it, we’ll use the following components:

Agentic Interview App flow

Frontend (Next.js)–Provides the user interface for candidate interaction. It captures user input (name, language, email) and streamlines the interview conversation.
Backend (Python)–Acts as the trigger point for the workflow. It starts the Conductor workflow and serves as middleware to process frontend user input to the backend.
OpenAI–Provides the reasoning layer. Generates questions, interprets responses, and guides the flow through hints and feedback.
Google Cloud (Docs API)–Generates and formats the interview transcript as a sharable document.
SendGrid–Send the final report and feedback to the candidate via email.
Orkes Conductor–Orchestrates the entire workflow: managing state, coordinating tasks, handling API integrations and logic.

Now, let’s walk through how these tools come together to form the agentic interview application.

Building an agentic interview app

You can follow this tutorial using the free Developer Playground. Create an account to get started.

We’ll build an interview app powered by the following Conductor workflow:

Interview Agentic Workflow in Orkes Conductor

After gathering the candidate details, the interview will begin. Conductor’s Start Workflow task then starts the Core Interview Loop workflow and generates a coding question using OpenAI. It then enters a loop to collect and evaluate the candidate’s responses (with support for hints or simplifications), finally scoring the answer and storing a structured evaluation as a scorecard.

Core Interview Loop Workflow in Orkes Conductor

Now, here’s how you can create an agentic interview application on your own.

Step 1: Clone the repository

The source code for this example project is available in this GitHub repository. Clone it to your local machine:

git clone https://github.com/conductor-oss/awesome-conductor-apps/

Open the project and navigate to the interview app folder:

cd python/interview_agentic_app

Step 2: Get credentials and set up environment variables

This application interacts with multiple tools and APIs. You’ll need to configure a few credentials before running it locally.

Access key and secret from Conductor

Get the access key and secret from Developer Playground by creating a new application from Access Control > Applications. Set the Application role to Worker and Metadata API.

Generate the access key and secret.

Generating access key and secret from Orkes Conductor

Open your project in the IDE of your choice and set the following environment variables:

export CONDUCTOR_SERVER_URL=https://developer.orkescloud.com/api
export CONDUCTOR_AUTH_KEY=<YOUR_CONDUCTOR_AUTH_KEY>
export CONDUCTOR_AUTH_SECRET=<YOUR_CONDUCTOR_AUTH_SECRET>

OpenAI API key

Generate your API key from the OpenAI platform, then set the environment variable:

export OPENAI_API_KEY=<YOUR_OPENAI_KEY>

Google authentication credentials

Google authentication credentials are required to access Google APIs. Ensure Google Docs API is enabled on your GCP project.

Create a service account in your Google Cloud console and download the JSON key file. It will be named something like project-name.json.

Creating service account key from GCP console

Next, we need to stringify the contents of the JSON file using the following command:

python3 -c 'import json; print(json.dumps(json.load(open("<PATH-TO-JSON-FILE>/project-name.json"))))'

Copy the output (the stringified JSON), then set the environment variables:

export GOOGLE_SERVICE_ACCOUNT_JSON='<PASTE_THE_STRING_HERE>'
export ENV=prod

SendGrid API key

SendGrid automatically sends the final interview summary and feedback to the candidate via email.

Start by creating an account on SendGrid, then create and verify the sender email address. Next, generate an API key from SendGrid.

Generating API key from SendGrid

Now, set the environment variable for your API Key:

export SEND_GRID_API_KEY=<YOUR_SEND_GRID_API_KEY>

Then, set the environment variable for your verified sender email:

export SEND_GRID_EMAIL_ADDRESS=<YOUR_SEND_GRID_VERIFIED_EMAIL_ADDRESS>

Finally, update your workflow JSON to use this sender email. Navigate to python/interview_agentic_app/resources/interviewAgenticWorkflow.json file.

Go to line 878 and replace the from.email value:

//workflow JSON

 "from": {
                  "email": "<SENDGRID_VERIFIED_SENDER_EMAIL>" //Line 878
                },
                "subject": "Thank you ${workflow.variables.name} for interviewing with Orkes",

Go to line 996 and replace the from.email value:

//workflow JSON 
                "from": {
                  "email": "<SENDGRID_VERIFIED_SENDER_EMAIL>" //Line 996
                },
                "subject": "Thank you ${workflow.variables.name} for interviewing with Orkes",

Step 3: Start the backend server

This step launches the Python-based backend that listens for incoming requests and triggers the Conductor workflow.

Set the Python path:

export PYTHONPATH=/[PATH_TO_REPO]/awesome-conductor-apps/python/interview_agentic_app

Create and activate a virtual environment:

python3 -m venv venv
source venv/bin/activate

Install required dependencies:

pip3 install -r requirements.txt

Start the backend server:

cd workflow
python app.py

Step 4: Start the frontend server

This starts the Next.js frontend, where candidates can enter their details and take the interview in an interactive chat UI.

Open a new terminal and navigate to your project:

cd python/interview_agentic_app

Next, navigate to the frontend directory:

cd interview-chat

Install dependencies:

npm install --legacy-peer-deps

Start the frontend server:

npm run dev

The app should now be running locally.

Step 5: View the interview app in action

To start the interview, open it in your browser at http://localhost:3000, this launches the interview experience locally.

Agentic Interview App UI

In your Developer Playground, you’ll notice that the required OpenAI integrations, AI prompts, and workflow definitions are automatically created.

Resources created in Developer Playground

As the candidate proceeds, the Interview Agentic Workflow is triggered. Each interview question is handled by a separate workflow, Core Interview Loop. So, for a three-question interview, three workflows run in sequence using Conductor’s Start Workflow task.

Triggered workflows in Developer Playground

Once the interview is complete, a thank you email with personalized feedback and scores is instantly sent to the candidate via SendGrid API.

Thank you email triggered via SendGrid on interview completion

In parallel, a fully formatted Google Docs transcript is generated and stored in Google Drive. The interviewer and candidate are granted access, and a link is shared in another email.

Interview transcript generated and stored in Google Docs

That’s it—you’ve now built a fully automated coding interview for a Software Engineer, from candidate intake to evaluation and reporting.

With the core workflow in place, you can easily tailor it to your own needs: customize the prompts, add more questions, or plug it into tools of your choice.

This agentic interview app isn’t just for demos—it’s built for production. You can deploy the backend on platforms like Render and the frontend on Vercel, with your Conductor workflows running in the Orkes Cloud.

What next?

Now that you’ve built and tested your agentic interview workflow, here are a few ways you can take it further:

Add a progress tracker–Show candidates how they’re doing after each question to make the experience more engaging.
Plug into ATS or HRIS–Send the final report straight to your ATS or HR dashboard for easier review.
Extend beyond coding–Replace coding questions with behavioral prompts or design discussions to adapt the workflow for non-technical interviews.

Wrapping up

This tutorial walked you through building a complete agentic workflow—one that goes beyond single-shot prompts to orchestrate a real-world, multi-step interview process. Using OpenAI for reasoning, Google Docs for formatting, SendGrid for communication, a modern frontend and backend stack, and Orkes Conductor at the core, you now have a robust foundation for deploying agentic workflows in production.

Whether you’re building technical assessments, onboarding flows, or multi-agent decision systems, Conductor gives you the structure to scale—with full control over inputs, logic, and integrations.

Explore more AI-powered workflows built with Conductor

Try building and running your first agentic workflow using the free Developer Playground. Drop by our Slack community and show us what you’ve built.

Implementing Easy-to-Build Workflows with Conductor’s System Tasks

Orkes Developers — Mon, 14 Apr 2025 13:52:00 +0000

This is Part 2 of the series on System Tasks. Check out Part 1, which discusses the pros and cons of using system tasks and alternative solutions.

Whether you’re orchestrating a simple API flow or building complex AI-powered applications, system tasks make it easy to plug functionality into your workflows. In this post, we’ll explore the most common system tasks in Orkes Conductor and demonstrate how you can leverage them to accelerate development and unlock the full potential of Orkes Conductor.

Recap: what are system tasks?

Orkes Conductor is a workflow orchestration platform packed with developer-first features.

System tasks in Orkes Conductor represent reusable business logic components, like HTTP calls and database queries. These tasks run within the Conductor engine, allowing developers to quickly build complex and durable workflows without reinventing the wheel.

Overview of Conductor’s system tasks

With over twenty system tasks and extensive SDK support, you can quickly build out and deploy powerful workflows. Let’s explore example workflows that make use of these most commonly-used tasks:

Event
HTTP
HTTP Poll
Inline
JSON JQ Transform
Wait For Webhook
Alerting tasks
AI tasks

Commonly-used system tasks in Orkes Conductor.

Event

In distributed systems, async communication is key一and that’s where the Event task comes in. It allows your workflow to publish messages to an external event queue or topic, enabling decoupled communication between services. Whether you’re using Kafka, SQS, or another messaging system, Orkes Conductor integrates with many providers, making it easy to trigger downstream processes or notify external systems as part of your workflow.

Example implementation

Imagine a payment workflow that processes a transaction and needs to notify an inventory service once completed. The workflow first makes an HTTP call to a payment API to confirm the transaction. Once confirmed, instead of directly calling the inventory service, an Event task publishes a message to a Kafka topic. The inventory service listens to the Kafka topic, consumes the event, and asynchronously updates stock levels, decoupling the processes for scalability and flexibility.

Here’s the workflow visualized:

Workflow using an Event task.

To create this workflow in Conductor, you must first integrate the required message broker, and then add the relevant system tasks. Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Define the HTTP task for calling the dummy JSONPlaceholder API
    http_task = HttpTask(
        task_ref_name="fetch_order_details",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/posts/\${workflow.input.orderId}",
            "method": "GET",
            "headers": {
                "Content-Type": "application/json"
            }
        }
    )

    # 2) Define the Event task using EventTaskInterface
    event_task = EventTaskInterface(
        task_ref_name="notify_inventory",
        event_prefix="kafka",
        event_suffix="quickstart-events"
    )

    # Update input parameters for the event task using data from the HTTP task response
    event_task.input_parameters.update({
        "orderId": "\${fetch_order_details.output.response.body.id}",
        "title": "\${fetch_order_details.output.response.body.title}",
        "body": "\${fetch_order_details.output.response.body.body}"
    })

    # Define the workflow and add the tasks
    workflow = ConductorWorkflow(
        name="inventory_management_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(http_task)
    workflow.add(event_task)

    # Register the workflow
    workflow.register(True)

    return workflow

Check out the full sample code for the Event Workflow.

HTTP

The HTTP task is used to make calls to remote services exposed over HTTP/HTTPS, a fundamental part of most service-based applications. It provides a clean abstraction that makes it intuitive to declare API calls. This simplifies integration with external systems, eliminating the need for custom workers and manual integration..

Example implementation

In a payment processing system, you need to interact with an external API to charge a customer’s credit card. It sends a POST request via an HTTP task to the payment API with the order ID and amount, along with an authorization token.

If the payment succeeds (status code 200), the workflow continues to update the order status and marks the payment as successful.
If the payment fails (any status code other than 200), the workflow notifies the user of the failure.

The workflow ensures efficient handling of payment success or failure through conditional branching with the use of the Switch task.

Here’s the workflow visualized:

Workflow using an HTTP task.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Define the HTTP task for calling the payment API
    http_task = HttpTask(
        task_ref_name="call_payment_api",
        http_input={
            "uri": "https://httpbin.org/status/200",
            "method": "POST",
            "headers": {
                "Content-Type": "application/json",
                "Authorization": "Bearer \${workflow.input.auth_token}"
            },
            "body": {
                "orderId": "\${workflow.input.orderId}",
                "amount": "\${workflow.input.amount}"
            }
        }
    )

    # 2) Define the SwitchTask with case expression
    switch_task = SwitchTask(
        task_ref_name="check_payment_status",
        case_expression="\${call_payment_api.output.response.statusCode}",
        use_javascript=False
    )

    # 3) Define SetVariableTask for success case
    set_payment_success_variable = SetVariableTask(
        task_ref_name="set_payment_success_variable"
    )
    set_payment_success_variable.input_parameters.update({
        "payment_status": "SUCCESS"
    })

    # Define SetVariableTask for failure case
    set_payment_failure_variable = SetVariableTask(
        task_ref_name="set_payment_failure_variable"
    )
    set_payment_failure_variable.input_parameters.update({
        "payment_status": "FAILURE"
    })

    # Configure the decision logic for SwitchTask
    switch_task.switch_case("200", set_payment_success_variable)
    switch_task.default_case(set_payment_failure_variable)

    # Define the workflow and add the tasks
    workflow = ConductorWorkflow(
        name="payment_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(http_task)
    workflow.add(switch_task)

    # Register the workflow
    workflow.register(True)

    return workflow

Check out the full sample code for the HTTP Workflow.

HTTP Poll

The HTTP Poll task in Orkes Conductor enables workflows to send HTTP requests to a specified endpoint at regular intervals, continuing until a defined termination condition is met. This is particularly useful for monitoring the status of long-running operations or waiting for external processes to complete.

Example implementation

Consider a scenario where a workflow needs to monitor the status of a file processing job initiated by an external service. The workflow can use the HTTP Poll task to periodically check the status endpoint of the external service until the job is completed.

Here’s the workflow visualized:

Workflow using an HTTP Poll task.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Define the HTTP Poll input
    http_poll_input = HttpPollInput(
        uri="https://httpbin.org/delay/10",
        method="GET",
        accept="application/json",
        termination_condition="\$.output.response.statusCode == 200",
        polling_interval=10
    )

    # 2) Define the HTTP Poll task for checking the file processing status
    http_poll_task = HttpPollTask(
        task_ref_name="check_file_processing_status",
        http_input=http_poll_input
    )

    # 3) Define SetVariableTask for the success case
    set_processing_complete_variable = SetVariableTask(
        task_ref_name="set_processing_complete_variable"
    )
    set_processing_complete_variable.input_parameters.update({
        "processing_status": "COMPLETED"
    })

    # Define SetVariableTask for the failure case
    set_processing_failed_variable = SetVariableTask(
        task_ref_name="set_processing_failed_variable"
    )
    set_processing_failed_variable.input_parameters.update({
        "processing_status": "FAILED"
    })

    # 4) Define SwitchTask to handle success and failure cases
    switch_task = SwitchTask(
        task_ref_name="check_processing_status",
        case_expression="\${check_file_processing_status.output.response.statusCode}",
        use_javascript=False
    )

    # Configure the decision cases for the SwitchTask
    switch_task.switch_case(200, set_processing_complete_variable)
    switch_task.default_case(set_processing_failed_variable)

    # Define the workflow and add the tasks
    workflow = ConductorWorkflow(
        name="file_processing_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(http_poll_task)
    workflow.add(switch_task)

    # Register the workflow
    workflow.register(True)

    return workflow

Check out the full sample code for the HTTP Poll Workflow.

Inline

The Inline task in Orkes Conductor allows workflows to execute custom JavaScript code during runtime, facilitating dynamic data processing and decision-making without the need for external workers. This task is particularly useful for implementing lightweight logic, such as data transformations, conditional checks, or calculations, directly within the workflow.

Example implementation

In an e-commerce platform, the Inline task can be used to dynamically calculate prices based on various factors like loyalty status and active promotions. Instead of relying on an external service, the task executes custom JavaScript within the workflow to calculate the final price.

Here’s the workflow visualized:

Workflow using an Inline task.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) HTTP task to fetch a random number from random.org
    fetch_random_number_task = HttpTask(
        task_ref_name="fetch_random_number",
        http_input={
            "uri": "https://www.random.org/integers/?num=1&min=1&max=100&col=1&base=10&format=plain&rnd=new",
            "method": "GET",
            "headers": {
                "Content-Type": "application/json"
            }
        }
    )

    # 2) Set variable for base price using the random number
    set_base_price = SetVariableTask(
        task_ref_name="set_base_price"
    )
    set_base_price.input_parameters.update({
        "base_price": "\${fetch_random_number.output.response.body}"
    })

    # 3) Inline task to calculate the final price with discounts
    calculate_price_task = InlineTask(
        task_ref_name="calculate_final_price",
        script="""
            (function() {
                let basePrice = \$.base_price;
                let loyaltyDiscount = \$.loyalty_discount === "gold" ? 0.2 : 0;
                let promotionDiscount = \$.promotion_discount ? 0.1 : 0;
                return basePrice * (1 - loyaltyDiscount - promotionDiscount);
            })();
        """,
        bindings={
            "base_price": "\${workflow.variables.base_price}",
            "loyalty_discount": "\${workflow.input.loyalty_status}",
            "promotion_discount": "\${workflow.input.is_promotion_active}"
        }
    )

    # 4) Set the final calculated price in a workflow variable
    set_price_variable = SetVariableTask(
        task_ref_name="set_final_price_variable"
    )
    set_price_variable.input_parameters.update({
        "final_price": "\${calculate_final_price.output.result}"
    })

    # Define the workflow and add all tasks
    workflow = ConductorWorkflow(
        name="dynamic_pricing_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(fetch_random_number_task)
    workflow.add(set_base_price)
    workflow.add(calculate_price_task)
    workflow.add(set_price_variable)

    # Register the workflow
    workflow.register(overwrite=True)

    return workflow

Check out the full sample code for the Inline Workflow.

JSON JQ Transform

The JSON JQ Transform task in Orkes Conductor enables workflows to process and transform JSON data using JQ, a powerful command-line tool for querying and manipulating JSON. This task allows you to apply JQ expressions to input JSON, facilitating operations like filtering, mapping, and restructuring data directly within your workflow.

Example implementation

Consider a scenario where a workflow retrieves a list of user data from an external API, but only a subset of fields is required for subsequent processing. The JSON JQ Transform task can be employed to filter and format the API response efficiently.

Here’s the workflow visualized:

Workflow using an JSON JQ Transform task.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # JQ script to transform user data
    jq_script = """
      .users | map({
        username: .login,
        profile_url: .html_url,
        avatar: .avatar_url
      })
    """

    # 1) Create the JSON JQ task to transform incoming user data
    jq_task = JsonJQTask(
        task_ref_name="transform_user_data",
        script=jq_script
    )
    jq_task.input_parameters.update({
        "users": "\${workflow.input.users}"
    })

    # 2) Create a SetVariableTask to store the transformed result
    set_variable_task = SetVariableTask(
        task_ref_name="store_filtered_users"
    )
    set_variable_task.input_parameters.update({
        "filtered_users": "\${transform_user_data.output.result}"
    })

    # Define and register the workflow
    workflow = ConductorWorkflow(
        name="simple_user_data_transform",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.input_parameters = ["users"]
    workflow.add(jq_task)
    workflow.add(set_variable_task)

    # Register the workflow
    workflow.register(overwrite=True)

    return workflow

Check out the full sample code for the JSON JQ Transform Workflow.

Wait For Webhook

The Wait for Webhook task in Orkes Conductor is designed to pause a workflow’s execution until a specific external event is received via a webhook. This integration enables workflows to respond dynamically to real-time events from various platforms, including GitHub, Slack, Stripe, Microsoft Teams, and SendGrid. This is best when the external system can push events out.

Example implementation

A workflow waits for a webhook event indicating that a specific comment has been posted on a GitHub issue. Using a Wait for Webhook task, the workflow pauses until it receives the event with a comment from a specific author, such as “user123”. Once triggered, the workflow stores the comment content in a mock database, allowing subsequent tasks to process the comment, such as analyzing its content or triggering automated responses.

Here’s the workflow visualized:

Workflow using a Wait For Webhook task.

Here’s the code snippet for creating the workflow in code:

def register_workflow(workflow_executor: WorkflowExecutor) -> ConductorWorkflow:
    # 1) Define the WaitForWebhookTask to pause until a specific comment is received
    wait_for_comment_task = WaitForWebHookTask(
        task_ref_name="wait_for_comment",
        matches={
            "\$['comment']['author']": "user123"
        }
    )

    # 2) Define an HTTP task to store the received comment in a mock database
    store_comment_task = HttpTask(
        task_ref_name="store_comment",
        http_input={
            "uri": "https://jsonplaceholder.typicode.com/posts",
            "method": "POST",
            "headers": {
                "Content-Type": "application/json"
            },
            "body": {
                "comment_body": "\${wait_for_comment.output.comment.body}",
                "author": "user123"
            }
        }
    )

    # Define and register the workflow
    workflow = ConductorWorkflow(
        name="comment_trigger_workflow",
        executor=workflow_executor
    )
    workflow.version = 1
    workflow.add(wait_for_comment_task)
    workflow.add(store_comment_task)

    # Register the workflow
    workflow.register(overwrite=True)

    return workflow

Check out the full sample code for the Wait for Webhook Workflow.

Before running the workflow you must set up a webhook in Conductor for the workflow comment_trigger_workflow and GitHub.

After running the workflow, call this curl command to complete the workflow:

curl -H "Content-Type:application/json" -H "Accept:application/json" \
-H 'key: value' \
-X POST '[INSERT_WEBHOOK_URL]' \
-d '{"comment": {"body": "This is my comment.", "author": "user123"}}'

Alerting Tasks

Alerting tasks are specialized system tasks designed to monitor workflows and trigger alerts based on specific conditions. These tasks enhance observability and enable proactive incident management within your workflows, helping teams stay informed and act quickly when systems deviate from expected behavior. Explore how to use Alerting tasks with our suite of tutorials.

AI Tasks

Orkes Conductor offers a suite of AI tasks that facilitate seamless integration of Large Language Models (LLMs) and vector databases into workflows. These tasks enable functionalities such as text generation, embedding creation, document indexing, and retrieval, allowing developers to build both AI-driven applications and agentic workflows.

By incorporating these AI tasks, users can effectively orchestrate complex AI processes within their workflows, enhancing automation and intelligence in their applications. Explore how to use AI tasks with our suite of tutorials.

Wrap up

Using system tasks in Conductor speeds up development cycles, bringing processes and products from paper to reality in no time at all. In this post, we’ve demonstrated building workflows using system tasks through Python. As a developer-first platform, Orkes Conductor is built for creating workflows in any programming language (Java, Python, JavaScript, C#, Go, Clojure), or even using a visual diagram editor. Learn more about how exactly system tasks make developing workflows so much faster in Part 1.

Unlocking Developer Productivity: Orchestrate Workflows with Built-in System Tasks

Orkes Developers — Mon, 14 Apr 2025 11:20:55 +0000

This is Part 1 of the series on System Tasks. Check out Part 2, which dives into all the use cases and implementation details for Conductor’s most popular system tasks.

Orchestration-driven development can slash time-to-launch into half, especially when combined with built-in, reusable tasks for common cases, like calling an LLM, publishing events to a message broker, or even just calling an HTTP endpoint. Let’s dive deeper into Orkes Conductor’s system tasks and what makes them so powerful in unlocking developer productivity.

What is a workflow engine?

A workflow engine orchestrates complex processes by defining tasks in a declarative format (such as JSON). Unlike traditional in-code approaches, where developers handle retries, logging, and error management, workflow engines like Orkes Conductor automate task execution, abstracting these concerns.

Conductor abstracts away error management and plumbing code, making it effortless to build durable, stateful applications.

How to write workflows using system tasks? Check out our code implementation guide.

A workflow engine decouples execution code from business logic. This yields two time-saving advantages: one, the business logic of tasks and workflows are now reusable, scalable, and more easily maintainable. Two, developers can focus on task dependencies without having to worry about resilience and execution guarantees, which are managed by the engine.

Clear separation of concerns: the workflow engine handles execution, error management, and state tracking; while the business logic resides in the tasks.

Interested to learn more about how Conductor works under the hood? Check out the in-depth blog on durable execution.

Over time, common patterns emerge across applications, like data transformation, webhook integrations, and HTTP calls. This is where system tasks come in to automate business logic into reusable components, further speeding up development time for common use cases.

What are system tasks?

System tasks in Orkes Conductor represent reusable business logic components that combine to form orchestration patterns. These tasks run within the engine, allowing developers to quickly build complex workflows without reinventing the wheel, leading to a flexible, maintainable automation process.

All system tasks in Orkes Conductor.

When building workflows in Orkes Conductor, system tasks are your go-to tools for getting things done—fast. These are built-in, ready-to-use tasks that handle common operations like making HTTP calls, waiting for external events, or even running AI tasks一all without the need to write custom worker tasks. Since they run within Conductor’s JVM, system tasks help streamline development and remove the overhead of managing the custom infrastructure.

System tasks vs worker tasks in Conductor.

Why use system tasks?

System tasks bring clarity and consistency to workflows. Their standardized structure makes it easy to share, read, and reason about workflows across teams一no guesswork, no surprises.

They also make complex operations easier to express, with less boilerplate. Since Conductor handles the execution, you get enterprise-grade resilience without lifting a single finger. With Conductor’s built-in monitoring and debugging tools, it’s easier to track failures, inspect state, and keep things running smoothly.

Benefits

System tasks offer a declarative, reliable way to build workflows. By abstracting common logic into reusable components, they reduce development effort, enhance clarity, and ensure consistent behavior—making them ideal for predictable, maintainable workflow design.

Time to Market: Eliminate the need for custom code, accelerating development for tasks like HTTP calls, database updates, and event handling.
Standardization: Offer consistency across workflows, making them easier to understand than custom worker task implementations.
Scalability: Workflows using system tasks are inherently easier to scale, since execution is managed by Orkes Conductor itself.

What about custom business logic?

Conductor offers all the benefits of workflow orchestration, even for custom business logic. This can be done through two approaches: Inline tasks and Worker tasks.

Inline tasks let you execute lightweight JavaScript code directly within the workflow without the need for external workers. For more complex tasks, you can deploy custom worker tasks. Worker tasks provide full flexibility to implement advanced logic, such as API calls and long-running operations, while still benefiting from Conductor’s automated retries, status updates, and monitoring.

Inline task approach

An Inline task is a type of system task that allows you to execute custom JavaScript code directly within the workflow. It’s ideal for lightweight operations like simple data transformations, conditional checks, or small calculations. Inline tasks are executed within the Conductor JVM, which means they benefit from low latency, no network overhead, and easier debugging.

Find out how to use Inline tasks in Part Two of this blog series.

Key Features:

Executes directly within the Conductor JVM.
Best for lightweight logic and transformations.
No network calls or external dependencies.
Easier to implement and debug but limited to JavaScript.

Worker task approach

A Worker task (also known as Simple task) is an external custom function or service that performs a specific task in a workflow. Written in any language of choice (Python, Java, etc.), it can execute complex business logic, custom algorithms, or long-running operations. Worker tasks run outside the Conductor server, meaning they require additional infrastructure set-up and logging mechanisms.

Key Features:

External execution (outside the Conductor JVM).
Full flexibility to implement complex logic.
Requires task-level logging within the worker code.
Typically involves network calls (e.g., HTTP requests).

System tasks vs Worker tasks

Both system tasks and Worker tasks offer different advantages, and the choice to use one over the other depends on your requirements. Here’s a comparison between the two.

	System tasks	Worker tasks
Advantages	Speed: System tasks run within the Conductor JVM, enabling fast, low-latency execution without network overhead, serialization processing, or external polling. Availability Guarantees: System tasks are always guaranteed to be available and operate within Conductor’s JVM.	Full Control: You can implement more complex or specialized logic that may not be easily achievable using system tasks. Flexibility: Workers can be deployed, restarted, and updated independently from the workflow. Decoupling: Worker crashes are isolated and will not bring the entire system down.
Trade-offs	Tight Coupling: System tasks are harder to scale, redeploy, or maintain independently. Limited Customizability: There is limited leeway for using custom libraries, frameworks, or stacks in system tasks.	Increased Development Effort: Customizing each worker adds development overhead for writing, testing, deploying, and maintaining more code. Monitoring & Debugging Complexity: With workers, you need to implement your own task-level logging within the worker code. Infrastructure Overhead: Workers rely on the availability and health of the external environment in which they're running, which has to be self-maintained.

Wrap up

Orkes Conductor is a workflow engine that offers the best of both worlds: low-code agility when you need to build out common use cases, and full customization control when you need to dive deep into building unique solutions. It’s a development platform made to empower developers in a fast-paced world. In Part Two of this blog series, discover how to use system tasks in Conductor, as we spill the full implementation details.