AIPython

AI Agent Workflow Architectures — LangGraph로 구현하는 4가지 설계

Toma

2026년 7월 29일

약 54분

AIPython

AI Agent Workflow Architectures — LangGraph로 구현하는 4가지 설계

Toma

2026년 7월 29일

약 54분

🏗️ AI Agent Workflow Architectures — LangGraph로 구현하는 4가지 설계

📌 2024년 12월 Anthropic이 발표한 Building effective agents에서 소개한 워크플로우 아키텍처 4가지를 LangGraph 코드로 직접 구현하며 정리한 글입니다.

🤔 workflow와 agent는 무엇이 다른가

Anthropic은 이 둘을 엄격히 구분합니다. 이 구분이 아키텍처 선택의 출발점입니다.

Workflow	Agent
AI 모델과 도구가 미리 정의된 코드 경로에 따라 조정되는 시스템	AI 모델이 자신의 프로세스를 스스로 선택하는 시스템
개발자가 흐름의 통제권을 쥔다	모델이 작업 수행 방법에 대한 통제를 유지한다
경로가 예측 가능하고 디버깅이 쉽다	유저의 질문·필요에 따라 사용할 도구와 순서를 스스로 조절한다

💡 예를 들어 유저의 질문 또는 필요에 따라 에이전트가 스스로 사용할 도구를 선택하고 사용 순서도 조절하는 것을 에이전트라고 할 수 있습니다. 아래 4가지는 모두 workflow — 즉 경로가 코드로 미리 정해진 설계입니다.

🔗 1. Prompt Chaining Architecture (프롬프트 연결 기법)

핵심은 이전 결과가 다음 단계에서 처리된다는 것. 즉, 각 LLM 호출이 이전 단계의 출력을 이어받아 처리하는 방식입니다.

프로그래밍적으로 프로세스를 검증하는 gate 단계가 있으며, 조건에 맞지 않으면 종료됩니다.

✅ 언제 사용? 작업을 고정된 하위 작업으로 깔끔하게 분해할 수 있을 때 이상적.
시간이 걸리더라도 정확도를 얻고 싶다면 작업을 더 잘게 쪼개서 LLM에게 전달합니다.

🍳 예제: 요리사 워크플로우

요리 이름 in → 재료 나열 → 레시피 생성 → 플레이팅 방법 제안

먼저 State와 구조화 출력(structured output) 스키마를 정의합니다.

python

from typing_extensions import TypedDict
from typing import List
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from pydantic import BaseModel

llm = init_chat_model("openai:gpt-4o-mini")

class State(TypedDict):

    dish: str
    ingredients: list[dict]
    recipe_steps: str
    plating_instructions: str

class Ingredient(BaseModel):

    name: str
    quantity: str
    unit: str

class IngredientsOutput(BaseModel):

    ingredients: List[Ingredient]

각 노드는 이전 단계의 출력을 사용합니다. 이것이 프롬프트 체이닝의 본질입니다.

python

def list_ingredients(state: State):
   # LLM 호출 후 구조화된 출력(structured output)을 받음.
   structured_llm = llm.with_structured_output(IngredientsOutput)
   response = structured_llm.invoke(f"List 5-8 ingredients needed to make {state['dish']}")

   return {
      # LLM 응답에 ingredient가 있는걸 알고 있음 -> 출력 형식을 강제했기 때문에.
      "ingredients": response.ingredients
   }

def create_recipe(state: State):
   response = llm.invoke(f"Write a step by step cooking instruction for {state["dish"]}, using these ingredients {state['ingredients']}")

   return {
      "recipe_steps": response.content
   }

def describe_plating(state: State):
   response = llm.invoke(f"Describe how to beautifully plate this dish {state["dish"]} based on this recipe {state["recipe_steps"]}")
   return {
       "plating_instructions" : response.content
   }

💡 with_structured_output(IngredientsOutput)으로 출력 형식을 강제했기 때문에 response.ingredients가 존재한다고 확신할 수 있습니다. LLM 응답을 문자열 파싱하지 않아도 되는 이유입니다.

🚪 검증 단계 (gate)

gate는 LLM이 아니라 순수 파이썬 코드로 조건을 검사합니다. 재료 개수가 비정상이면 그래프를 즉시 종료시킵니다.

python

def gate(state: State):
   ingredients = state["ingredients"]

   if len(ingredients) > 8 or len(ingredients) < 3:
      return False
   
   return True

add_conditional_edges로 gate의 반환값(True/False)을 다음 노드에 매핑합니다.

python

graph_builder = StateGraph(State)

graph_builder.add_node("list_ingredients", list_ingredients)
graph_builder.add_node("create_recipe", create_recipe)
graph_builder.add_node("describe_plating", describe_plating)

graph_builder.add_edge(START, "list_ingredients")

