Cloudflare Workflows 는 단계를 연결하고, 실패 시 재시도하며, 장기 실행 프로세스에서 상태를 유지할 수 있는 지속형 실행 엔진입니다. 개발자는 Workflows를 사용하여 백그라운드 에이전트를 구동하고, 데이터 파이프라인을 관리하며, 휴먼인더루프(human-in-the-loop) 승인 시스템을 구축하는 등의 작업을 수행할 수 있습니다.
지난 달, 저희는 Cloudflare에 배포된 모든 워크플로의 대시보드에 완전한 시각적 다이어그램이 표시된다고 발표했습니다.
애플리케이션을 시각화할 수 있는 능력이 그 어느 때보다 중요한 지금 이러한 상황이 되었습니다. 코딩 에이전트는 사용자가 읽을 수도 있고 읽지 않을 수도 있는 코드를 작성합니다. 하지만 단계가 어떻게 연결되는지, 어디에서 분기되며, 실제로 어떤 일이 일어나고 있는지 구축하는 형태가 여전히 중요합니다.
이전에 비주얼 워크플로우 빌더의 다이어그램을 본 적이 있다면, 이들은 일반적으로 JSON 구성, YAML, 드래그 앤 드롭 등 선언적인 것으로 작동합니다. 하지만 Cloudflare Workflows는 코드에 불과합니다. 여기에는 Promises, Promise.all, 루프, 조건문 이 포함되거나 함수 또는 클래스에 중첩될 수 있습니다. 이 동적 실행 모델 때문에 다이어그램 렌더링이 조금 더 복잡해집니다.
추상 구문 트리(AST)를 사용하여 그래프를 정적으로 도출하고, Promise 및 await 관계를 추적하여 무엇이 병렬로 실행되는지, 어떤 블록이 작동하는지, 조각들이 어떻게 연결되는지 이해합니다.
계속 읽으면서 다이어그램을 구축한 방법을 알아보거나, 첫 워크플로우를 배포하고 다이어그램을 직접 확인하세요.
다음은 Cloudflare Workflows 코드에서 생성된 다이어그램의 예입니다.
일반적으로 워크플로우 엔진은 동적 또는 순차적(정적) 실행 순서에 따라 실행될 수 있습니다. 순차적 실행이 더 직관적인 솔루션처럼 보일 수 있습니다. 워크플로 → A 단계 → B 단계 → 엔진이 A 단계를 완료한 후 즉시 B 단계의 실행이 시작되는 C 단계를 트리거합니다.
Cloudflare Workflows 는 동적 실행 모델을 따릅니다. 워크플로우는 코드이므로 런타임이 단계를 만나면 실행됩니다. 런타임에서 단계를 검색하면 해당 단계는 워크플로우 엔진으로 전달되며, 워크플로우 엔진에서는 실행을 관리합니다. await를 거치지 않는 한 단계는 본질적으로 순차적이지 않습니다. 엔진은 기다리지 않은 모든 단계를 병렬로 실행합니다. 이러한 방식으로 추가 래퍼나 지시문 없이 워크플로 코드를 흐름 제어로 작성할 수 있습니다. 핸드오프가 작동하는 방식은 다음과 같습니다.
해당 인스턴스의 "슈퍼바이저" Durable Object 역할을 하는 엔진이 작동합니다. 엔진은 실제 워크플로 실행 로직을 담당합니다.
엔진은 사용자 Worker 를 동적 디스패치를 통해 트리거하여 Workers 런타임으로 제어권을 넘깁니다.
런타임이 step.do를 만나면, 실행을 엔진으로 다시 넘깁니다.
엔진이 단계를 실행하고, 결과를 유지한(또는 해당하는 경우 오류가 발생함), 사용자 Worker를 다시 트리거합니다.
이 아키텍처에서는 엔진이 본질적으로 실행 중인 단계의 순서를 "알 수는 없지만" 다이어그램의 경우 단계 순서가 중요한 정보가 됩니다. 대부분의 워크플로를 진단에 유용한 그래프로 정확하게 변환하는 것이 과제입니다. 베타 버전의 다이어그램을 사용하여 이러한 표현을 계속 반복하고 개선할 예정입니다.
런타임이 아니라 배포 시 스크립트를 가져오면 전체 워크플로우를 구문 분석하여 다이어그램을 정적으로 생성할 수 있습니다.
한 걸음 물러서서, 워크플로우 배포의 수명은 다음과 같습니다.
다이어그램을 생성하기 위해 Workers를 배포하는 내부 구성 서비스에서 번들로 제공한 스크립트를 가져옵니다(Workflow 배포의 2단계). 그런 다음 파서를 사용하여 워크플로를 나타내는 추상 구문 트리(AST)를 생성하면 내부 서비스에서 모든 WorkflowEntrypoint와 워크플로 단계에 대한 호출이 있는 중간 그래프를 생성하고 순회합니다. Cloudflare에서는 API의 최종 결과에 따라 다이어그램을 렌더링합니다.
Worker가 배포되면 달리 명시되지 않는 한 구성 서비스는 코드를 번들로 묶고(기본적으로 esbuild 사용) 최소화합니다. 이는 또 다른 문제를 야기합니다. TypeScript의 Workflows는 직관적인 패턴을 따르지만, 축소된 Javascript(JS)는 밀도가 높고 소화하기 어려울 수 있습니다. 번들러에 따라 코드를 최소화할 수 있는 다른 방법도 있습니다.
다음은 병렬로 실행되는 에이전트를 보여주는 Workflow 코드의 예입니다.
const summaryPromise = step.do(
`summary agent (loop ${loop})`,
async () => {
return runAgentPrompt(
this.env,
SUMMARY_SYSTEM,
buildReviewPrompt(
'Summarize this text in 5 bullet points.',
draft,
input.context
)
);
}
);
const correctnessPromise = step.do(
`correctness agent (loop ${loop})`,
async () => {
return runAgentPrompt(
this.env,
CORRECTNESS_SYSTEM,
buildReviewPrompt(
'List correctness issues and suggested fixes.',
draft,
input.context
)
);
}
);
const clarityPromise = step.do(
`clarity agent (loop ${loop})`,
async () => {
return runAgentPrompt(
this.env,
CLARITY_SYSTEM,
buildReviewPrompt(
'List clarity issues and suggested fixes.',
draft,
input.context
)
);
}
);
rspack과 함께 번들링하면 축소된 코드의 일부는 다음과 같습니다.
class pe extends e{async run(e,t){de("workflow.run.start",{instanceId:e.instanceId});const r=await t.do("validate payload",async()=>{if(!e.payload.r2Key)throw new Error("r2Key is required");if(!e.payload.telegramChatId)throw new Error("telegramChatId is required");return{r2Key:e.payload.r2Key,telegramChatId:e.payload.telegramChatId,context:e.payload.context?.trim()}}),s=await t.do("load source document from r2",async()=>{const e=await this.env.REVIEW_DOCUMENTS.get(r.r2Key);if(!e)throw new Error(`R2 object not found: ${r.r2Key}`);const t=(await e.text()).trim();if(!t)throw new Error("R2 object is empty");return t}),n=Number(this.env.MAX_REVIEW_LOOPS??"5"),o=this.env.RESPONSE_TIMEOUT??"7 days",a=async(s,i,c)=>{if(s>n)return le("workflow.loop.max_reached",{instanceId:e.instanceId,maxLoops:n}),await t.do("notify max loop reached",async()=>{await se(this.env,r.telegramChatId,`Review stopped after ${n} loops for ${e.instanceId}. Start again if you still need revisions.`)}),{approved:!1,loops:n,finalText:i};const h=t.do(`summary agent (loop ${s})`,async()=>te(this.env,"You summarize documents. Keep the output short, concrete, and factual.",ue("Summarize this text in 5 bullet points.",i,r.context)))...
또는, vite와 함께 번들로 제공되는 최소화된 스니펫은 다음과 같습니다.
class ht extends pe {
async run(e, r) {
b("workflow.run.start", { instanceId: e.instanceId });
const s = await r.do("validate payload", async () => {
if (!e.payload.r2Key)
throw new Error("r2Key is required");
if (!e.payload.telegramChatId)
throw new Error("telegramChatId is required");
return {
r2Key: e.payload.r2Key,
telegramChatId: e.payload.telegramChatId,
context: e.payload.context?.trim()
};
}), n = await r.do(
"load source document from r2",
async () => {
const i = await this.env.REVIEW_DOCUMENTS.get(s.r2Key);
if (!i)
throw new Error(`R2 object not found: ${s.r2Key}`);
const c = (await i.text()).trim();
if (!c)
throw new Error("R2 object is empty");
return c;
}
), o = Number(this.env.MAX_REVIEW_LOOPS ?? "5"), l = this.env.RESPONSE_TIMEOUT ?? "7 days", a = async (i, c, u) => {
if (i > o)
return H("workflow.loop.max_reached", {
instanceId: e.instanceId,
maxLoops: o
}), await r.do("notify max loop reached", async () => {
await J(
this.env,
s.telegramChatId,
`Review stopped after ${o} loops for ${e.instanceId}. Start again if you still need revisions.`
);
}), {
approved: !1,
loops: o,
finalText: c
};
const h = r.do(
`summary agent (loop ${i})`,
async () => _(
this.env,
et,
K(
"Summarize this text in 5 bullet points.",
c,
s.context
)
)
)...
축소된 코드는 상당히 복잡해질 수 있으며, 번들러에 따라 여러 방향으로 복잡해질 수 있습니다.
우리는 다양한 형태의 축소된 코드를 빠르고 정확하게 구문 분석할 방법이 필요했습니다. 저희는 oxc-parser 가 JavaScript Oxidation Compiler (OXC)에서 이 작업에 완벽하다고 판단했습니다. 저희는 Rust를 실행하는 컨테이너를 통해 이 아이디어를 먼저 테스트했습니다. 모든 스크립트 ID가 Cloudflare Queue로 전송된 다음, 메시지가 팝업되어 처리할 컨테이너로 전송되었습니다. 이 접근 방식이 효과가 있음을 확인한 후, Cloudflare는 Rust로 작성된 Worker로 이전했습니다. Workers는 WebAssembly를 통한 Rust 실행을 지원하며, 패키지는 이 과정이 간단할 만큼 작았습니다.
Rust Worker는 먼저 축소된 JS를 AST 노드 유형으로 변환한 다음, AST 노드 유형을 대시보드에 렌더링되는 그래픽 버전의 워크플로로 변환하는 일을 담당합니다. 이를 위해 각 워크플로우에 대해 미리 정의된 노드 유형 의 그래프를 생성하고, 일련의 노드 매핑을 통해 그래프 표현으로 변환합니다.
워크플로의 다이어그램 버전을 렌더링하는 데는 두 가지 과제가 있었습니다. 단계와 기능 관계를 올바르게 추적하는 방법, 그리고 모든 표면 영역을 다루면서 워크플로 노드 유형을 최대한 간단하게 정의하는 방법이었습니다.
단계와 기능 관계를 올바르게 추적하려면 기능과 단계 이름을 모두 수집해야 했습니다. 앞서 설명한 것처럼 엔진에는 단계에 대한 정보만 있지만, 단계는 기능에 종속될 수 있으며, 그 반대의 경우도 마찬가지입니다. 예를 들어, 개발자는 함수로 단계를 래핑하거나 함수를 단계로 정의할 수 있습니다. 또한 다른 모듈 에서 온 함수 내에서 단계를 호출하거나 단계의 이름을 바꿀 수 있습니다.
라이브러리는 AST를 제공하여 초기 장애물을 통과하지만, 우리는 여전히 라이브러리를 구문 분석하는 방법을 결정해야 합니다. 추가적인 창의성이 필요한 코드 패턴도 있습니다. 예를 들어 함수 — WorkflowEntrypoint 내에는 단계를 직접, 간접적으로 호출하거나, 호출하지 않는 함수가 있을 수 있습니다. 고려해 보겠습니다 functionA는 console.log(await functionB(), await functionC())를 포함하며, 여기서 functionB 는 step.do()를 호출합니다. 이 경우 functionA 와 functionB 는 모두 워크플로우 다이어그램에 포함되어야 합니다. 하지만 functionC 는 포함되어서는 안 됩니다. 직접 및 간접 단계 호출이 포함된 모든 함수를 포착하기 위해 각 함수에 대한 하위 그래프를 만들고 여기에 단계 호출 자체가 포함되어 있는지 또는 그것이 있을 수 있는 다른 함수를 호출하는지 여부를 확인합니다. 이러한 하위 그래프는 모든 관련 노드를 포함하는 기능 노드로 표현됩니다. 함수 노드가 그래프의 리프이거나 직접 또는 간접적인 워크플로우 단계가 없는 경우에는 최종 출력에서 잘립니다.
저희는 최대 10가지 방법으로 정의된 워크플로우 다이어그램 또는 변수를 추론할 수 있는 정적 단계 목록을 포함하여 다른 패턴도 확인합니다. 스크립트에 여러 워크플로우가 포함된 경우 저희는 한 수준 더 높은 수준에서 추상화된 함수에 대해 생성된 하위 그래프와 유사한 패턴을 따릅니다.
모든 AST 노드 유형에 대해, 워크플로우 내에서 사용할 수 있는 모든 방법을 고려해야 했습니다. 루프, 분기, 프라미스, 병렬, 대기, 화살표 함수 등 목록의 계속입니다. 이러한 경로 안에도 수십 개의 가능성이 존재합니다. 루프를 실행하는 몇 가지 방법을 고려해보세요.
// for...of
for (const item of items) {
await step.do(`process ${item}`, async () => item);
}
// while
while (shouldContinue) {
await step.do('poll', async () => getStatus());
}
// map
await Promise.all(
items.map((item) => step.do(`map ${item}`, async () => item)),
);
// forEach
await items.forEach(async (item) => {
await step.do(`each ${item}`, async () => item);
});
루핑을 넘어서서 분기를 처리하는 방법은 다음과 같습니다.
// switch / case
switch (action.type) {
case 'create':
await step.do('handle create', async () => {});
break;
default:
await step.do('handle unknown', async () => {});
break;
}
// if / else if / else
if (status === 'pending') {
await step.do('pending path', async () => {});
} else if (status === 'active') {
await step.do('active path', async () => {});
} else {
await step.do('fallback path', async () => {});
}
// ternary operator
await (cond
? step.do('ternary true branch', async () => {})
: step.do('ternary false branch', async () => {}));
// nullish coalescing with step on RHS
const myStepResult =
variableThatCanBeNullUndefined ??
(await step.do('nullish fallback step', async () => 'default'));
// try/catch with finally
try {
await step.do('try step', async () => {});
} catch (_e) {
await step.do('catch step', async () => {});
} finally {
await step.do('finally step', async () => {});
}
우리의 목표는 너무 복잡하지 않으면서 개발자가 알아야 할 것을 전달할 수 있는 간결한 API를 만드는 것이었습니다. 하지만 워크플로우를 다이어그램으로 변환하려면 가능한 모든 패턴(모범 사례 준수 여부 여부)과 에지 사례를 고려해야 했습니다. 앞서 설명한 것처럼 각 단계는 기본적으로 다른 단계로 명시적으로 순차적이지 않습니다. 워크플로가 await 및 Promise.all()을 활용하지 않으면, 단계가 발생한 순서대로 실행될 것으로 가정합니다. 하지만 워크플로에 await, Promise 또는 Promise.all()이 포함된 경우, 우리는 이러한 관계를 추적할 방법이 필요했습니다.
우리는 각 노드에 starts: 및 resolves: 필드가 있는 실행 순서를 추적하기로 결정했습니다. starts 및 resolves 인덱스는 프라미스가 실행을 시작한 시점과 즉각적인 후속 결론 없이 시작된 첫 번째 프라미스를 기준으로 종료되는 시점을 알려줍니다. 이는 다이어그램 UI에서의 수직적 위치 지정과 관련이 있습니다(즉, starts:1 인 모든 단계가 인라인입니다). 단계가 선언될 때 await가 사용되는 경우 starts 및 resolves 이 정의되지 않고 런타임에 단계가 표시된 순서대로 워크플로가 실행됩니다.
구문 분석을 하는 동안 기다리지 않은 Promise 또는 Promise.all()을 만나면 해당 노드에는 starts 필드에 나타난 엔트리 번호가 표시됩니다. 해당 프라미스에서 await 가 발생하면 엔트리 번호가 1씩 증가하고 종료 번호에 저장됩니다(resolves의 값입니다). 이렇게 하면 어떤 프라미스가 동시에 실행되고 언제 완료되는지 서로 연관하여 알 수 있습니다.
export class ImplicitParallelWorkflow extends WorkflowEntrypoint<Env, Params> {
async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
const branchA = async () => {
const a = step.do("task a", async () => "a"); //starts 1
const b = step.do("task b", async () => "b"); //starts 1
const c = await step.waitForEvent("task c", { type: "my-event", timeout: "1 hour" }); //starts 1 resolves 2
await step.do("task d", async () => JSON.stringify(c)); //starts 2 resolves 3
return Promise.all([a, b]); //resolves 3
};
const branchB = async () => {
const e = step.do("task e", async () => "e"); //starts 1
const f = step.do("task f", async () => "f"); //starts 1
return Promise.all([e, f]); //resolves 2
};
await Promise.all([branchA(), branchB()]);
await step.sleep("final sleep", 1000);
}
}
다이어그램에서 단계의 정렬을 볼 수 있습니다.
이러한 패턴을 모두 고려한 후 다음과 같은 노드 유형 목록을 결정했습니다.
| StepSleep
| StepDo
| StepWaitForEvent
| StepSleepUntil
| LoopNode
| ParallelNode
| TryNode
| BlockNode
| IfNode
| SwitchNode
| StartNode
| FunctionCall
| FunctionDef
| BreakNode;
다음은 다양한 동작에 대한 API 출력 샘플입니다.
function 호출:
{
"functions": {
"runLoop": {
"name": "runLoop",
"nodes": []
}
}
}
if 조건이 step.do로 분기됩니다:
{
"type": "if",
"branches": [
{
"condition": "loop > maxLoops",
"nodes": [
{
"type": "step_do",
"name": "notify max loop reached",
"config": {
"retries": {
"limit": 5,
"delay": 1000,
"backoff": "exponential"
},
"timeout": 10000
},
"nodes": []
}
]
}
]
}
step.do 및 waitForEvent와 waitForEvent.
{
"type": "parallel",
"kind": "all",
"nodes": [
{
"type": "step_do",
"name": "correctness agent (loop ${...})",
"config": {
"retries": {
"limit": 5,
"delay": 1000,
"backoff": "exponential"
},
"timeout": 10000
},
"nodes": [],
"starts": 1
},
...
{
"type": "step_wait_for_event",
"name": "wait for user response (loop ${...})",
"options": {
"event_type": "user-response",
"timeout": "unknown"
},
"starts": 3,
"resolves": 4
}
]
}
이 Workflow 다이어그램의 최종 목표는 풀 서비스 디버깅 도구로 기능하는 것입니다. 즉, 다음을 수행할 수 있습니다.
워크플로 개요 페이지에서 다이어그램을 확인하세요. 기능 요청이나 버그를 발견하면 Discord의 Cloudflare 개발자 커뮤니티에 참여하여 Cloudflare 팀에 직접 피드백을 공유해 주세요.