동적 Workflow 정의

:::실험적

동적 Workflow는 베타 버전입니다. API가 안정될 때까지 주요 버전 변경 없이 주요 변경 사항이 발생할 수 있습니다.

:::

동적 Workflow 정의는 Mastra.addDynamicWorkflow(), 저장된 Workflow 서버 경로 및 Client SDK Workflow API에서 허용하는 JSON 호환 DynamicWorkflowGraph입니다. 전체 설정 및 사용 예시는 동적 Workflow를 참조하세요.

정의 필드

필드	유형	필수	설명
id	string	예	고유한 Workflow ID입니다. Workflow를 가져오고 실행할 때도 이 ID를 사용합니다.
description	string	아니요	사람이 읽을 수 있는 설명
inputSchema	JsonSchema	예	Workflow 입력의 JSON 스키마
outputSchema	JsonSchema	예	Workflow 출력의 JSON 스키마
stateSchema	JsonSchema	아니요	공유 Workflow 상태의 JSON 스키마
requestContextSchema	JsonSchema	아니요	요청 컨텍스트에서 읽는 값의 JSON 스키마
metadata	Record<string, unknown>	아니요	스토리지를 통해 보존되는 임의의 JSON 메타데이터
graph	SerializedStepFlowEntry[]	예	Workflow를 구성하는 단계 항목
스키마는 Zod 대신 JSON 스키마를 사용하므로 정의는 JSON을 통해 왕복될 수 있습니다. Mastra는 Workflow를 등록할 때 각 스키마를 Zod로 변환합니다.

{
  "id": "greeting-workflow",
  "description": "Returns a greeting for the supplied name",
  "inputSchema": {
    "type": "object",
    "properties": { "name": { "type": "string" } },
    "required": ["name"]
  },
  "outputSchema": {
    "type": "object",
    "properties": { "message": { "type": "string" } },
    "required": ["message"]
  },
  "graph": [
    {
      "type": "mapping",
      "id": "create-greeting",
      "mapConfig": "{\"message\":{\"template\":\"Hello, ${initData.name}!\"}}"
    }
  ]
}

그래프 항목

graph의 항목은 순서대로 실행됩니다. 각 항목은 이전 항목의 출력을 받으며 첫 번째 항목은 Workflow 입력을 받습니다.

항목 유형	설명
agent	등록된 Agent 호출
tool	등록된 Tool 호출
mapping	단계 사이에서 데이터 형태 변경
workflow	등록된 Workflow를 중첩 단계로 호출
parallel	여러 단계를 동시에 실행하고 출력을 병합
conditional	조건자가 참인 모든 분기를 동시에 실행
foreach	배열 입력의 각 항목에 대해 단계 하나를 실행
loop	조건자가 성립하는 동안 또는 성립할 때까지 단계를 반복
sleep	고정된 시간 동안 일시 중지
sleepUntil	지정된 날짜까지 일시 중지
.agent()와 .tool()을 사용하는 코드 정의 Workflow는 직렬화할 때 동일한 선언적 항목을 생성합니다.

Agent 단계

agent 항목은 ID로 등록된 Agent를 호출합니다. Agent 단계는 기본적으로 { prompt: string }을 입력으로 받고 { text: string }을 반환합니다.

{
  "type": "agent",
  "id": "summarize",
  "agentId": "support-agent"
}

id는 Workflow 내에서 이 호출 위치를 식별합니다. 이후 단계에서는 Agent 자체 ID와 관계없이 결과를 stepResults.summarize로 참조합니다. Agent에 구조화된 출력을 요청하려면 outputSchema를 추가하세요.

{
  "type": "agent",
  "id": "extract-subtopics",
  "agentId": "support-agent",
  "outputSchema": {
    "type": "array",
    "items": {
      "type": "object",
      "properties": { "title": { "type": "string" } },
      "required": ["title"]
    }
  }
}

Agent 앞에 mapping 항목을 사용하여 Workflow 데이터로 { prompt } 입력을 구성합니다. Agent 항목에는 선택적 description과 options 객체를 지정할 수 있습니다.