# 재료가 많지 않은지 gate(검증 단계)
graph_builder.add_conditional_edges("list_ingredients", gate, {
    True: "create_recipe",
    False: END
})

graph_builder.add_edge("create_recipe", "describe_plating")
graph_builder.add_edge("describe_plating", END)

graph = graph_builder.compile()

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2026-07-17_09.39.47.png

🔀 2. Routing Architecture (라우팅 아키텍처)

핵심: 입력을 분류하고 이를 각 전문화된 후속 작업으로 안내합니다. 관심사를 분리하고 더 전문화된 프롬프트를 만드는 것을 가능하게 합니다.

✅ 언제 사용? 요청을 정확하게 분류할 수 있는 경우.
어디에 사용? 다양한 유형의 고객 서비스 요청을 안내하는 데 유용합니다.

💰 예제: 난이도에 따른 모델 라우팅

classifier(분류기)가 쉬운 질문인지 어려운 질문인지 판단하여, 쉬운 질문이면 저렴한 모델을, 어려운 질문이면 좀 더 비싼 모델을 사용하도록 분기하는 예시입니다. 비용 최적화의 전형적인 패턴입니다.

python

from typing_extensions import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from langchain.chat_models import init_chat_model
from pydantic import BaseModel

llm = init_chat_model("openai:gpt-4o-mini")

dumb_llm = init_chat_model("openai:gpt-3.5-turbo")
average_llm = init_chat_model("openai:gpt-4o")
smart_llm = init_chat_model("openai:gpt-5")

class State(TypedDict):

    question: str
    difficulty: str
    answer: str
    model_used: str

class DifficultyResponse(BaseModel):

    difficulty_level: Literal["easy", "medium", "hard"]

⚠️ Literal의 값과 == 비교 문자열은 반드시 정확히 일치해야 합니다.
Literal["easy", "medium", "hard "]처럼 공백 하나만 달라도 구조화 출력은 모델을 "hard "로 강제하는 반면 코드는 == "hard"로 비교하므로 해당 분기가 영원히 매칭되지 않습니다. 그 결과 goto가 미할당 상태로 Command(goto=goto)에 도달해 UnboundLocalError가 발생합니다. 하필 어려운 질문이 들어올 때만 터지므로 발견이 늦습니다.

각 난이도별 노드는 서로 다른 모델을 호출하고, 어떤 모델을 썼는지 State에 기록합니다.

python

def dumb_node(state: State):
   response = dumb_llm.invoke(state["question"])
   return {
      "answer" : response.content,
      "model_used" : "gpt-3.5"
   }

def average_node(state: State):
   response = average_llm.invoke(state["question"])
   return {
      "answer" : response.content,
      "model_used" : "gpt-4o"

   }

def smart_node(state: State):
   response = smart_llm.invoke(state["question"])
   return {
      "answer" : response.content,
      "model_used" : "gpt-5"

   }

난이도 평가 노드는 LLM으로 분류한 뒤 Command****로 해당 노드에 직접 점프합니다.

python

# 난이도 평가 노드
# 난이도에 따라 별도 노드로 점프.
def assess_difficulty(state: State):
   structured_llm = llm.with_structured_output(DifficultyResponse) 

   response = structured_llm.invoke(
      f"""
      Assess the difficulty of this question
      Question: {state["question"]}

      - EASY : Simple facts, basic definitions, yes/no answers
      - MEDIUM : Requires explanation, comparison, analysis
      - HARD : Complex reasoning, multiple steps, deep expertise.
      """
   )

   difficulty_level = response.difficulty_level

   if difficulty_level == "easy":
      goto = "dumb_node"
   elif difficulty_level == "medium":
      goto = "average_node"
   elif difficulty_level == "hard":
      goto = "smart_node"

   return Command(goto=goto, update={"difficulty": difficulty_level})

💡 **Command**는 LangGraph에서 일종의 치트키 — 엣지를 사용하지 않고 다른 노드로 transfer 할 수 있습니다. assess_difficulty와 dumb/average/smart 노드 사이에는 연결(엣지)이 없지만, assess_difficulty가 LLM을 호출하고 Command를 사용해서 해당하는 노드로 전송합니다.

🎨 destinations — 그래프를 예쁘게 보여주는 힌트

python

graph_builder = StateGraph(State)

graph_builder.add_node("dumb_node", dumb_node)
graph_builder.add_node("average_node", average_node)
graph_builder.add_node("smart_node", smart_node)
graph_builder.add_node("assess_difficulty", assess_difficulty, destinations=("dumb_node", "average_node", "smart_node"))

graph_builder.add_edge(START, "assess_difficulty")
graph_builder.add_edge("dumb_node", END)
graph_builder.add_edge("average_node", END)
graph_builder.add_edge("smart_node", END)

graph = graph_builder.compile()

