4.2 Control Flow

Tujuan bab

Setelah bab ini kamu bisa:

• Menjalankan langkah secara paralel dengan .parallel() dan membaca struktur keluarannya
• Memilih jalur berdasarkan kondisi dengan .branch()
• Mengulang langkah dengan .dountil(), .dowhile(), dan .foreach()
• Memilih antara .parallel() dan .foreach() dengan tepat
• Mengubah bentuk data antar langkah dengan .map()

Prasyarat

• 4.1 Workflow Pertama

Peta lengkap

Sebelum masuk satu per satu, ini gambaran keseluruhannya:

Tabel rujukan cepat yang layak kamu tandai:

Metode	Kegunaan	Masukan	Keluaran	Konkurensi
.then(step)	Pemrosesan berurutan	T	U	Satu per satu
.parallel([a, b])	Operasi berbeda atas masukan sama	T	{ a: U, b: V }	Semua bersamaan
.foreach(step)	Operasi sama atas tiap item array	T[]	U[]	Bisa diatur, default 1
.branch([...])	Memilih satu jalur	T	{ stepTerpilih: U }	Hanya satu yang jalan

.parallel() — beberapa langkah sekaligus

Pakai saat beberapa pekerjaan tidak saling bergantung dan bisa dikerjakan bersamaan. Semua langkah paralel harus selesai sebelum workflow melanjutkan.

// src/mastra/workflows/analisis-workflow.ts
import { createWorkflow, createStep } from '@mastra/core/workflows'
import { z } from 'zod'

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ formatted: z.string() }),
  execute: async ({ inputData }) => ({
    formatted: inputData.message.toUpperCase(),
  }),
})

const step2 = createStep({
  id: 'step-2',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ emphasized: z.string() }),
  execute: async ({ inputData }) => ({
    emphasized: `${inputData.message}!!!`,
  }),
})

const step3 = createStep({
  id: 'step-3',
  inputSchema: z.object({
    'step-1': z.object({ formatted: z.string() }),
    'step-2': z.object({ emphasized: z.string() }),
  }),
  outputSchema: z.object({ combined: z.string() }),
  execute: async ({ inputData }) => {
    const { formatted } = inputData['step-1']
    const { emphasized } = inputData['step-2']
    return { combined: `${formatted} | ${emphasized}` }
  },
})

export const analisisWorkflow = createWorkflow({
  id: 'analisis-workflow',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ combined: z.string() }),
})
  .parallel([step1, step2])
  .then(step3)
  .commit()

Bagian terpenting dari contoh ini ada di inputSchema milik step3.

Keluaran langkah paralel dikunci berdasarkan id langkahnya. Bukan digabung jadi satu objek datar, melainkan bersarang di bawah nama tiap langkah:

{
  'step-1': { formatted: 'HALO' },
  'step-2': { emphasized: 'Halo!!!' },
}

Karena itu langkah setelah .parallel() harus mendeklarasikan schema bersarang seperti di atas, dan mengaksesnya dengan inputData['step-1']. Ini sumber kebingungan yang sangat umum saat pertama memakai .parallel() — orang menulis inputData.formatted dan mendapat undefined.

Konsekuensi praktis dari penguncian berdasarkan id

Karena id langkah menjadi kunci pada objek keluaran, mengubah id sebuah langkah paralel akan merusak langkah setelahnya. Perlakukan id langkah yang dipakai dalam .parallel() seperti nama field publik.

.branch() — memilih jalur

export const penilaianWorkflow = createWorkflow({
  id: 'penilaian-workflow',
  inputSchema: z.object({ value: z.number() }),
  outputSchema: z.object({ result: z.string() }),
})
  .then(step1)
  .branch([
    [async ({ inputData: { value } }) => value > 10, stepA],
    [async ({ inputData: { value } }) => value <= 10, stepB],
  ])
  .commit()

Tiap cabang berupa pasangan [kondisi, langkah]. Kondisinya fungsi async yang mengembalikan boolean.

Ada satu aturan yang harus dipatuhi dan sering dilanggar: semua langkah dalam sebuah branch harus punya inputSchema dan outputSchema yang sama. Alasannya masuk akal — percabangan hanya bisa bekerja kalau semua jalur bisa saling menggantikan, dan itu menuntut kontrak yang seragam.

Struktur keluarannya serupa .parallel(): hasilnya dikunci berdasarkan id langkah yang benar-benar dijalankan. Bedanya, hanya satu cabang yang jalan — yang kondisinya pertama kali bernilai true.