{
  "type": "agent",
  "id": "summarize",
  "agentId": "support-agent",
  "description": "Summarize the incoming request",
  "options": { "retries": 2, "metadata": { "team": "support" } }
}

retries와 metadata만 유지됩니다. 코드로 정의한 Workflow를 저장할 때 onFinish와 같이 함수 값을 갖는 옵션 및 함수 값을 갖는 toolChoice는 거부됩니다. 그 밖의 Agent 호출 옵션은 유지되지 않습니다.

Tool 단계

tool 항목은 Mastra의 tools 객체에 등록된 키로 Tool을 호출합니다. Mastra는 Workflow를 등록할 때 레지스트리에서 Tool의 입력 및 출력 스키마를 확인합니다.

{
  "type": "tool",
  "id": "lookup",
  "toolId": "lookup-customer"
}

Tool 항목에는 Agent 항목과 동일한 선택적 description 및 options 필드를 지정할 수 있습니다. retries와 metadata만 유지됩니다.

매핑 단계

mapping 항목은 데이터 형태를 변경합니다. 이 항목의 mapConfig는 객체를 인코딩한 JSON 문자열입니다. 각 키는 단계 출력의 키가 되며 각 설명자는 하나의 소스를 정의합니다.

설명자	설명
{ "value": ... }	상수 JSON 값
{ "template": "..." }	${...} 자리 표시자로 구성된 문자열
{ "initData": true, "path": "a.b" }	Workflow 입력의 값
{ "step": "step-id", "path": "a.b" }	앞선 단계 출력의 값
{ "requestContextPath": "a.b" }	요청 컨텍스트의 값
step 소스에는 단계 ID 배열도 지정할 수 있습니다.

{ "step": ["escalate", "auto-reply"], "path": "text" }

비어 있지 않은 결과가 있는 첫 번째 나열된 단계가 값을 제공합니다. 이것은 다음 이후에 실행된 분기를 선택할 수 있습니다.conditional entry.

템플릿은 initData, inputData, state, requestContext, stepResults.<step-id>에 대한 자리 표시자를 해석합니다.

{
  "type": "mapping",
  "id": "build-prompt",
  "mapConfig": "{\"prompt\":{\"template\":\"Summarize this request: ${initData.request}\"}}"
}

템플릿에서 해석된 객체와 배열은 JSON 문자열로 변환됩니다. 존재하는 결과 내부의 null 값은 빈 문자열로 렌더링됩니다. 성공한 출력이 없는 단계를 참조하는 템플릿은 실행을 실패하게 합니다. 매핑 항목은 최상위 그래프 항목이어야 합니다. parallel, conditional, foreach, loop 컨테이너 내부에는 배치할 수 없습니다.

중첩된 Workflow 단계

workflow 항목은 등록된 다른 Workflow를 호출합니다. 대상은 코드로 정의되거나 저장된 Workflow일 수 있습니다.

{
  "type": "workflow",
  "id": "lookup-first",
  "workflowId": "lookup-customer-workflow"
}

id는 호출 위치를 식별합니다. 동일한 중첩 Workflow가 서로 다른 호출 위치 ID로 여러 번 나타날 수 있으며, 이후 단계에서는 각 결과를 stepResults.<id>로 참조합니다. workflow 항목에는 선택적 description도 지정할 수 있습니다.

병렬 항목

parallel 항목은 여러 단일 단계를 동시에 실행하고 단계 ID를 키로 사용하는 객체에 출력을 병합합니다.

{
  "type": "parallel",
  "steps": [
    { "type": "tool", "id": "first", "toolId": "lookup-customer" },
    { "type": "tool", "id": "second", "toolId": "lookup-customer" }
  ]
}

각 하위 항목은 agent, tool, workflow 항목 중 하나여야 합니다. 모든 하위 항목은 parallel 항목의 입력을 직접 받습니다.

조건부 항목

conditional 항목은 각 단계를 선언적 조건자와 연결하고 조건자가 참인 모든 분기를 실행합니다.

