OpenFang 자동화 워크플로우 구축 가이드

7분 읽기

단일 에이전트의 역량은 한계가 있지만, 여러 에이전트를 오케스트레이션하면 진정한 엔드투엔드 자동화를 실현할 수 있습니다. 본 가이드에서는 4가지 핵심 사례(일일 산업 브리핑, 경쟁사 모니터링, 콘텐츠 제작 파이프라인, 고객 지원 분류)를 통해 OpenFang 워크플로우 엔진의 모든 기능을 마스터하는 방법을 안내합니다.

OpenFang 자동화 워크플로우 구축 가이드: 단일 에이전트에서 에이전트 팀으로

Slug: openfang-workflow-automation-guide 분류: usage-guides (사용 가이드) 목표 키워드: OpenFang 워크플로우, OpenFang 자동화 프로세스, OpenFang Multi-Agent 오케스트레이션 검색 의도: 정보형 (복잡한 에이전트 자동화 프로세스 구축 희망) 목표 분량: ~2000자 언어: ko


왜 워크플로우가 필요한가요?

단일 에이전트가 수행할 수 있는 작업은 제한적입니다. "주제 조사 → 기사 작성 → 사실 검증 → 게시 → 피드백 모니터링"과 같은 엔드투엔드 프로세스가 필요할 때는 여러 에이전트를 워크플로우로 엮어야 합니다.

OpenFang 워크플로우 엔진은 직렬 파이프라인(Pipeline), 병렬 분산(Fan-out), 조건부 분기(Conditional) 등 세 가지 오케스트레이션 모드를 지원합니다. 레고 블록을 조립하듯 Hands와 Agent를 조합하여 복잡한 자동화 프로세스를 구성할 수 있습니다.

본 가이드에서는 4가지 실전 사례를 통해 간단한 단계부터 복잡한 단계까지 OpenFang 워크플로우 구축 방법을 익혀보겠습니다.

워크플로우 핵심 개념

사례를 살펴보기 전에 핵심 개념을 먼저 이해해 보세요:

개념설명
Task워크플로우의 최소 실행 단위, 하나의 Task는 하나의 Hand 또는 Agent에 대응
Pipeline여러 Task를 직렬로 연결, 이전 Task의 출력이 다음 Task의 입력으로 자동 전달
Fan-out작업을 여러 에이전트로 분산하여 병렬 처리한 후 결과를 취합
Trigger워크플로우를 실행하는 이벤트 소스: 스케줄링, Webhook, 파일 변경, 메시지 명령어
Context워크플로우 단계 간에 전달되는 공유 상태, 프로그래밍 언어의 변수 스코프와 유사

사례 1: 일일 산업 브리핑 (단순 파이프라인)

매일 아침 자동으로 산업 브리핑을 생성하는 가장 기본적인 파이프라인 워크플로우입니다:

toml
[[workflows]]name = "daily_industry_briefing"description = "매일 아침 8시, 어제자 산업 동향 브리핑 생성"[workflows.trigger]type = "schedule"cron = "0 8 * * 1-5"             # 평일 아침 8시[[workflows.tasks]]id = "monitor"hand = "collector"config = { keywords = ["AI agent", "Rust framework"], sources = ["news", "github"] }timeout = 300[[workflows.tasks]]id = "summarize"hand = "researcher"config = { depth = "shallow", focus = "summary" }depends_on = ["monitor"]timeout = 600[[workflows.tasks]]id = "notify"channel = "slack"config = { channel = "#industry-news", format = "rich_text" }depends_on = ["summarize"][workflows.output]format = "slack_message"save_to = "workspace/briefings/{{date}}.md"

워크플로우 실행 순서: Collector 수집 → Researcher 요약 → Slack 알림. depends_on을 통해 Task 간의 의존 관계를 정의합니다.

사례 2: 경쟁사 모니터링 및 경고 (병렬 + 취합)

여러 경쟁사의 동향을 동시에 모니터링하고, 병렬로 수집된 결과를 취합하여 분석하는 워크플로우입니다:

toml
[[workflows]]name = "competitor_monitor"description = "5개 경쟁사 사이트 병렬 모니터링 및 변경 사항 알림"[workflows.trigger]type = "schedule"cron = "0 */6 * * *"# Stage 1: 병렬 수집 (Fan-out)[[workflows.tasks]]id = "check_competitor_a"hand = "collector"config = { target = "competitor-a.com", mode = "diff" }[[workflows.tasks]]id = "check_competitor_b"hand = "collector"config = { target = "competitor-b.com", mode = "diff" }[[workflows.tasks]]id = "check_competitor_c"hand = "collector"config = { target = "competitor-c.com", mode = "diff" }# Stage 2: 취합 및 분석 (모든 수집 완료 대기)[[workflows.tasks]]id = "aggregate"hand = "researcher"config = { action = "merge_and_prioritize" }depends_on = ["check_competitor_a", "check_competitor_b", "check_competitor_c"]# Stage 3: 조건 판단 — 중요 변경 사항이 있을 때만 알림[[workflows.tasks]]id = "check_significance"hand = "predictor"config = { action = "assess_impact" }depends_on = ["aggregate"]# Stage 4: 조건부 알림[[workflows.tasks]]id = "alert_high"channel = "telegram"config = { chat_id = "-100xxx", priority = "high" }depends_on = ["check_significance"]condition = "ctx.significance_score > 0.7"[[workflows.tasks]]id = "alert_low"channel = "email"config = { to = "[email protected]", priority = "low" }depends_on = ["check_significance"]condition = "ctx.significance_score <= 0.7"

핵심 패턴:

  • Fan-out: Stage 1의 5개 수집 Task는 depends_on 없이 병렬로 실행됩니다.
  • Barrier: Stage 2의 Aggregate는 5개 수집 Task가 모두 완료될 때까지 대기합니다.
  • Conditional: Stage 4는 중요도 점수에 따라 알림 채널을 다르게 선택합니다.

사례 3: 콘텐츠 제작 파이프라인 (멀티 에이전트 협업)

AI 작성, AI 검토, AI 이미지 생성이 포함된 전형적인 멀티 에이전트 협업 시나리오입니다:

toml
[[workflows]]name = "content_pipeline"description = "주제 선정부터 게시까지 완전 자동화된 콘텐츠 제작 파이프라인"[workflows.trigger]type = "webhook"endpoint = "/webhook/content/new"secret = "${WEBHOOK_SECRET}"# Stage 1: 조사[[workflows.tasks]]id = "research"hand = "researcher"config = {  depth = "deep",  source_types = ["academic", "news", "social"],  target_length = "3000_words"}timeout = 900# Stage 2: 작성 (다른 모델 사용)[[workflows.tasks]]id = "draft"hand = "researcher"              # Researcher Hand 재사용, 다른 프롬프트 적용config = {  model = "claude-opus-4-8",  instruction = "draft_article",  style = "professional_yet_approachable"}depends_on = ["research"]# Stage 3: 사실 검증[[workflows.tasks]]id = "fact_check"hand = "researcher"config = {  model = "claude-haiku-4-5",   # 빠른 모델로 검증  action = "verify_claims",  strictness = "high"}depends_on = ["draft"]# Stage 4: 이미지 생성[[workflows.tasks]]id = "generate_cover"hand = "clip"config = {  action = "generate_cover_image",  style = "modern_tech",  resolution = "1280x720"}depends_on = ["draft"]           # 사실 검증에 의존하지 않으므로 병렬 실행 가능# Stage 5: SEO 최적화[[workflows.tasks]]id = "seo_optimize"hand = "researcher"config = { action = "seo_optimize", target_keyword = "{{.keyword}}" }depends_on = ["fact_check"]# Stage 6: 게시[[workflows.tasks]]id = "publish"channel = "webhook"config = { url = "{{.cms_endpoint}}", method = "POST" }depends_on = ["seo_optimize", "generate_cover"]# Stage 7: 홍보[[workflows.tasks]]id = "promote"channel = "twitter"config = { action = "post_thread", hashtags = ["#AI", "#Automation"] }depends_on = ["publish"]

Stage 4의 generate_coverfact_check를 기다리지 않고 draft만 완료되면 실행되므로, 전체 대기 시간을 줄일 수 있습니다.

사례 4: 고객 지원 분류 및 에스컬레이션 (조건부 라우팅)

문의 유형에 따라 처리 프로세스를 자동으로 라우팅하는 지능형 고객 지원 워크플로우입니다:

toml
[[workflows]]name = "support_triage"description = "지능형 고객 지원: 분류 → 라우팅 → 처리 → 에스컬레이션"[workflows.trigger]type = "message"channel = "whatsapp"pattern = "/help *"# 분류[[workflows.tasks]]id = "classify"hand = "researcher"config = {  model = "claude-haiku-4-5",  action = "classify_intent",  categories = ["billing", "technical", "feature_request", "complaint"]}# 결제 관련 → 시스템 조회[[workflows.tasks]]id = "handle_billing"channel = "webhook"config = { url = "https://billing-api.company.com/lookup" }depends_on = ["classify"]condition = "ctx.intent == 'billing'"# 기술 지원 → 지식 베이스 검색[[workflows.tasks]]id = "search_kb"hand = "researcher"config = { action = "search_knowledge_base", max_results = 3 }depends_on = ["classify"]condition = "ctx.intent == 'technical'"# 기능 요청 → Product Board 기록[[workflows.tasks]]id = "log_feature"channel = "webhook"config = { url = "https://productboard-api.company.com/notes" }depends_on = ["classify"]condition = "ctx.intent == 'feature_request'"# 불만 사항 → 즉시 담당자 에스컬레이션[[workflows.tasks]]id = "escalate"channel = "slack"config = { channel = "#urgent-support", mention = "@oncall" }depends_on = ["classify"]condition = "ctx.intent == 'complaint'"# 최종 응답[[workflows.tasks]]id = "reply"channel = "whatsapp"config = { reply_to = "{{.original_message.id}}" }depends_on = ["handle_billing", "search_kb", "log_feature", "escalate"]

워크플로우 디버깅 및 테스트

운영 환경에 배포하기 전, Dry Run 모드를 사용하여 워크플로우를 테스트하세요:

bash
# Dry run (실제 외부 작업은 수행하지 않음)openfang workflow run competitor_monitor --dry-run# 특정 Task까지만 실행openfang workflow run content_pipeline --until draft# 워크플로우 실행 기록 확인openfang workflow history competitor_monitor --limit 10# 특정 실행의 상세 로그 확인openfang workflow inspect competitor_monitor --run-id abc123

워크플로우 모범 사례

  1. 적절한 타임아웃 설정: 각 Task에 timeout을 설정하여 특정 단계가 전체 프로세스를 멈추지 않도록 하세요.
  2. 재시도 메커니즘 추가: 네트워크 요청 Task에는 retry = 3과 지수 백오프(Exponential Backoff)를 설정하세요.
  3. 조건부 실행 활용: 불필요한 Task 실행을 방지하여 API 호출 비용을 절감하세요.
  4. 실행 로그 기록: 문제 해결을 위해 log_level = "debug"를 설정하세요.
  5. 워크플로우 상태 모니터링: Prometheus/Grafana와 연동하여 워크플로우 지표를 모니터링하세요.

자주 묻는 질문

하나의 워크플로우에 최대 몇 개의 Task를 포함할 수 있나요?
하드 제한은 없습니다. 하지만 유지보수성을 위해 하나의 워크플로우당 Task 수를 20개 이내로 관리하는 것을 권장합니다. 더 복잡한 프로세스는 하위 워크플로우로 분리하세요.
Task 실행 실패 시 어떻게 처리하나요?
각 Task마다 독립적인 실패 전략을 설정할 수 있습니다:

``toml
[workflows.tasks.fallback]
on_failure = "continue" # continue / retry / abort / escalate
retry_count = 3
retry_delay_seconds = 60
fallback_task = "manual_review" # 실패 시 실행할 대체 Task
``

워크플로우끼리 서로 호출할 수 있나요?
네, 가능합니다. trigger.type = "workflow"를 사용하면 한 워크플로우가 다른 워크플로우를 트리거할 수 있습니다. 조건부 실행과 결합하면 복잡한 중첩 로직을 구현할 수 있습니다.
워크플로우 내에서 동적 파라미터를 어떻게 전달하나요?
템플릿 문법 {{.variable_name}}을 사용하세요. 컨텍스트 변수는 트리거 데이터, 이전 Task의 출력, 전역 설정 변수 등에서 가져올 수 있습니다.

다음 단계