💬 add_node에서 destinations에 Command로 점프할 목적지 목록을 전달하면 그래프를 좀 더 이쁘게 볼 수 있습니다.
**destinations**는 작동방식에 영향을 주지 않고 단지 그래프를 좀 더 직관적으로 보여주는 역할만 수행합니다.

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2026-07-17_10.01.49.png

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2026-07-17_10.02.28.png

⚡ 3. Parallelization Architecture (병렬화 아키텍처)

핵심: 여러 노드를 동시에 병렬로 실행합니다. LLM은 가끔 작업을 동시에 수행하고 각자의 출력을 프로그래밍적으로 집계할 수 있습니다.

병렬화는 2가지 주요 변형으로 나눌 수 있습니다.

Sectioning (구역 나누기) : 작업 하나를 가져와서 독립적인 하위 작업으로 나눈 다음 병렬로 실행
Voting (결과 투표) : 같은 작업을 여러 번 실행해서 다양한 출력을 얻을 수 있게 함. 같은 작업을 동시에 병렬로 실행한다는 의미. 그리고 **집계 노드(Aggregator Node)**에서 어떤 결과가 가장 좋은지 결정.

📄 예제: 문서를 4가지 관점에서 동시 분석

python

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model

llm = init_chat_model("openai:gpt-4o-mini")

class State(TypedDict):

    document: str
    summary: str
    sentiment: str
    key_points: str
    recommendation: str
    final_analysis: str

요약·감정·핵심포인트·추천을 각각 담당하는 4개 노드는 서로 독립적이므로 동시에 실행될 수 있습니다.

python

def get_summary(state: State):
   response = llm.invoke(f"Write a 3-sentence summary of this document {state["document"]}")
   return {
      "summary" : response.content,
   }

def get_sentiment(state: State):
   response = llm.invoke(f"Analyse the sentiment and tone of this document {state["document"]}")
   return {
      "sentiment" : response.content,
   }

def get_key_points(state: State):
   response = llm.invoke(f"List the 5 most important points of this document {state["document"]}")
   return {
      "key_points" : response.content,
   }

def get_recommendation(state: State):
   response = llm.invoke(f"Based on the document, list 3 recommended next steps {state["document"]}")
   return {
      "recommendation" : response.content,
   }

⚠️ 병렬 노드가 리듀서 없는 같은 State 키에 쓰면 그래프가 터집니다.
Annotated 리듀서가 없는 TypedDict 키는 LangGraph에서 LastValue 채널로 처리되며, 한 스텝에 값을 하나만 받을 수 있습니다. 만약 get_sentiment가 실수로 "sentiment" 대신 "summary"를 반환하면 get_summary와 동시에 같은 키에 쓰게 되어 InvalidUpdateError가 발생합니다 — 게다가 state['sentiment']는 영영 비어 있게 됩니다. 여러 값을 모아야 한다면 Annotated[list, add]를 사용하세요.

집계 노드는 4개 노드의 결과를 모아 최종 분석을 만듭니다.

python

def get_final_analysis(state: State):
   response = llm.invoke(
      f"""
      Give me an analysis of the following report
      DOCUMENT ANALYSIS REPORT
      ========================
      EXECUTIVE SUMMARY:
      {state['summary']}

      SENTIMENT ANALYSIS:
      {state['sentiment']}

      KEY POINTS:
      {state.get("key_points", "")}
      
      RECOMMENDATION:
      {state.get('recommendation', "N/A")}
      """
   )
   return {
      "final_analysis" : response.content,
   }

🔧 병렬 실행의 정체는 "같은 노드 뒤에 여러 엣지"

몇 개의 노드를 같은 노드 뒤에서 실행하게 만들어주면 됩니다. 별도의 병렬 API가 필요 없습니다.

python

graph_builder = StateGraph(State)

graph_builder.add_node("get_summary", get_summary)
graph_builder.add_node("get_sentiment", get_sentiment)
graph_builder.add_node("get_key_points", get_key_points)
graph_builder.add_node("get_recommendation", get_recommendation)
graph_builder.add_node("get_final_analysis", get_final_analysis)

# 모두 START 노드 뒤에서 실행 -> Langgraph가 병렬 실행
graph_builder.add_edge(START, "get_summary")
graph_builder.add_edge(START, "get_sentiment")
graph_builder.add_edge(START, "get_key_points")
graph_builder.add_edge(START, "get_recommendation")

graph_builder.add_edge("get_summary", "get_final_analysis")
graph_builder.add_edge("get_sentiment", "get_final_analysis")
graph_builder.add_edge("get_key_points", "get_final_analysis")
graph_builder.add_edge("get_recommendation", "get_final_analysis")
graph_builder.add_edge("get_final_analysis", END)

graph = graph_builder.compile()

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2026-07-17_10.44.40.png

