Skip to main content

​
워크플로우 패턴 (Workflow Patterns)

LangGraph에서 자주 사용되는 워크플로우 설계 패턴을 소개합니다. 각 패턴은 특정 유형의 문제를 효과적으로 해결합니다.

​
프롬프트 체이닝 (Prompt Chaining)

하나의 작업을 여러 단계로 분해하여 순차적으로 실행합니다. 각 단계의 출력이 다음 단계의 입력이 됩니다. 
from typing import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END

llm = init_chat_model("gpt-4o-mini", temperature=0)

class State(TypedDict):
    topic: str
    draft: str
    review: str
    final: str

def write_draft(state: State) -> State:
    response = llm.invoke(f"'{state['topic']}'에 대해 블로그 초안을 작성하세요.")
    return {"draft": response.content}

def review_draft(state: State) -> State:
    response = llm.invoke(
        f"다음 초안을 검토하고 개선점을 반영하여 수정하세요:\n\n{state['draft']}"
    )
    return {"review": response.content}

def format_final(state: State) -> State:
    response = llm.invoke(
        f"다음 글을 마크다운 형식으로 깔끔하게 포맷팅하세요:\n\n{state['review']}"
    )
    return {"final": response.content}

graph = StateGraph(State)
graph.add_node("draft", write_draft)
graph.add_node("review", review_draft)
graph.add_node("format", format_final)

graph.add_edge(START, "draft")
graph.add_edge("draft", "review")
graph.add_edge("review", "format")
graph.add_edge("format", END)

app = graph.compile()
result = app.invoke({"topic": "RAG의 미래"})

​
병렬화 (Parallelization)

독립적인 작업을 동시에 실행하여 처리 시간을 단축합니다.

​
Send API 활용

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
import operator

class State(TypedDict):
    topics: list[str]
    summaries: Annotated[list[str], operator.add]

class TopicState(TypedDict):
    topic: str

def generate_summary(state: TopicState) -> dict:
    response = llm.invoke(f"'{state['topic']}'을 한 문장으로 요약하세요.")
    return {"summaries": [response.content]}

def fan_out(state: State):
    return [Send("summarize", {"topic": t}) for t in state["topics"]]

graph = StateGraph(State)
graph.add_node("summarize", generate_summary)
graph.add_conditional_edges(START, fan_out, ["summarize"])
graph.add_edge("summarize", END)

app = graph.compile()
result = app.invoke({"topics": ["RAG", "LangGraph", "Vector DB"]})
# → 3개 주제가 병렬로 요약됨

​
라우팅 (Routing)

입력을 분석하여 적절한 처리 경로로 분기합니다. 
from typing import Literal

def router(state: State) -> Literal["technical", "general", "code"]:
    """질문 유형을 분류하여 라우팅"""
    response = llm.invoke(
        f"다음 질문을 분류하세요 (technical/general/code): {state['question']}"
    )
    category = response.content.strip().lower()
    if "technical" in category:
        return "technical"
    elif "code" in category:
        return "code"
    return "general"

graph = StateGraph(State)
graph.add_node("technical", technical_handler)
graph.add_node("general", general_handler)
graph.add_node("code", code_handler)

graph.add_conditional_edges(START, router, {
    "technical": "technical",
    "general": "general",
    "code": "code",
})
graph.add_edge("technical", END)
graph.add_edge("general", END)
graph.add_edge("code", END)

app = graph.compile()

​
오케스트레이터-워커 (Orchestrator-Worker)

중앙 오케스트레이터가 작업을 분배하고 조율합니다. 
class OrchestratorState(TypedDict):
    task: str
    subtasks: list[str]
    results: Annotated[list[str], operator.add]
    final_output: str

def orchestrator(state: OrchestratorState) -> OrchestratorState:
    if not state.get("subtasks"):
        # 작업 분해
        response = llm.invoke(
            f"다음 작업을 3개의 하위 작업으로 분해하세요:\n{state['task']}"
        )
        subtasks = response.content.strip().split("\n")
        return {"subtasks": subtasks}
    else:
        # 결과 종합
        combined = "\n\n".join(state["results"])
        response = llm.invoke(f"다음 결과들을 종합하세요:\n{combined}")
        return {"final_output": response.content}

​
평가자-최적화자 (Evaluator-Optimizer)

생성 결과를 평가하고 반복적으로 개선합니다. 
class EvalState(TypedDict):
    task: str
    output: str
    feedback: str
    score: float
    iteration: int

def generator(state: EvalState) -> EvalState:
    prompt = f"작업: {state['task']}"
    if state.get("feedback"):
        prompt += f"\n\n이전 피드백: {state['feedback']}\n이전 결과를 개선하세요."
    response = llm.invoke(prompt)
    return {"output": response.content, "iteration": state.get("iteration", 0) + 1}

def evaluator(state: EvalState) -> EvalState:
    response = llm.invoke(
        f"다음 결과를 1~10점으로 평가하고 피드백을 제공하세요:\n\n{state['output']}"
    )
    # 점수와 피드백 파싱
    return {"score": 8.0, "feedback": response.content}

def should_continue(state: EvalState) -> str:
    if state["score"] >= 8.0 or state["iteration"] >= 3:
        return "end"
    return "improve"

graph = StateGraph(EvalState)
graph.add_node("generate", generator)
graph.add_node("evaluate", evaluator)

graph.add_edge(START, "generate")
graph.add_edge("generate", "evaluate")
graph.add_conditional_edges("evaluate", should_continue, {
    "improve": "generate",
    "end": END,
})

app = graph.compile()

​
패턴 선택 가이드

패턴적합한 경우복잡도
프롬프트 체이닝단계적 처리 (초안→검토→완성)낮음
병렬화독립적 작업 동시 처리중간
라우팅입력 유형별 다른 처리낮음
오케스트레이터-워커복잡한 작업 분배와 조율높음
평가자-최적화자반복적 품질 개선중간
시작 추천: 프롬프트 체이닝과 라우팅부터 시작하세요. 처리 시간이 중요하면 병렬화를, 품질이 중요하면 평가자-최적화자를 추가하세요.