멀티스텝 추론 및 에이전시 워크플로우: 계획하고 실행하는 AI 구축

Source: Dev.to

Quick Reference: Terms You’ll Encounter

Technical Acronyms

• DAG: Directed Acyclic Graph — 순환 의존성이 없는 워크플로우 구조
• FSM: Finite State Machine — 정의된 상태와 전이로 이루어진 시스템
• CoT: Chain of Thought — 단계별 사고를 유도하는 프롬프트 기법
• ReAct: Reasoning + Acting — 사고와 도구 사용을 결합한 패턴
• LLM: Large Language Model — 트랜스포머 기반 텍스트 생성 시스템

Statistical & Mathematical Terms

• State: 워크플로우 내 모든 변수의 현재 스냅샷
• Transition: 조건에 따라 한 상태에서 다른 상태로 이동하는 것
• Topological Sort: DAG 노드들을 의존 관계가 앞에 오도록 정렬하는 방법
• Idempotent: 여러 번 실행해도 동일한 결과를 내는 연산

Introduction: From Single Prompts to Orchestrated Workflows

여행을 계획한다고 상상해 보세요. “캘리포니아까지 운전한다”라고만 말하고 바로 움직이지는 않을 겁니다. 대신:

• Decompose: 구간을 나눔 (시카고 → 덴버 → 라스베이거스 → LA)
• Plan: 주유소, 호텔, 관광지 파악
• Execute: 각 구간을 운전하고 교통·날씨에 따라 조정
• Track State: 현재 위치, 남은 연료, 완료된 작업을 파악

단일 LLM 호출은 “캘리포니아에 어떻게 가나요?” 라고 묻는 것과 같습니다. 경로는 얻지만 실행은 없습니다. 에이전시 워크플로우는 실제로 여행을 수행하며, 우회로, 펑크, 폐쇄된 도로 등을 처리합니다.

Analogy 1: 단일 프롬프트는 함수이고, 에이전시 워크플로우는 프로그램이다.
함수는 하나의 작업을 계산한다. 프로그램은 여러 함수를 조율하고, 상태를 관리하며, 오류를 처리하고, 복합적인 결과를 만든다.

Analogy 2: 전통적인 LLM 사용은 계산기이고, 에이전시 AI는 스프레드시트다.
계산기는 하나의 질문에 답한다. 스프레드시트는 상태를 유지하고, 셀 간에 의존성이 있으며, 입력이 바뀌면 자동으로 업데이트된다.

Task Decomposition: Breaking Problems Apart

from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import json

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"

@dataclass
class Task:
    """Represents a single task in a workflow."""
    id: str
    description: str
    dependencies: List[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    result: Optional[Any] = None
    error: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class TaskPlan:
    """A decomposed plan with multiple tasks."""
    goal: str
    tasks: List[Task]
    created_at: str = ""

    def get_ready_tasks(self) -> List[Task]:
        """Get tasks whose dependencies are all completed."""
        completed_ids = {t.id for t in self.tasks if t.status == TaskStatus.COMPLETED}
        ready = []
        for task in self.tasks:
            if task.status == TaskStatus.PENDING:
                if all(dep in completed_ids for dep in task.dependencies):
                    ready.append(task)
        return ready

    def is_complete(self) -> bool:
        """Check if all tasks are completed."""
        return all(t.status == TaskStatus.COMPLETED for t in self.tasks)

    def has_failed(self) -> bool:
        """Check if any task has failed."""
        return any(t.status == TaskStatus.FAILED for t in self.tasks)

class TaskDecomposer:
    """
    Decompose complex goals into executable task plans.

    Two approaches:
    1. LLM‑based: Let the model break down the task
    2. Template‑based: Use predefined patterns for known task types
    """

    DECOMPOSITION_PROMPT = """Break down this goal into specific, executable tasks.

Goal: {goal}

Context: {context}

Rules:
1. Each task should be atomic (one clear action)
2. Identify dependencies between tasks
3. Tasks should be ordered logically
4. Include validation/verification tasks where appropriate

Output JSON format:
{
    "tasks": [
        {"id": "task_1", "description": "...", "dependencies": []},
        {"id": "task_2", "description": "...", "dependencies": ["task_1"]}
    ]
}

Output only valid JSON."""

    def __init__(self, llm_client):
        self.llm = llm_client

    def decompose(self, goal: str, context: str = "") -> TaskPlan:
        """Decompose a goal into tasks using LLM."""
        prompt = self.DECOMPOSITION_PROMPT.format(goal=goal, context=context)

        response = self.llm.generate(prompt)

        # Parse response
        try:
            # Handle markdown code blocks
            if "```" in response:
                response = response.split("```")[1]
                if response.startswith("json"):
                    response = response[4:]

            data = json.loads(response.strip())

            tasks = [
                Task(
                    id=t["id"],
                    description=t["description"],
                    dependencies=t.get("dependencies", [])
                )
                for t in data["tasks"]
            ]

            return TaskPlan(goal=goal, tasks=tasks)

        except (json.JSONDecodeError, KeyError) as e:
            # Fallback: single task
            return TaskPlan(
                goal=goal,
                tasks=[Task(id="task_1", description=goal)]
            )

    def decompose_with_template(
        self,
        goal: str,
        template: str
    ) -> TaskPlan:
        """Use predefined templates for common task patterns."""

        templates = {
            "research": [
                Task(id="search", description="Search for relevant sources", dependencies=[]),
                Task(id="extract", description="Extract key information", dependencies=["search"]),
                Task(id="synthesize", description="Synthesize findings", dependencies=["extract"]),
                Task(id="validate", description="Validate accuracy", dependencies=["synthesize"])
            ],
            "data_pipeline": [
                Task(id="extract", description="Extract data from source", dependencies=[]),
                Task(id="validate_input", description="Validate input data", dependencies=["extract"]),
                Task(id="transform", description="Transform data", dependencies=["validate_input"]),
                Task(id="validate_output", description="Validate output data", dependencies=["transform"]),
                Task(id="load", description="Load to destination", dependencies=["validate_output"])
            ],
            "analysis": [
                Task(id="gather", description="Gather relevant data", dependencies=[]),
                Task(id="clean", description="Clean and prepare data", dependencies=["gather"]),
                Task(id="analyze", description="Perform analysis", dependencies=["clean"]),
                Task(id="interpret", description="Interpret results", dependencies=["analyze"]),
                Task(id="report", description="Generate report", dependencies=["interpret"])
            ]
        }

        if template not in templates:
            raise ValueError(f"Unknown template: {template}")

        # Customize task descriptions with goal
        tasks = []
        for t in templates[template]:
            tasks.append(Task(
                id=t.id,
                description=f"{t.description} for: {goal}",
                dependencies=t.dependencies.copy()
            ))

        return TaskPlan(goal=goal, tasks=tasks)

# Simple LLM client interface
class LLMClient:
    """Provider‑agnostic LLM client."""

    def __init__(self, provider: str = "openai", model: str = None):
        self.provider = provider
        self.model = model or self._default_model()

    def _default_model(self) -> str:
        return {
            "openai": "gpt-4o-mini",
            "anthropic": "claude-3-haiku-20240307"
        }.get(self.provider, "gpt-4o-mini")

    def generate(self, prompt: str, temperature: float = 0) -> str:
        if self.provider == "openai":
            # Implementation would call OpenAI API
            pass
        elif self.provider == "anthropic":
            # Implementation would call Anthropic API
            pass
        else:
            raise NotImplementedError(f"Provider {self.provider} not supported")