stream_mode="updates"로 실행하면 병렬 노드들이 완료되는 순서를 실시간으로 관찰할 수 있습니다.

python

with open("fed_transcript.md", "r", encoding="utf-8") as file:
    document = file.read()

for chunk in graph.stream(
    {"document" : document},
    stream_mode="updates"  # 노드나 작업이 상태를 업데이트할 때 발생하는 실시간 이벤트를 볼 수 있음.
):
    print(chunk, "\n")

🎛️ 4. Orchestrator-workers Architecture

병렬화 아키텍처와 비슷하나 하나의 노드가 더 있으며, 그 노드가 기본적으로 몇 명의 worker를 실행할지 결정합니다.

✅ 언제 사용? 가끔 몇 개의 노드가 있는지, edge가 있는지 모르는 경우 — 즉 필요한 subtask를 예측할 수 없을 때.
어떻게? LangGraph의 Send API를 사용해서 필요한 만큼의 node를 동적으로 실행.

📝 예제: 문단 수를 모르는 문서 요약

각 문단을 요약하기 위한 목적의 워크플로우입니다. 문단 수가 몇 개인지 모르기 때문에 필요한 만큼 동적으로 node, edge 생성이 필요합니다.

python

from typing_extensions import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langgraph.types import Send
from operator import add

llm = init_chat_model("openai:gpt-4o-mini")

class State(TypedDict):

    document: str
    final_summary: str
    summaries: Annotated[list[dict], add]

💡 summaries: Annotated[list[dict], add]가 이 아키텍처의 핵심입니다. 동적으로 생성된 N개의 worker가 동시에 summaries에 쓰기 때문에, add 리듀서로 값들을 리스트에 누적시켜야 합니다. 리듀서가 없으면 앞서 본 LastValue 채널의 InvalidUpdateError를 만나게 됩니다.

dispatch_summarizers는 문서를 chunk로 쪼갠 뒤, chunk 개수만큼 Send 명령을 반환합니다.

python

def summarize_p(args):
   paragraph = args["paragraph"]
   index = args["index"]
   response = llm.invoke(
      f"Write a 3-sentence summary for this paragraph: {paragraph}"
   )
   return {
      "summaries": [
         {
            "summary": response.content,
            "index": index
         }
      ]
   }

def dispatch_summarizers(state: State):
   # 줄바꿈을 기준으로 chunk 생성
   chunks = state["document"].split("\n\n")

   # 각 chunk마다 Send command return
   return [
      # enumerate는 지금 처리 중인 element의 index번호를 줌.
      Send("summarize_p", {"paragraph": chunk, "index": index}) for index, chunk in enumerate(chunks)
   ]

def final_summary(state: State):
   response = llm.invoke(f"Using the following summaries, give me a final one {state["summaries"]}")

   return {
      "final_summary": response.content
   }

💡 summarize_p의 인자가 state: State가 아니라 **args**인 점에 주목하세요. Send("summarize_p", {...})로 전달한 payload를 직접 받기 때문입니다. 각 worker는 전체 State가 아니라 자기가 처리할 문단 하나만 봅니다.

노드는 단 하나만 등록하지만, 실행 시점에는 문단 개수만큼 인스턴스가 생성됩니다.

python

graph_builder = StateGraph(State)

graph_builder.add_node("summarize_p", summarize_p)
graph_builder.add_node("final_summary", final_summary)


# dispatch_summarizers는 Send command를 사용해서 summarize_p를 필요한만큼 생성하고 실행.
graph_builder.add_conditional_edges(START, dispatch_summarizers, ["summarize_p"])

graph_builder.add_edge("summarize_p", "final_summary")
graph_builder.add_edge("final_summary", END)

graph = graph_builder.compile()

python

with open("fed_transcript.md", "r", encoding="utf-8") as file:
    document = file.read()

for chunk in graph.stream(
    {"document" : document},
    stream_mode="updates" 
):
    print(chunk, "\n")

🗺️ 4가지 아키텍처 한눈에 비교

아키텍처	핵심	언제 사용	LangGraph 도구
Prompt Chaining	이전 출력이 다음 단계 입력이 됨	고정된 하위 작업으로 깔끔히 분해 가능할 때	`add_edge` • gate용 `add_conditional_edges`
Routing	입력을 분류해 전문화된 경로로 안내	요청을 정확히 분류할 수 있을 때	`Command(goto=...)` • `destinations`
Parallelization	독립 작업을 동시 실행 후 집계	하위 작업이 서로 독립적이고 개수를 아는 경우	같은 노드 뒤에 여러 `add_edge`
Orchestrator-workers	worker 수를 런타임에 결정해 동적 확산	필요한 subtask 수를 예측할 수 없을 때	`Send` API + `Annotated[list, add]`

