Temporal AI 워크플로우 오케스트레이션 — 신뢰할 수 있는 다단계 AI 파이프라인

# Option A: Docker Compose(local dev 권장)
git clone https://github.com/temporalio/docker-compose.git
cd docker-compose
docker compose up -d

# Option B: Temporal Cloud(managed, infra 불필요)
# cloud.temporal.io에서 가입하고 namespace 생성

# 서버 실행 확인
temporal cluster health

pip install temporalio

import asyncio
from temporalio import worker, workflow, activity
from temporalio.client import Client
from temporalio.common import RetryPolicy

# activity 정의(개별 단계)
@activity.defn
async def load_dataset(dataset_name: str):
    """데이터셋 로드 및 검증."""
    print(f"데이터셋 로드 중: {dataset_name}")
    data = {"samples": 10000, "features": 128}
    activity.info(f"{data['samples']} 샘플 로드 완료")
    return data

@activity.defn
async def preprocess(data: dict):
    """데이터 정리 및 정규화."""
    print("데이터 전처리 중...")
    processed = {
        "cleaned_samples": data["samples"],
        "normalized": True,
        "feature_count": data["features"]
    }
    return processed

@activity.defn
async def train_model(preprocessed_data: dict, epochs: int = 10):
    """전처리 데이터로 모델 훈련."""
    print(f"{epochs} epoch 동안 모델 훈련 중...")
    metrics = {
        "final_loss": 0.0234,
        "final_accuracy": 0.9456,
        "epochs_trained": epochs
    }
    activity.info(f"훈련 완료: accuracy={metrics['final_accuracy']:.4f}")
    return metrics

@activity.defn
async def deploy_model(metrics: dict):
    """훈련된 모델을 프로덕션에 배포."""
    print("모델을 프로덕션에 배포 중...")
    deployment = {
        "model_id": f"model-{metrics['final_accuracy']:.4f}",
        "status": "deployed",
        "endpoint": "https://api.example.com/v1/predict"
    }
    activity.info(f"모델 배포됨: {deployment['model_id']}")
    return deployment

# workflow 정의
@workflow.defn
class MLTrainingPipeline:
    @workflow.run
    async def run(self, dataset_name: str, epochs: int = 10) -> dict:
        # 각 단계는 activity 호출
        data = await workflow.execute_activity(
            load_dataset, dataset_name,
            retry=RetryPolicy(max_attempts=3)
        )
        
        processed = await workflow.execute_activity(
            preprocess, data,
            retry=RetryPolicy(max_attempts=2)
        )
        
        metrics = await workflow.execute_activity(
            train_model, processed, epochs,
            retry=RetryPolicy(max_attempts=3, initial_interval=10)
        )
        
        deployment = await workflow.execute_activity(
            deploy_model, metrics,
            retry=RetryPolicy(max_attempts=2)
        )
        
        return deployment

# worker.py
import asyncio
from temporalio.worker import Worker
from my_workflow import MLTrainingPipeline, load_dataset, preprocess, train_model, deploy_model

async def main():
    worker = Worker(
        client,  # Temporal Client instance
        task_queue="ml-pipeline",
        workflows=[MLTrainingPipeline],
        activities=[load_dataset, preprocess, train_model, deploy_model]
    )
    print("Worker 시작됨. Ctrl+C를 눌러 종료.")
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

from temporalio import workflow, activity

@activity.defn
async def generate_with_gpt4(prompt: str) -> str:
    """먼저 GPT-4 시도."""
    response = await call_openai(prompt, model="gpt-4o")
    return response

@activity.defn
async def generate_with_claude(prompt: str) -> str:
    """Claude로 폴백."""
    response = await call_anthropic(prompt, model="claude-sonnet-4")
    return response

@activity.defn
async def generate_with_local(prompt: str) -> str:
    """마지막 수단: local 모델."""
    response = await call_ollama(prompt, model="llama3.2")
    return response

