Mastra Workflow로 병렬 실행·Human-in-the-Loop·멀티 에이전트를 TypeScript 하나로 연결하는 법

import { createWorkflow } from '@mastra/core/workflows';
 
const contentWorkflow = createWorkflow({ id: 'content-pipeline' })
  .addStep(fetchTopicStep)                                  // 주제 받아오기
  .parallel([researchStep, competitorAnalysisStep])         // 동시 실행
  .addStep(editorialStep)                                   // 두 결과 합산 후 편집
  .commit();

// 레거시 방식: .after()로 fan-out 표현
myWorkflow
  .step(fetchTopicStep)
  .then(researchStep)
  .after(fetchTopicStep)
  .step(competitorAnalysisStep)
  .commit();

import { createStep, createWorkflow } from '@mastra/core/workflows';
import { z } from 'zod';
 
const approvalStep = createStep({
  id: 'approval-step',
  inputSchema: z.object({ content: z.string() }),
  resumeSchema: z.object({
    approved: z.boolean(),
    feedback: z.string().optional(),
  }),
  suspendSchema: z.object({ reason: z.string() }),
 
  async execute({ inputData, resumeData, suspend }) {
    // 첫 실행: 아직 승인 여부가 없으면 일시정지
    if (resumeData?.approved === undefined) {
      await suspend({ reason: '검토자 승인이 필요합니다.' });
      return;
    }
 
    // 재개 후: resumeData에 사람의 판단이 담겨 있음
    if (!resumeData.approved) {
      throw new Error(`거절됨: ${resumeData.feedback ?? '피드백 없음'}`);
    }
 
    return { result: inputData.content };
  },
});

// 예: POST /approve 핸들러에서 호출
await workflow.resume({
  runId: 'run-123',
  stepId: 'approval-step',
  resumeData: { approved: true, feedback: '확인 완료' },
});

import { Agent } from '@mastra/core/agent';
import { anthropic } from '@ai-sdk/anthropic';
 
// 전문 서브에이전트
const researchAgent = new Agent({
  name: 'ResearchAgent',
  instructions: '주어진 주제를 심층 조사하고 핵심 내용을 요약한다.',
  model: anthropic('claude-sonnet-4-6'),
});
 
const writerAgent = new Agent({
  name: 'WriterAgent',
  instructions: '조사 결과를 바탕으로 독자 친화적인 블로그 포스트를 작성한다.',
  model: anthropic('claude-sonnet-4-6'),
});
 
// 슈퍼바이저: agents 배열로 위임 대상 등록
const supervisorAgent = new Agent({
  name: 'SupervisorAgent',
  instructions: '사용자 요청을 분석하여 적절한 전문 에이전트에게 위임한다.',
  model: anthropic('claude-opus-4-7'), // 라우팅 판단은 더 강력한 모델로
  agents: [researchAgent, writerAgent],
});

const result = await supervisorAgent.stream('AI 트렌드 블로그 글 작성해줘', {
  onDelegationStart({ agent, messages }) {
    console.log(`→ ${agent.name}에게 위임 시작`);
    return messages; // 수정 없이 그대로 전달; 수정하려면 변환 후 반환
  },
});

import { Mastra } from '@mastra/core';
import { createStep } from '@mastra/core/workflows';
 
// 에이전트를 Mastra 인스턴스에 등록
const mastra = new Mastra({
  agents: { ResearchAgent: researchAgent },
});
 
const researchStep = createStep({
  id: 'research-step',
  inputSchema: z.object({ query: z.string() }),
  outputSchema: z.object({ result: z.string() }),
 
  async execute({ inputData, mastra }) {
    const agent = mastra.getAgent('ResearchAgent'); // 등록된 에이전트 참조
    const response = await agent.generate(inputData.query);
    return { result: response.text };
  },
});

import { createWorkflow, createStep } from '@mastra/core/workflows';
import { z } from 'zod';
 
// fetchTopicStep: 이 예시에서는 외부에서 주입된다고 가정 (스캐폴딩 혹은 별도 파일)
// 최소 stub 예시:
// const fetchTopicStep = createStep({
//   id: 'fetch-topic', inputSchema: z.object({ keyword: z.string() }),
//   outputSchema: z.object({ topic: z.string() }),
//   async execute({ inputData }) { return { topic: inputData.keyword }; },
// });
 
// 1. 병렬로 실행될 두 스텝
const researchStep = createStep({
  id: 'research',
  inputSchema: z.object({ topic: z.string() }),
  outputSchema: z.object({ summary: z.string() }),
  async execute({ inputData, mastra }) {
    const agent = mastra.getAgent('ResearchAgent');
    const res = await agent.generate(`${inputData.topic}에 대해 조사해줘`);
    return { summary: res.text };
  },
});
 