🎯 Parallelization과 Orchestrator의 결정적 차이는 "개수를 아는가"입니다.
문서를 요약·감정·핵심포인트·추천 4가지로 분석하는 건 개수가 고정이라 엣지를 하드코딩할 수 있지만, 문단 수를 모르는 문서 요약은 런타임에만 개수를 알 수 있어 Send가 필요합니다.

📚 참고

🏗️ AI Agent Workflow Architectures — LangGraph로 구현하는 4가지 설계

📌 2024년 12월 Anthropic이 발표한 Building effective agents에서 소개한 워크플로우 아키텍처 4가지를 LangGraph 코드로 직접 구현하며 정리한 글입니다.

🤔 workflow와 agent는 무엇이 다른가

Anthropic은 이 둘을 엄격히 구분합니다. 이 구분이 아키텍처 선택의 출발점입니다.

Workflow	Agent
AI 모델과 도구가 미리 정의된 코드 경로에 따라 조정되는 시스템	AI 모델이 자신의 프로세스를 스스로 선택하는 시스템
개발자가 흐름의 통제권을 쥔다	모델이 작업 수행 방법에 대한 통제를 유지한다
경로가 예측 가능하고 디버깅이 쉽다	유저의 질문·필요에 따라 사용할 도구와 순서를 스스로 조절한다

💡 예를 들어 유저의 질문 또는 필요에 따라 에이전트가 스스로 사용할 도구를 선택하고 사용 순서도 조절하는 것을 에이전트라고 할 수 있습니다. 아래 4가지는 모두 workflow — 즉 경로가 코드로 미리 정해진 설계입니다.

🔗 1. Prompt Chaining Architecture (프롬프트 연결 기법)

핵심은 이전 결과가 다음 단계에서 처리된다는 것. 즉, 각 LLM 호출이 이전 단계의 출력을 이어받아 처리하는 방식입니다.

프로그래밍적으로 프로세스를 검증하는 gate 단계가 있으며, 조건에 맞지 않으면 종료됩니다.

✅ 언제 사용? 작업을 고정된 하위 작업으로 깔끔하게 분해할 수 있을 때 이상적.
시간이 걸리더라도 정확도를 얻고 싶다면 작업을 더 잘게 쪼개서 LLM에게 전달합니다.

🍳 예제: 요리사 워크플로우

요리 이름 in → 재료 나열 → 레시피 생성 → 플레이팅 방법 제안

먼저 State와 구조화 출력(structured output) 스키마를 정의합니다.

python

from typing_extensions import TypedDict
from typing import List
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from pydantic import BaseModel

llm = init_chat_model("openai:gpt-4o-mini")

class State(TypedDict):

    dish: str
    ingredients: list[dict]
    recipe_steps: str
    plating_instructions: str

class Ingredient(BaseModel):

    name: str
    quantity: str
    unit: str

class IngredientsOutput(BaseModel):

    ingredients: List[Ingredient]

각 노드는 이전 단계의 출력을 사용합니다. 이것이 프롬프트 체이닝의 본질입니다.

python

def list_ingredients(state: State):
   # LLM 호출 후 구조화된 출력(structured output)을 받음.
   structured_llm = llm.with_structured_output(IngredientsOutput)
   response = structured_llm.invoke(f"List 5-8 ingredients needed to make {state['dish']}")

   return {
      # LLM 응답에 ingredient가 있는걸 알고 있음 -> 출력 형식을 강제했기 때문에.
      "ingredients": response.ingredients
   }

def create_recipe(state: State):
   response = llm.invoke(f"Write a step by step cooking instruction for {state["dish"]}, using these ingredients {state['ingredients']}")

   return {
      "recipe_steps": response.content
   }

def describe_plating(state: State):
   response = llm.invoke(f"Describe how to beautifully plate this dish {state["dish"]} based on this recipe {state["recipe_steps"]}")
   return {
       "plating_instructions" : response.content
   }

💡 with_structured_output(IngredientsOutput)으로 출력 형식을 강제했기 때문에 response.ingredients가 존재한다고 확신할 수 있습니다. LLM 응답을 문자열 파싱하지 않아도 되는 이유입니다.

🚪 검증 단계 (gate)

gate는 LLM이 아니라 순수 파이썬 코드로 조건을 검사합니다. 재료 개수가 비정상이면 그래프를 즉시 종료시킵니다.

python

def gate(state: State):
   ingredients = state["ingredients"]

   if len(ingredients) > 8 or len(ingredients) < 3:
      return False
   
   return True

add_conditional_edges로 gate의 반환값(True/False)을 다음 노드에 매핑합니다.

python

graph_builder = StateGraph(State)

graph_builder.add_node("list_ingredients", list_ingredients)
graph_builder.add_node("create_recipe", create_recipe)
graph_builder.add_node("describe_plating", describe_plating)

graph_builder.add_edge(START, "list_ingredients")

