[이제와서 시작하는 Claude AI 마스터하기 #18] 커스텀 워크플로우 구축

복잡한 비즈니스 프로세스를 AI로 자동화

커스텀 워크플로우를 구축하면 Claude의 능력을 최대한 활용해 복잡한 비즈니스 프로세스를 자동화할 수 있습니다. 이번 편에서는 실제 기업 환경에서 활용 가능한 고급 워크플로우 구축 방법을 다룹니다.

워크플로우 아키텍처 설계

1. 워크플로우 엔진 구현

// 확장 가능한 워크플로우 엔진
interface WorkflowEngine {
  define(workflow: WorkflowDefinition): void;
  execute(workflowId: string, input: any): Promise<WorkflowResult>;
  monitor(executionId: string): ExecutionStatus;
  retry(executionId: string): Promise<WorkflowResult>;
}

class AdvancedWorkflowEngine implements WorkflowEngine {
  private workflows: Map<string, WorkflowDefinition> = new Map();
  private executions: Map<string, WorkflowExecution> = new Map();
  private eventBus: EventEmitter;
  
  define(workflow: WorkflowDefinition) {
    // 워크플로우 검증
    this.validateWorkflow(workflow);
    
    // DAG(Directed Acyclic Graph) 구성
    const dag = this.buildDAG(workflow.steps);
    
    this.workflows.set(workflow.id, {
      ...workflow,
      dag,
      compiledConditions: this.compileConditions(workflow.steps)
    });
  }
  
  async execute(workflowId: string, input: any): Promise<WorkflowResult> {
    const workflow = this.workflows.get(workflowId);
    if (!workflow) throw new Error(`Workflow ${workflowId} not found`);
    
    const execution = new WorkflowExecution(workflow, input);
    this.executions.set(execution.id, execution);
    
    try {
      // 실행 시작 이벤트
      this.eventBus.emit('workflow.started', {
        workflowId,
        executionId: execution.id,
        input
      });
      
      // 병렬 실행 지원
      const result = await this.executeDAG(execution);
      
      // 완료 이벤트
      this.eventBus.emit('workflow.completed', {
        workflowId,
        executionId: execution.id,
        result
      });
      
      return result;
    } catch (error) {
      // 실패 처리
      await this.handleFailure(execution, error);
      throw error;
    }
  }
  
  private async executeDAG(execution: WorkflowExecution) {
    const { dag } = execution.workflow;
    const results = new Map();
    
    // 토폴로지 정렬로 실행 순서 결정
    const executionOrder = this.topologicalSort(dag);
    
    for (const layer of executionOrder) {
      // 같은 레이어의 스텝들은 병렬 실행
      const layerResults = await Promise.all(
        layer.map(step => this.executeStep(step, execution, results))
      );
      
      // 결과 저장
      layer.forEach((step, index) => {
        results.set(step.id, layerResults[index]);
      });
    }
    
    return this.aggregateResults(results, execution.workflow);
  }
}

2. 복잡한 워크플로우 정의

# 복잡한 비즈니스 워크플로우 예시
name: "고객 온보딩 자동화"
id: "customer-onboarding-v2"
version: "2.0.0"

triggers:
  - type: "webhook"
    endpoint: "/api/webhooks/new-customer"
  - type: "schedule"
    cron: "0 9 * * MON"
  - type: "event"
    source: "crm.customer.created"

inputs:
  customer:
    type: "object"
    required: ["email", "company", "plan"]
    
steps:
  # 병렬 실행 그룹 1
  - parallel:
    - id: "verify-email"
      type: "http"
      config:
        url: "https://api.emailverify.com/check"
        method: "POST"
        body:
          email: ""
      
    - id: "company-research"
      type: "claude"
      config:
        prompt: |
          Research company: 
          Find: industry, size, recent news, competitors
        model: "claude-3-opus-20240229"
        
    - id: "risk-assessment"
      type: "function"
      config:
        handler: "assessCustomerRisk"
        params:
          customer: ""

  # 조건부 분기
  - id: "risk-check"
    type: "condition"
    config:
      if: " > 0.7"
      then:
        - id: "manual-review"
          type: "human-task"
          config:
            assignee: "compliance-team"
            deadline: "24h"
      else:
        - id: "auto-approve"
          type: "function"
          config:
            handler: "approveCustomer"

  # 병렬 실행 그룹 2
  - parallel:
    - id: "create-account"
      type: "api"
      config:
        service: "account-service"
        method: "createAccount"
        
    - id: "setup-billing"
      type: "api"
      config:
        service: "billing-service"
        method: "setupSubscription"
        
    - id: "generate-welcome-kit"
      type: "claude"
      config:
        prompt: |
          Create personalized welcome materials for:
          Company: 
          Industry: 
          Plan: 
          
          Include:
          1. Welcome email
          2. Getting started guide
          3. Best practices for their industry

  # 최종 단계
  - id: "send-communications"
    type: "multi-channel"
    config:
      channels:
        - email:
            to: ""
            template: "welcome"
            data: ""
        - slack:
            channel: "#new-customers"
            message: "New customer onboarded: "
        - crm:
            action: "updateStatus"
            status: "onboarded"

error_handling:
  retry:
    max_attempts: 3
    backoff: "exponential"
    
  fallback:
    - type: "notification"
      target: "ops-team"
    - type: "queue"
      queue: "failed-onboarding"
      
monitoring:
  metrics:
    - "execution_time"
    - "success_rate"
    - "step_duration"
  alerts:
    - condition: "execution_time > 300s"
      channel: "slack"

3. 동적 워크플로우 생성

// AI가 워크플로우를 생성하는 시스템
class DynamicWorkflowBuilder {
  private claude: ClaudeAPIClient;
  
  async buildWorkflowFromRequirements(
    requirements: string,
    constraints: WorkflowConstraints
  ): Promise<WorkflowDefinition> {
    // Claude에게 워크플로우 설계 요청
    const design = await this.claude.createMessage({
      system: `You are a workflow architect. Design a workflow based on requirements.
Consider: error handling, parallel execution, conditional logic, integrations.
Output as structured JSON matching WorkflowDefinition schema.`,
      messages: [{
        role: 'user',
        content: `Requirements: ${requirements}\nConstraints: ${JSON.stringify(constraints)}`
      }],
      temperature: 0.7
    });
    
    const workflowDef = JSON.parse(design.content[0].text);
    
    // 검증 및 최적화
    await this.validateAndOptimize(workflowDef);
    
    return workflowDef;
  }
  
  async optimizeExistingWorkflow(
    workflow: WorkflowDefinition,
    executionHistory: ExecutionHistory[]
  ): Promise<OptimizedWorkflow> {
    const analysis = await this.analyzePerformance(executionHistory);
    
    const optimization = await this.claude.createMessage({
      system: 'Analyze workflow performance and suggest optimizations.',
      messages: [{
        role: 'user',
        content: `
Workflow: ${JSON.stringify(workflow)}
Performance data: ${JSON.stringify(analysis)}

Suggest:
1. Steps that can be parallelized
2. Unnecessary steps to remove
3. Caching opportunities
4. Error handling improvements
`
      }]
    });
    
    return this.applyOptimizations(workflow, optimization.content[0].text);
  }
}

실전 워크플로우 예제

1. 콘텐츠 퍼블리싱 파이프라인

// 멀티채널 콘텐츠 퍼블리싱 워크플로우
class ContentPublishingWorkflow {
  private workflow = {
    id: 'content-publishing',
    name: 'Multi-channel Content Publishing',
    steps: [
      {
        id: 'content-analysis',
        type: 'claude',
        action: async (input: { content: string; metadata: any }) => {
          return await this.analyzeContent(input);
        }
      },
      {
        id: 'adaptation',
        type: 'parallel',
        steps: [
          {
            id: 'adapt-twitter',
            action: async (content: AnalyzedContent) => {
              return await this.adaptForTwitter(content);
            }
          },
          {
            id: 'adapt-linkedin',
            action: async (content: AnalyzedContent) => {
              return await this.adaptForLinkedIn(content);
            }
          },
          {
            id: 'adapt-blog',
            action: async (content: AnalyzedContent) => {
              return await this.adaptForBlog(content);
            }
          }
        ]
      },
      {
        id: 'quality-check',
        type: 'claude',
        action: async (adaptations: Adaptations) => {
          return await this.qualityCheck(adaptations);
        }
      },
      {
        id: 'schedule-publishing',
        type: 'function',
        action: async (approved: ApprovedContent) => {
          return await this.schedulePublishing(approved);
        }
      }
    ]
  };
  
  private async analyzeContent(input: { content: string; metadata: any }) {
    const analysis = await claude.createMessage({
      system: `Analyze content for multi-channel publishing:
- Extract key messages
- Identify target audience
- Determine tone and style
- Find quotable snippets
- Suggest hashtags and keywords`,
      messages: [{
        role: 'user',
        content: input.content
      }]
    });
    
    return JSON.parse(analysis.content[0].text);
  }
  
  private async adaptForTwitter(content: AnalyzedContent) {
    const threads = await claude.createMessage({
      system: `Create Twitter thread:
- First tweet: Hook (max 280 chars)
- 3-5 follow-up tweets with key points
- Final tweet: CTA with link
- Include relevant hashtags
- Optimize for engagement`,
      messages: [{
        role: 'user',
        content: JSON.stringify(content)
      }]
    });
    
    return {
      platform: 'twitter',
      content: JSON.parse(threads.content[0].text),
      scheduledTime: this.calculateOptimalTime('twitter', content.audience)
    };
  }
  
  private async schedulePublishing(approved: ApprovedContent) {
    const schedule = [];
    
    for (const platform of approved.platforms) {
      const job = await this.scheduler.schedule({
        platform: platform.name,
        content: platform.content,
        time: platform.scheduledTime,
        retryPolicy: {
          attempts: 3,
          backoff: 'exponential'
        }
      });
      
      schedule.push(job);
    }
    
    return {
      scheduled: schedule,
      dashboard: `https://dashboard.example.com/campaigns/${approved.campaignId}`
    };
  }
}

2. 인시던트 대응 자동화

// AI 기반 인시던트 대응 워크플로우
class IncidentResponseWorkflow {
  async handleIncident(alert: Alert): Promise<IncidentResolution> {
    const workflow = new WorkflowExecution('incident-response');
    
    // 1단계: 초기 분석
    const analysis = await workflow.execute('initial-analysis', {
      alert,
      logs: await this.fetchRecentLogs(alert.service),
      metrics: await this.fetchMetrics(alert.service, '15m')
    });
    
    // 2단계: 심각도 평가
    const severity = await workflow.execute('assess-severity', analysis);
    
    // 3단계: 병렬 대응
    const responses = await workflow.parallel([
      {
        id: 'notify-team',
        condition: severity.level >= 3,
        action: async () => {
          return await this.notifyOncall(severity);
        }
      },
      {
        id: 'auto-remediation',
        condition: severity.autoRemediable,
        action: async () => {
          return await this.attemptAutoRemediation(analysis);
        }
      },
      {
        id: 'gather-context',
        action: async () => {
          return await this.gatherAdditionalContext(alert);
        }
      }
    ]);
    
    // 4단계: 지능형 진단
    const diagnosis = await this.aiDiagnosis(analysis, responses);
    
    // 5단계: 실행 계획
    const plan = await this.createActionPlan(diagnosis);
    
    // 6단계: 실행 및 모니터링
    return await this.executeAndMonitor(plan);
  }
  
  private async aiDiagnosis(
    analysis: InitialAnalysis,
    responses: ParallelResponses
  ) {
    const prompt = `
Diagnose the incident based on:

Alert: ${JSON.stringify(analysis.alert)}
Error patterns: ${JSON.stringify(analysis.errorPatterns)}
Recent changes: ${JSON.stringify(responses.context.recentDeployments)}
Similar incidents: ${JSON.stringify(responses.context.historicalIncidents)}

Provide:
1. Root cause analysis
2. Impact assessment
3. Recommended actions
4. Preventive measures
`;
    
    const diagnosis = await claude.createMessage({
      model: 'claude-3-opus-20240229',
      messages: [{ role: 'user', content: prompt }],
      temperature: 0.3
    });
    
    return JSON.parse(diagnosis.content[0].text);
  }
  
  private async createActionPlan(diagnosis: Diagnosis) {
    // 동적 플레이북 생성
    const playbook = await this.generatePlaybook(diagnosis);
    
    return {
      immediate: playbook.immediate.map(action => ({
        ...action,
        executor: this.assignExecutor(action),
        timeout: this.calculateTimeout(action),
        rollback: this.defineRollback(action)
      })),
      followUp: playbook.followUp,
      monitoring: {
        metrics: playbook.monitoringMetrics,
        duration: playbook.monitoringDuration,
        successCriteria: playbook.successCriteria
      }
    };
  }
}

3. 데이터 파이프라인 오케스트레이션

// 복잡한 데이터 처리 워크플로우
class DataPipelineOrchestrator {
  async runETLPipeline(config: PipelineConfig): Promise<PipelineResult> {
    const pipeline = new DistributedWorkflow('etl-pipeline');
    
    // 추출 단계 (병렬)
    const extractionTasks = config.sources.map(source => ({
      id: `extract-${source.name}`,
      type: 'extract',
      config: {
        source,
        partitioning: this.calculatePartitions(source),
        validation: this.getValidationRules(source)
      }
    }));
    
    const extracted = await pipeline.parallel(extractionTasks);
    
    // 변환 단계 (AI 지원)
    const transformed = await pipeline.execute('transform', {
      data: extracted,
      transformations: await this.generateTransformations(extracted)
    });
    
    // 품질 검증
    const quality = await pipeline.execute('quality-check', {
      data: transformed,
      rules: config.qualityRules,
      aiValidation: true
    });
    
    // 조건부 처리
    if (quality.score < config.qualityThreshold) {
      await pipeline.execute('remediation', {
        issues: quality.issues,
        data: transformed
      });
    }
    
    // 로드 단계
    const loaded = await pipeline.execute('load', {
      data: quality.validData,
      destinations: config.destinations,
      strategy: 'incremental'
    });
    
    return {
      summary: loaded,
      lineage: pipeline.getLineage(),
      metrics: pipeline.getMetrics()
    };
  }
  
  private async generateTransformations(data: ExtractedData[]) {
    // Claude를 사용해 데이터 변환 로직 생성
    const sample = this.getSample(data);
    
    const transformations = await claude.createMessage({
      system: `Generate data transformation logic:
- Analyze data structure and quality
- Suggest normalization steps
- Create mapping rules
- Define aggregations
Output as executable transformation specs.`,
      messages: [{
        role: 'user',
        content: `Data sample: ${JSON.stringify(sample)}\nTarget schema: ${this.targetSchema}`
      }]
    });
    
    return this.compileTransformations(transformations.content[0].text);
  }
}

고급 기능

1. 워크플로우 버전 관리

class WorkflowVersionControl {
  private versions: Map<string, WorkflowVersion[]> = new Map();
  
  async deployNewVersion(
    workflowId: string,
    newVersion: WorkflowDefinition,
    strategy: DeploymentStrategy
  ) {
    const currentVersion = await this.getCurrentVersion(workflowId);
    
    switch (strategy) {
      case 'canary':
        await this.canaryDeploy(workflowId, currentVersion, newVersion);
        break;
        
      case 'blue-green':
        await this.blueGreenDeploy(workflowId, currentVersion, newVersion);
        break;
        
      case 'rolling':
        await this.rollingDeploy(workflowId, currentVersion, newVersion);
        break;
    }
  }
  
  private async canaryDeploy(
    workflowId: string,
    current: WorkflowVersion,
    next: WorkflowVersion
  ) {
    // 5% 트래픽으로 시작
    await this.router.setWeights(workflowId, {
      [current.version]: 0.95,
      [next.version]: 0.05
    });
    
    // 메트릭 모니터링
    const monitor = new CanaryMonitor(workflowId, next.version);
    
    // 점진적 롤아웃
    for (const percentage of [10, 25, 50, 75, 100]) {
      await this.wait('5m');
      
      const metrics = await monitor.getMetrics();
      if (metrics.errorRate > current.metrics.errorRate * 1.1) {
        await this.rollback(workflowId, current.version);
        throw new Error('Canary deployment failed');
      }
      
      await this.router.setWeights(workflowId, {
        [current.version]: (100 - percentage) / 100,
        [next.version]: percentage / 100
      });
    }
  }
}

2. 워크플로우 최적화 엔진

class WorkflowOptimizer {
  async optimizePerformance(
    workflow: WorkflowDefinition,
    executionHistory: ExecutionRecord[]
  ): Promise<OptimizedWorkflow> {
    // 병목 지점 분석
    const bottlenecks = this.identifyBottlenecks(executionHistory);
    
    // AI 기반 최적화 제안
    const suggestions = await this.getSuggestions(workflow, bottlenecks);
    
    // 최적화 적용
    const optimized = this.applyOptimizations(workflow, suggestions);
    
    // 시뮬레이션
    const simulation = await this.simulate(optimized, executionHistory);
    
    return {
      workflow: optimized,
      expectedImprovement: simulation.improvement,
      changes: suggestions.applied
    };
  }
  
  private identifyBottlenecks(history: ExecutionRecord[]) {
    const stepDurations = new Map<string, number[]>();
    
    // 각 스텝의 실행 시간 수집
    history.forEach(execution => {
      execution.steps.forEach(step => {
        if (!stepDurations.has(step.id)) {
          stepDurations.set(step.id, []);
        }
        stepDurations.get(step.id)!.push(step.duration);
      });
    });
    
    // 통계 분석
    return Array.from(stepDurations.entries())
      .map(([stepId, durations]) => ({
        stepId,
        avgDuration: avg(durations),
        p95Duration: percentile(durations, 95),
        variance: variance(durations)
      }))
      .filter(stat => stat.p95Duration > 5000) // 5초 이상
      .sort((a, b) => b.avgDuration - a.avgDuration);
  }
}

모니터링과 관찰성

워크플로우 대시보드

class WorkflowDashboard {
  getMetrics(): DashboardMetrics {
    return {
      overview: {
        totalExecutions: this.getTotalExecutions(),
        successRate: this.getSuccessRate(),
        avgDuration: this.getAverageDuration(),
        activeWorkflows: this.getActiveWorkflows()
      },
      
      performance: {
        throughput: this.getThroughput(),
        latency: this.getLatencyPercentiles(),
        queueDepth: this.getQueueDepth(),
        workerUtilization: this.getWorkerUtilization()
      },
      
      errors: {
        errorRate: this.getErrorRate(),
        topErrors: this.getTopErrors(),
        failedSteps: this.getFailedSteps()
      },
      
      cost: {
        apiCalls: this.getAPICalls(),
        computeTime: this.getComputeTime(),
        estimatedCost: this.estimateCost()
      }
    };
  }
}

다음 편 예고

다음 편에서는 “팀 협업 환경에서의 활용”을 다룰 예정입니다. Claude를 팀 전체가 효과적으로 활용하는 방법과 협업 도구 구축을 알아보겠습니다.

---

💡 오늘의 과제: 자신의 업무 중 반복적인 프로세스를 하나 선택해 간단한 워크플로우로 자동화해보세요. Claude API를 활용해 지능적인 의사결정을 포함시켜보세요!