const competitorStep = createStep({
  id: 'competitor-analysis',
  inputSchema: z.object({ topic: z.string() }),
  outputSchema: z.object({ analysis: z.string() }),
  async execute({ inputData, mastra }) {
    const agent = mastra.getAgent('ResearchAgent');
    const res = await agent.generate(`${inputData.topic} 관련 경쟁사 현황 분석`);
    return { analysis: res.text };
  },
});
 
// assessQuality: 초안 품질을 0~1 사이 점수로 반환하는 함수.
// 실제 구현은 LLM 평가, 규칙 기반 체크(길이, 키워드 밀도 등) 등 프로젝트에 따라 다름.
// 아래는 글자 수 기반 간단한 더미 예시:
function assessQuality(text: string): number {
  if (text.length < 300) return 0.4;
  if (text.length < 800) return 0.6;
  return 0.85;
}
 
// 2. 편집 + 품질 검토(HITL 포함) 스텝
const editorialStep = createStep({
  id: 'editorial',
  inputSchema: z.object({ summary: z.string(), analysis: z.string() }),
  resumeSchema: z.object({ approved: z.boolean(), feedback: z.string().optional() }),
  suspendSchema: z.object({ draft: z.string(), reason: z.string() }),
  async execute({ inputData, resumeData, suspend, mastra }) {
    // 재개 경로: 사람이 판단을 내린 경우
    if (resumeData !== undefined) {
      if (!resumeData.approved) {
        // 이 에러는 워크플로우 전체를 실패 상태로 만듭니다.
        // 재시도 로직이 필요하다면 워크플로우 레벨에서 별도 처리가 필요합니다.
        throw new Error(`편집 거절: ${resumeData.feedback}`);
      }
      return { published: true };
    }
 
    // 초기 실행: 초안 생성
    const agent = mastra.getAgent('WriterAgent');
    const draft = await agent.generate(
      `조사 내용: ${inputData.summary}\n경쟁사 분석: ${inputData.analysis}\n\n이를 바탕으로 블로그 초안을 작성해줘`
    );
 
    const qualityScore = assessQuality(draft.text);
 
    // 품질 기준 미달 시 사람 검토 요청
    if (qualityScore < 0.7) {
      await suspend({ draft: draft.text, reason: `품질 점수 미달: ${qualityScore}` });
      return;
    }
 
    return { published: true, draft: draft.text };
  },
});
 
// 3. 워크플로우 조립
const contentPipelineWorkflow = createWorkflow({ id: 'content-pipeline' })
  .addStep(fetchTopicStep)
  .parallel([researchStep, competitorStep])  // 병렬 실행
  .addStep(editorialStep)                    // 합산 후 편집 + HITL
  .commit();

import { Agent } from '@mastra/core/agent';
import { anthropic } from '@ai-sdk/anthropic';
 
const techSupportAgent = new Agent({
  name: 'TechSupportAgent',
  instructions: '기술적인 문제를 진단하고 해결 방법을 안내한다.',
  model: anthropic('claude-sonnet-4-6'),
});
 
const billingAgent = new Agent({
  name: 'BillingAgent',
  instructions: '결제 관련 문의를 처리한다. 개인정보는 마스킹된 형태로만 다룬다.',
  model: anthropic('claude-sonnet-4-6'),
});
 
const refundAgent = new Agent({
  name: 'RefundAgent',
  instructions: '환불 요청을 검토하고 처리 절차를 안내한다.',
  model: anthropic('claude-sonnet-4-6'),
});
 
const supportSupervisor = new Agent({
  name: 'SupportSupervisor',
  instructions: `고객 문의를 분석하여 적절한 전문 에이전트에게 위임한다.
    - 기술 문제 → TechSupportAgent
    - 결제 문의 → BillingAgent
    - 환불 요청 → RefundAgent`,
  model: anthropic('claude-opus-4-7'), // 라우팅 판단은 강력한 모델이 맡고
  agents: [techSupportAgent, billingAgent, refundAgent],
});
 
// onDelegationStart의 반환값이 서브에이전트에 실제로 전달되는 메시지입니다.
// 결제/환불 에이전트에는 카드번호를 마스킹해서 전달합니다.
const result = await supportSupervisor.stream(customerMessage, {
  onDelegationStart({ agent, messages }) {
    if (['BillingAgent', 'RefundAgent'].includes(agent.name)) {
      return messages.map((msg) => ({
        ...msg,
        content:
          typeof msg.content === 'string'
            ? msg.content.replace(/\d{4}-\d{4}-\d{4}-\d{4}/g, '****-****-****-****')
            : msg.content,
      }));
    }
    return messages;
  },
});