Pastikan ada cabang yang selalu cocok

Kalau tidak ada kondisi yang terpenuhi, tidak ada langkah yang dijalankan. Perhatikan contoh di atas: value > 10 dan value <= 10 bersama-sama menutupi seluruh kemungkinan. Kalau cabang keduamu ditulis value < 10, nilai tepat 10 akan lolos tanpa penanganan.

Tiga cara mengulang

.dountil() — ulangi sampai kondisi benar

export const hitungWorkflow = createWorkflow({ /* ... */ })
  .then(step1)
  .dountil(step2, async ({ inputData: { number } }) => number > 10)
  .commit()

Langkah dijalankan berulang sampai kondisinya bernilai true.

.dowhile() — ulangi selama kondisi benar

export const hitungWorkflow = createWorkflow({ /* ... */ })
  .then(step1)
  .dowhile(step2, async ({ inputData: { number } }) => number < 10)
  .commit()

Keduanya menghasilkan perilaku yang setara kalau kondisinya kamu balik. Pilih yang kalimatnya paling wajar dibaca untuk kasusmu — dountil(… > 10) dibaca "ulangi sampai lebih dari 10", dowhile(… < 10) dibaca "ulangi selama kurang dari 10".

.foreach() — jalankan untuk tiap item array

const step1 = createStep({
  id: 'huruf-besar',
  inputSchema: z.string(),
  outputSchema: z.string(),
  execute: async ({ inputData }) => inputData.toUpperCase(),
})

Masukannya harus bertipe array supaya loop bisa menelusuri isinya. Yang diterima tiap iterasi adalah satu item, bukan seluruh array — perhatikan bahwa inputSchema di atas adalah z.string(), bukan z.array(z.string()).

Konkurensinya bisa diatur:

workflow.foreach(downloadStep, { concurrency: 5 }).then(aggregateStep)

Default-nya 1, artinya item diproses satu per satu. Menaikkannya berguna untuk pekerjaan yang menunggu jaringan — tapi hati-hati dengan batas rate API yang kamu panggil.

.parallel() atau .foreach()?

Ini pasangan yang paling sering tertukar. Dokumentasi memberi pembeda yang tajam:

.parallel() — satu masukan, beberapa pemrosesan berbeda:

// Data pengguna yang sama, diproses dengan cara berbeda-beda
workflow.parallel([validateStep, enrichStep, scoreStep]).then(combineResultsStep)

.foreach() — banyak masukan, satu pemrosesan yang sama:

// Banyak URL, masing-masing diproses dengan cara yang sama
workflow.foreach(downloadStep, { concurrency: 5 }).then(aggregateStep)

Pertanyaan pemutusnya: apakah yang berbeda itu pekerjaannya, atau datanya? Pekerjaan berbeda → .parallel(). Data berbeda → .foreach().

.map() — menjembatani schema

Kadang keluaran satu langkah tidak persis cocok dengan masukan langkah berikutnya. Alih-alih mengubah salah satu schema — yang bisa merusak pemakaian lain — sisipkan .map():

export const testWorkflow = createWorkflow({ /* ... */ })
  .map(async ({ inputData }) => {
    const { message } = inputData
    return {
      prompt: `Ubah pesan ini menjadi poin-poin: ${message}`,
    }
  })
  .then(agentStep)
  .commit()

Ini juga cara utama menyiapkan masukan untuk agent yang dipakai sebagai step, yang dibahas di 4.4.

Kapan memakai nested workflow

Untuk kasus di mana tiap item array butuh beberapa langkah berurutan, gabungkan .foreach() dengan workflow bersarang: bungkus rangkaian langkah itu menjadi satu workflow, lalu pakai workflow itu sebagai langkah di dalam .foreach().

Ini juga menjaga workflow utamamu tetap terbaca. Workflow dengan dua puluh .then() berurutan sulit dipahami; workflow dengan lima langkah yang masing-masing merupakan workflow bernama jauh lebih mudah.

Kesalahan umum

Gejala: Setelah .parallel(), langkah berikutnya menerima undefined saat mengakses field hasil. Penyebab: Keluaran langkah paralel dikunci berdasarkan id langkah, jadi strukturnya bersarang. Mengakses inputData.formatted tidak akan menemukan apa pun karena yang benar adalah inputData['step-1'].formatted. Perbaikan: Deklarasikan inputSchema bersarang dengan id tiap langkah paralel sebagai kunci, dan akses lewat kunci itu.