# 재료가 많지 않은지 gate(검증 단계)
graph_builder.add_conditional_edges("list_ingredients", gate, {
    True: "create_recipe",
    False: END
})

graph_builder.add_edge("create_recipe", "describe_plating")
graph_builder.add_edge("describe_plating", END)

graph = graph_builder.compile()

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2026-07-17_09.39.47.png

🔀 2. Routing Architecture (라우팅 아키텍처)

핵심: 입력을 분류하고 이를 각 전문화된 후속 작업으로 안내합니다. 관심사를 분리하고 더 전문화된 프롬프트를 만드는 것을 가능하게 합니다.

✅ 언제 사용? 요청을 정확하게 분류할 수 있는 경우.
어디에 사용? 다양한 유형의 고객 서비스 요청을 안내하는 데 유용합니다.

💰 예제: 난이도에 따른 모델 라우팅

python

from typing_extensions import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from langchain.chat_models import init_chat_model
from pydantic import BaseModel

llm = init_chat_model("openai:gpt-4o-mini")

dumb_llm = init_chat_model("openai:gpt-3.5-turbo")
average_llm = init_chat_model("openai:gpt-4o")
smart_llm = init_chat_model("openai:gpt-5")

class State(TypedDict):

    question: str
    difficulty: str
    answer: str
    model_used: str

class DifficultyResponse(BaseModel):

    difficulty_level: Literal["easy", "medium", "hard"]

⚠️ Literal의 값과 == 비교 문자열은 반드시 정확히 일치해야 합니다.
Literal["easy", "medium", "hard "]처럼 공백 하나만 달라도 구조화 출력은 모델을 "hard "로 강제하는 반면 코드는 == "hard"로 비교하므로 해당 분기가 영원히 매칭되지 않습니다. 그 결과 goto가 미할당 상태로 Command(goto=goto)에 도달해 UnboundLocalError가 발생합니다. 하필 어려운 질문이 들어올 때만 터지므로 발견이 늦습니다.

각 난이도별 노드는 서로 다른 모델을 호출하고, 어떤 모델을 썼는지 State에 기록합니다.

python

def dumb_node(state: State):
   response = dumb_llm.invoke(state["question"])
   return {
      "answer" : response.content,
      "model_used" : "gpt-3.5"
   }

def average_node(state: State):
   response = average_llm.invoke(state["question"])
   return {
      "answer" : response.content,
      "model_used" : "gpt-4o"

   }

def smart_node(state: State):
   response = smart_llm.invoke(state["question"])
   return {
      "answer" : response.content,
      "model_used" : "gpt-5"

   }

난이도 평가 노드는 LLM으로 분류한 뒤 Command****로 해당 노드에 직접 점프합니다.

python

# 난이도 평가 노드
# 난이도에 따라 별도 노드로 점프.
def assess_difficulty(state: State):
   structured_llm = llm.with_structured_output(DifficultyResponse) 

   response = structured_llm.invoke(
      f"""
      Assess the difficulty of this question
      Question: {state["question"]}

      - EASY : Simple facts, basic definitions, yes/no answers
      - MEDIUM : Requires explanation, comparison, analysis
      - HARD : Complex reasoning, multiple steps, deep expertise.
      """
   )

   difficulty_level = response.difficulty_level

   if difficulty_level == "easy":
      goto = "dumb_node"
   elif difficulty_level == "medium":
      goto = "average_node"
   elif difficulty_level == "hard":
      goto = "smart_node"

   return Command(goto=goto, update={"difficulty": difficulty_level})

💡 **Command**는 LangGraph에서 일종의 치트키 — 엣지를 사용하지 않고 다른 노드로 transfer 할 수 있습니다. assess_difficulty와 dumb/average/smart 노드 사이에는 연결(엣지)이 없지만, assess_difficulty가 LLM을 호출하고 Command를 사용해서 해당하는 노드로 전송합니다.

🎨 destinations — 그래프를 예쁘게 보여주는 힌트

python

graph_builder = StateGraph(State)

graph_builder.add_node("dumb_node", dumb_node)
graph_builder.add_node("average_node", average_node)
graph_builder.add_node("smart_node", smart_node)
graph_builder.add_node("assess_difficulty", assess_difficulty, destinations=("dumb_node", "average_node", "smart_node"))

graph_builder.add_edge(START, "assess_difficulty")
graph_builder.add_edge("dumb_node", END)
graph_builder.add_edge("average_node", END)
graph_builder.add_edge("smart_node", END)

graph = graph_builder.compile()

💬 add_node에서 destinations에 Command로 점프할 목적지 목록을 전달하면 그래프를 좀 더 이쁘게 볼 수 있습니다.
**destinations**는 작동방식에 영향을 주지 않고 단지 그래프를 좀 더 직관적으로 보여주는 역할만 수행합니다.

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2026-07-17_10.01.49.png

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2026-07-17_10.02.28.png