{
  "type": "conditional",
  "steps": [
    { "type": "agent", "id": "escalate", "agentId": "support-agent" },
    { "type": "agent", "id": "auto-reply", "agentId": "support-agent" }
  ],
  "predicates": [
    { "op": "eq", "left": { "path": "inputData.priority" }, "right": { "literal": "urgent" } },
    { "op": "ne", "left": { "path": "inputData.priority" }, "right": { "literal": "urgent" } }
  ]
}

각 하위 항목은 agent, tool, workflow 항목 중 하나여야 하며 각 하위 항목에는 조건자가 필요합니다. 모든 하위 항목은 conditional 항목의 입력을 직접 받습니다.

술어

conditional 항목과 루프는 JSON 조건자 DSL을 사용합니다. 피연산자는 { "path": "..." } 참조 또는 { "literal": ... } 값입니다. 경로는 initData, inputData, stepResults, state를 기준으로 해석됩니다.

운영자	모양
eq, ne, lt, lte, gt, gte	{ "op": "eq", "left": ..., "right": ... }
in, notIn	{ "op": "in", "value": ..., "set": [...] }
exists, notExists	{ "op": "exists", "path": "..." }
truthy, falsy	{ "op": "truthy", "value": ... }
and, or	{ "op": "and", "args": [...] }
not	{ "op": "not", "arg": ... }

누락된 경로는 오류를 발생시키지 않습니다. 경로를 사용하는 연산자는 경로를 해석할 수 없으면 false를 반환합니다. 누락된 값과 falsy 값을 구분하려면 exists 또는 notExists를 사용하세요.

Foreach 항목

foreach 항목은 배열 입력의 각 항목에 대해 본문을 한 번씩 실행합니다. 앞선 항목은 원시 배열을 생성해야 합니다. 결과는 입력 순서를 유지하며 동시 실행 수의 기본값은 1입니다.

{
  "type": "foreach",
  "step": { "type": "workflow", "id": "write-blurb", "workflowId": "blurb-workflow" },
  "opts": { "concurrency": 3 }
}

본문은 agent, tool, workflow 항목 중 하나일 수 있지만 mapping 항목일 수는 없습니다.

루프 항목

loop는 조건자가 성립하는 동안(dowhile) 또는 성립할 때까지(dountil) 단계 하나를 반복합니다.

{
  "type": "loop",
  "loopType": "dountil",
  "step": { "type": "tool", "id": "poll", "toolId": "check-status" },
  "predicate": {
    "op": "eq",
    "left": { "path": "inputData.status" },
    "right": { "literal": "done" }
  }
}

루프 본문은 단일 단계여야 하며 저장된 루프에는 선언적 조건자가 필요합니다.

수면 항목

sleep 항목은 고정된 밀리초 동안 일시 중지합니다. sleepUntil 항목은 ISO 날짜 문자열로 나타낸 지정 날짜까지 일시 중지합니다. 저장된 정의에는 리터럴 값이 필요합니다.

{ "type": "sleep", "id": "wait", "duration": 5000 }

{ "type": "sleepUntil", "id": "wait-for-launch", "date": "2027-01-01T00:00:00.000Z" }

런타임 시 기간이나 날짜를 계산해야 하는 경우 코드 정의 Workflow를 사용합니다.

확인

Mastra는 정의를 유지하거나 등록하기 전에 정의를 검증합니다.

• 구조: 최상위에서만 허용되는 매핑과 같은 배치 규칙을 포함한 항목 형태 및 필수 필드입니다.
• 참조: 각 agentId와 workflowId는 라이브 레지스트리 또는 동일한 번들에서 확인할 수 있어야 합니다. toolId는 Tool 등록 키와 일치해야 합니다.
• 스키마 흐름: 각 항목의 입력은 추론된 매핑 출력을 포함하여 이전 출력과 호환되어야 합니다. 유효성 검사 오류에는 잘못된 항목을 식별하는 graph.2.steps.0과 같은 점 표기법 경로가 포함됩니다.

관련된

• 동적 Workflow 사용
• Mastra.addDynamicWorkflow()
• Mastra.addDynamicWorkflows()
• 클라이언트 SDK Workflow API