@workflow.defn
class ResilientLLMChain:
    @workflow.run
    async def run(self, prompt: str) -> dict:
        try:
            result = await workflow.execute_activity(
                generate_with_gpt4, prompt,
                timeout=timedelta(minutes=5),
                retry=RetryPolicy(max_attempts=2)
            )
            model_used = "gpt-4o"
        except Exception:
            try:
                result = await workflow.execute_activity(
                    generate_with_claude, prompt,
                    timeout=timedelta(minutes=5),
                    retry=RetryPolicy(max_attempts=2)
                )
                model_used = "claude-sonnet-4"
            except Exception:
                result = await workflow.execute_activity(
                    generate_with_local, prompt,
                    timeout=timedelta(minutes=10),
                    retry=RetryPolicy(max_attempts=3)
                )
                model_used = "local-llama"
        
        return {"response": result, "model_used": model_used, "fallback_chain": True}

@activity.defn
async def agent_research(query: str) -> dict:
    """연구 agent: 웹에서 정보 수집."""
    results = await search_web(query)
    return {"type": "research", "sources": len(results), "summary": summarize(results)}

@activity.defn
async def agent_analysis(research_data: dict) -> dict:
    """분석 agent: 발견 사항 평가."""
    analysis = await analyze_findings(research_data["summary"])
    return {"type": "analysis", "confidence": analysis["confidence_score"]}

@activity.defn
async def agent_synthesis(research: dict, analysis: dict) -> dict:
    """합성 agent: 연구와 분석을 보고서로 결합."""
    report = await synthesize_report(research, analysis)
    return {"type": "synthesis", "report_length": len(report)}

@workflow.defn
class MultiAgentResearch:
    @workflow.run
    async def run(self, query: str) -> dict:
        research_handle = workflow.execute_activity(
            agent_research, query, start_to_close_timeout=timedelta(minutes=5)
        )
        
        research_result = await research_handle
        analysis_handle = workflow.execute_activity(
            agent_analysis, research_result, start_to_close_timeout=timedelta(minutes=3)
        )
        
        analysis_result = await analysis_handle
        
        final_report = await workflow.execute_activity(
            agent_synthesis, research_result, analysis_result,
            start_to_close_timeout=timedelta(minutes=5)
        )
        
        return final_report

@activity.defn
async def save_checkpoint(epoch: int, model_state: dict) -> str:
    """훈련 체크포인트를 영구 저장소에 저장."""
    checkpoint_path = f"s3://my-bucket/checkpoints/epoch_{epoch}.pt"
    await upload_to_s3(model_state, checkpoint_path)
    activity.info(f"체크포인트 저장됨: {checkpoint_path}")
    return checkpoint_path

@activity.defn
async def load_checkpoint(checkpoint_path: str) -> dict:
    """체크포인트에서 모델 상태 로드."""
    model_state = await download_from_s3(checkpoint_path)
    activity.info(f"체크포인트 로드됨: {checkpoint_path}")
    return model_state

@workflow.defn
class ResumableTraining:
    @workflow.run
    async def run(self, dataset_url: str, total_epochs: int, lr: float = 0.001) -> dict:
        checkpoint_path = workflow.info().get_memo_field("last_checkpoint")
        
        if checkpoint_path:
            model_state = await workflow.execute_activity(
                load_checkpoint, checkpoint_path,
                start_to_close_timeout=timedelta(minutes=2)
            )
            start_epoch = int(checkpoint_path.split("_")[-1].split(".")[0])
            activity.info(f"{start_epoch} epoch에서 재개")
        else:
            model_state = initialize_model(dataset_url)
            start_epoch = 0
        
        for epoch in range(start_epoch, total_epochs):
            result = await workflow.execute_activity(
                train_epoch, model_state, epoch, lr,
                start_to_close_timeout=timedelta(minutes=30),
                retry=RetryPolicy(max_attempts=3, backoff_coefficient=2.0)
            )
            
            model_state = result["state"]
            
            if (epoch + 1) % 5 == 0:
                cp_path = await workflow.execute_activity(
                    save_checkpoint, epoch + 1, model_state,
                    start_to_close_timeout=timedelta(minutes=5)
                )
                workflow.set_memo({"last_checkpoint": cp_path})
        
        return {"final_state": model_state, "total_epochs": total_epochs}