⚡ 3. Parallelization Architecture (병렬화 아키텍처)

핵심: 여러 노드를 동시에 병렬로 실행합니다. LLM은 가끔 작업을 동시에 수행하고 각자의 출력을 프로그래밍적으로 집계할 수 있습니다.

병렬화는 2가지 주요 변형으로 나눌 수 있습니다.

Sectioning (구역 나누기) : 작업 하나를 가져와서 독립적인 하위 작업으로 나눈 다음 병렬로 실행
Voting (결과 투표) : 같은 작업을 여러 번 실행해서 다양한 출력을 얻을 수 있게 함. 같은 작업을 동시에 병렬로 실행한다는 의미. 그리고 **집계 노드(Aggregator Node)**에서 어떤 결과가 가장 좋은지 결정.

📄 예제: 문서를 4가지 관점에서 동시 분석

python

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model

llm = init_chat_model("openai:gpt-4o-mini")

class State(TypedDict):

    document: str
    summary: str
    sentiment: str
    key_points: str
    recommendation: str
    final_analysis: str

요약·감정·핵심포인트·추천을 각각 담당하는 4개 노드는 서로 독립적이므로 동시에 실행될 수 있습니다.

python

def get_summary(state: State):
   response = llm.invoke(f"Write a 3-sentence summary of this document {state["document"]}")
   return {
      "summary" : response.content,
   }

def get_sentiment(state: State):
   response = llm.invoke(f"Analyse the sentiment and tone of this document {state["document"]}")
   return {
      "sentiment" : response.content,
   }

def get_key_points(state: State):
   response = llm.invoke(f"List the 5 most important points of this document {state["document"]}")
   return {
      "key_points" : response.content,
   }

def get_recommendation(state: State):
   response = llm.invoke(f"Based on the document, list 3 recommended next steps {state["document"]}")
   return {
      "recommendation" : response.content,
   }

⚠️ 병렬 노드가 리듀서 없는 같은 State 키에 쓰면 그래프가 터집니다.
Annotated 리듀서가 없는 TypedDict 키는 LangGraph에서 LastValue 채널로 처리되며, 한 스텝에 값을 하나만 받을 수 있습니다. 만약 get_sentiment가 실수로 "sentiment" 대신 "summary"를 반환하면 get_summary와 동시에 같은 키에 쓰게 되어 InvalidUpdateError가 발생합니다 — 게다가 state['sentiment']는 영영 비어 있게 됩니다. 여러 값을 모아야 한다면 Annotated[list, add]를 사용하세요.

집계 노드는 4개 노드의 결과를 모아 최종 분석을 만듭니다.

python

def get_final_analysis(state: State):
   response = llm.invoke(
      f"""
      Give me an analysis of the following report
      DOCUMENT ANALYSIS REPORT
      ========================
      EXECUTIVE SUMMARY:
      {state['summary']}

      SENTIMENT ANALYSIS:
      {state['sentiment']}

      KEY POINTS:
      {state.get("key_points", "")}
      
      RECOMMENDATION:
      {state.get('recommendation', "N/A")}
      """
   )
   return {
      "final_analysis" : response.content,
   }

🔧 병렬 실행의 정체는 "같은 노드 뒤에 여러 엣지"

몇 개의 노드를 같은 노드 뒤에서 실행하게 만들어주면 됩니다. 별도의 병렬 API가 필요 없습니다.

python

graph_builder = StateGraph(State)

graph_builder.add_node("get_summary", get_summary)
graph_builder.add_node("get_sentiment", get_sentiment)
graph_builder.add_node("get_key_points", get_key_points)
graph_builder.add_node("get_recommendation", get_recommendation)
graph_builder.add_node("get_final_analysis", get_final_analysis)

# 모두 START 노드 뒤에서 실행 -> Langgraph가 병렬 실행
graph_builder.add_edge(START, "get_summary")
graph_builder.add_edge(START, "get_sentiment")
graph_builder.add_edge(START, "get_key_points")
graph_builder.add_edge(START, "get_recommendation")

graph_builder.add_edge("get_summary", "get_final_analysis")
graph_builder.add_edge("get_sentiment", "get_final_analysis")
graph_builder.add_edge("get_key_points", "get_final_analysis")
graph_builder.add_edge("get_recommendation", "get_final_analysis")
graph_builder.add_edge("get_final_analysis", END)

graph = graph_builder.compile()

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2026-07-17_10.44.40.png

stream_mode="updates"로 실행하면 병렬 노드들이 완료되는 순서를 실시간으로 관찰할 수 있습니다.

python

with open("fed_transcript.md", "r", encoding="utf-8") as file:
    document = file.read()