Gejala: Workflow berhenti tanpa hasil setelah mencapai .branch(), tanpa error. Penyebab: Tidak ada kondisi cabang yang terpenuhi, sehingga tidak ada langkah yang dijalankan. Sering terjadi pada nilai batas — misalnya kondisi > 10 dan < 10 yang melewatkan nilai tepat 10. Perbaikan: Pastikan kondisi-kondisi cabang bersama-sama menutupi seluruh kemungkinan masukan. Cara paling aman adalah membuat cabang terakhir sebagai penampung yang kondisinya selalu benar.

Coba sendiri

Tantangan: Bangun pesanan-workflow yang memproses satu pesanan e-commerce, memakai keempat pola alur dalam satu workflow.

Alur yang diminta:

1. Validasi — langkah tunggal yang memeriksa kelengkapan data pesanan
2. Pemeriksaan paralel — tiga langkah bersamaan: cek stok, cek alamat pengiriman, dan cek metode pembayaran
3. Penggabungan — satu langkah yang membaca ketiga hasil paralel dan menyimpulkan apakah pesanan layak diproses
4. Percabangan — kalau layak, jalankan langkah "proses"; kalau tidak, jalankan langkah "tolak"
5. .foreach() — untuk tiap item dalam pesanan, hitung berat pengirimannya

Ketentuan:

1. Langkah penggabungan harus membaca hasil paralel lewat kunci id
2. Kedua cabang punya inputSchema dan outputSchema yang identik
3. Sertakan cabang penampung supaya tidak ada masukan yang lolos tanpa penanganan
4. Bagian .foreach() memakai concurrency selain default, dan kamu bisa menjelaskan alasan angkanya
5. Sisipkan minimal satu .map() untuk menjembatani schema yang tidak cocok
6. Jalankan tiga skenario: pesanan yang lolos semua pemeriksaan, pesanan yang gagal di satu pemeriksaan, dan pesanan dengan lima item

Checklist penerimaan:

• [ ] Ketiga pemeriksaan berjalan paralel — terlihat di Studio bahwa ketiganya mulai bersamaan
• [ ] Langkah penggabungan berhasil membaca ketiga hasil lewat kunci id
• [ ] Skenario gagal masuk ke cabang "tolak", bukan "proses"
• [ ] .foreach() memproses kelima item dan menghasilkan array hasil
• [ ] Tidak ada masukan yang membuat workflow berhenti tanpa cabang terpilih
• [ ] Kamu bisa menjelaskan kenapa pemeriksaan stok/alamat/pembayaran memakai .parallel() sementara perhitungan berat item memakai .foreach()

Petunjuk: Checklist terakhir adalah inti pemahaman bab ini. Tiga pemeriksaan itu adalah pekerjaan berbeda atas satu pesanan yang sama; perhitungan berat adalah pekerjaan sama atas banyak item berbeda. Kalau kamu bisa mengucapkan perbedaan itu tanpa melihat catatan, kamu sudah menguasai bagian tersulit dari bab ini.

Ikhtisar

• .then() berurutan, .parallel() bersamaan, .branch() memilih satu jalur, dan tiga metode loop untuk pengulangan.
• Keluaran .parallel() dan .branch() dikunci berdasarkan id langkah — strukturnya bersarang, bukan datar. Ini penyebab undefined yang paling umum.
• Semua langkah dalam satu .branch() harus punya schema yang sama, dan kondisi-kondisinya sebaiknya menutupi seluruh kemungkinan.
• .dountil() mengulang sampai kondisi benar; .dowhile() mengulang selama kondisi benar. Pilih yang paling wajar dibaca.
• .foreach() menjalankan satu langkah untuk tiap item array. Masukannya array, tapi tiap iterasi menerima satu item. Konkurensi default-nya 1.
• Pembeda .parallel() dan .foreach(): pekerjaan yang berbeda versus data yang berbeda.
• .map() menjembatani schema yang tidak cocok tanpa mengubah definisi langkah.

Lanjut ke mana

Sampai sini data hanya mengalir dari satu langkah ke langkah berikutnya. Tapi bagaimana kalau langkah kelima butuh sesuatu yang dihitung di langkah pertama, tanpa harus mengopernya lewat setiap langkah di antaranya? 4.3 Workflow State.