@activity.defn
async def stream_llm_response(prompt: str, max_tokens: int = 1024) -> list[str]:
    """LLM에서 토큰을 스트리밍하여 리스트로 반환."""
    tokens = []
    async for token in call_streaming_api(prompt, max_tokens):
        tokens.append(token)
        await asyncio.sleep(0.01)
    return tokens

@workflow.defn
class PriorityWorkflow:
    def __init__(self):
        self.priority = "normal"
        self.cancel_requested = False
    
    @workflow.signal
    def set_priority(self, new_priority: str):
        self.priority = new_priority
        workflow.logger.info(f"우선순위 {new_priority}(으)로 변경됨")
    
    @workflow.signal
    def cancel_workflow(self):
        self.cancel_requested = True
        workflow.logger.info("취소 요청됨")
    
    @workflow.run
    async def run(self, task_data: dict) -> dict:
        while not self.cancel_requested:
            result = await process_task(task_data, self.priority)
            await asyncio.sleep(0.1)
        return {"status": "cancelled", "partial_result": result}

@workflow.defn
class DataPreparation:
    @workflow.run
    async def run(self, raw_data: dict) -> dict:
        cleaned = await workflow.execute_activity(clean_data, raw_data)
        validated = await workflow.execute_activity(validate_data, cleaned)
        return validated

@workflow.defn
class FullMLPipeline:
    @workflow.run
    async def run(self, raw_data: dict, model_config: dict) -> dict:
        prepared_data = await workflow.child_execute(DataPreparation.run, raw_data)
        trained_model = await workflow.child_execute(ModelTraining.run, prepared_data, model_config)
        eval_results = await workflow.child_execute(ModelEvaluation.run, trained_model)
        return eval_results

from temporalio.client import Client

client = await Client.connect("localhost:7233")
handle = client.get_workflow_handle("training-job-001")

state = await handle.query(lambda wf: wf.current_state)
print(f"현재 상태: {state}")

info = await handle.describe()
print(f"상태: {info.status}")
print(f"시작 시간: {info.start_time}")

# 모든 워크플로우 나열
temporal workflow list --namespace default

# 특정 워크플로우 설명
temporal workflow describe --workflow-id training-job-001

# 워크플로우 히스토리 표시(실행 추적)
temporal workflow show --workflow-id training-job-001

# 워크플로우를 특정 지점으로 리셋
temporal workflow reset --workflow-id training-job-001 --reset-point LastAutoClose

# 실행 중인 워크플로우 종료
temporal workflow terminate --workflow-id training-job-001 --reason "사용자 요청"

import structlog
from temporalio import activity

logger = structlog.get_logger()

@activity.defn
async def train_with_logging(model_config: dict) -> dict:
    logger.info("training_start", config=model_config)
    
    for epoch in range(10):
        loss = perform_training_epoch(model_config)
        logger.info("epoch_complete", epoch=epoch, loss=loss, learning_rate=model_config["lr"])
    
    logger.info("training_complete", final_loss=loss)
    return {"final_loss": loss}

@activity.defn
async def long_training_job(config: dict):
    for epoch in range(100):
        activity.heartbeat(f"{epoch}/100 epoch 완료")
        loss = train_one_epoch(config)
    return {"final_loss": loss}

worker = Worker(
    client, task_queue="ml-workers",
    workflows=[MLTrainingPipeline],
    activities=[train_model, evaluate_model],
    max_concurrent_activities=50,
    max_concurrent_workflow_tasks=100,
)

retry=RetryPolicy(
    initial_interval=timedelta(seconds=1),
    maximum_interval=timedelta(minutes=5),
    backoff_coefficient=2.0,
    maximum_attempts=5
)