for chunk in graph.stream(
    {"document" : document},
    stream_mode="updates"  # 노드나 작업이 상태를 업데이트할 때 발생하는 실시간 이벤트를 볼 수 있음.
):
    print(chunk, "\n")

🎛️ 4. Orchestrator-workers Architecture

병렬화 아키텍처와 비슷하나 하나의 노드가 더 있으며, 그 노드가 기본적으로 몇 명의 worker를 실행할지 결정합니다.

✅ 언제 사용? 가끔 몇 개의 노드가 있는지, edge가 있는지 모르는 경우 — 즉 필요한 subtask를 예측할 수 없을 때.
어떻게? LangGraph의 Send API를 사용해서 필요한 만큼의 node를 동적으로 실행.

📝 예제: 문단 수를 모르는 문서 요약

각 문단을 요약하기 위한 목적의 워크플로우입니다. 문단 수가 몇 개인지 모르기 때문에 필요한 만큼 동적으로 node, edge 생성이 필요합니다.

python

from typing_extensions import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langgraph.types import Send
from operator import add

llm = init_chat_model("openai:gpt-4o-mini")

class State(TypedDict):

    document: str
    final_summary: str
    summaries: Annotated[list[dict], add]

💡 summaries: Annotated[list[dict], add]가 이 아키텍처의 핵심입니다. 동적으로 생성된 N개의 worker가 동시에 summaries에 쓰기 때문에, add 리듀서로 값들을 리스트에 누적시켜야 합니다. 리듀서가 없으면 앞서 본 LastValue 채널의 InvalidUpdateError를 만나게 됩니다.

dispatch_summarizers는 문서를 chunk로 쪼갠 뒤, chunk 개수만큼 Send 명령을 반환합니다.

python

def summarize_p(args):
   paragraph = args["paragraph"]
   index = args["index"]
   response = llm.invoke(
      f"Write a 3-sentence summary for this paragraph: {paragraph}"
   )
   return {
      "summaries": [
         {
            "summary": response.content,
            "index": index
         }
      ]
   }

def dispatch_summarizers(state: State):
   # 줄바꿈을 기준으로 chunk 생성
   chunks = state["document"].split("\n\n")

   # 각 chunk마다 Send command return
   return [
      # enumerate는 지금 처리 중인 element의 index번호를 줌.
      Send("summarize_p", {"paragraph": chunk, "index": index}) for index, chunk in enumerate(chunks)
   ]

def final_summary(state: State):
   response = llm.invoke(f"Using the following summaries, give me a final one {state["summaries"]}")

   return {
      "final_summary": response.content
   }

💡 summarize_p의 인자가 state: State가 아니라 **args**인 점에 주목하세요. Send("summarize_p", {...})로 전달한 payload를 직접 받기 때문입니다. 각 worker는 전체 State가 아니라 자기가 처리할 문단 하나만 봅니다.

노드는 단 하나만 등록하지만, 실행 시점에는 문단 개수만큼 인스턴스가 생성됩니다.

python

graph_builder = StateGraph(State)

graph_builder.add_node("summarize_p", summarize_p)
graph_builder.add_node("final_summary", final_summary)


# dispatch_summarizers는 Send command를 사용해서 summarize_p를 필요한만큼 생성하고 실행.
graph_builder.add_conditional_edges(START, dispatch_summarizers, ["summarize_p"])

graph_builder.add_edge("summarize_p", "final_summary")
graph_builder.add_edge("final_summary", END)

graph = graph_builder.compile()

python

with open("fed_transcript.md", "r", encoding="utf-8") as file:
    document = file.read()

for chunk in graph.stream(
    {"document" : document},
    stream_mode="updates" 
):
    print(chunk, "\n")

🗺️ 4가지 아키텍처 한눈에 비교

아키텍처	핵심	언제 사용	LangGraph 도구
Prompt Chaining	이전 출력이 다음 단계 입력이 됨	고정된 하위 작업으로 깔끔히 분해 가능할 때	`add_edge` • gate용 `add_conditional_edges`
Routing	입력을 분류해 전문화된 경로로 안내	요청을 정확히 분류할 수 있을 때	`Command(goto=...)` • `destinations`
Parallelization	독립 작업을 동시 실행 후 집계	하위 작업이 서로 독립적이고 개수를 아는 경우	같은 노드 뒤에 여러 `add_edge`
Orchestrator-workers	worker 수를 런타임에 결정해 동적 확산	필요한 subtask 수를 예측할 수 없을 때	`Send` API + `Annotated[list, add]`

🎯 Parallelization과 Orchestrator의 결정적 차이는 "개수를 아는가"입니다.
문서를 요약·감정·핵심포인트·추천 4가지로 분석하는 건 개수가 고정이라 엣지를 하드코딩할 수 있지만, 문단 수를 모르는 문서 요약은 런타임에만 개수를 알 수 있어 Send가 필요합니다.