[langgraph] map-reduce 브랜치를 병렬 실행하는 방법

LangGraph 공식문서를 번역한 내용입니다. 필요한 경우 부연 설명을 추가하였고 이해하기 쉽게 예제를 일부 변경하였습니다. 문제가 되면 삭제하겠습니다.

https://langchain-ai.github.io/langgraph/how-tos/map-reduce/

맵리듀스(Map-Reduce) 작업은 작업 분할과 병렬 처리를 효율적으로 수행하기 위해 필수적인 방식이다. 이 접근법은 작업을 더 작은 하위 작업으로 나누고, 각 하위 작업을 병렬로 처리한 후, 모든 하위 작업의 결과를 집계하는 과정을 포함한다.

다음과 같은 예를 생각해보자.

사용자가 일반적인 주제를 제시하면, 관련 주제 목록을 생성하고, 각 주제에 대한 농담을 만든 다음, 생성된 농담 중 가장 좋은 것을 선택하는 작업이다. 이 디자인 패턴에서는 첫 번째 노드가 객체(예: 관련 주제) 목록을 생성하고, 그 목록의 각 객체(예: 주제)에 대해 다른 노드(예: 농담 생성)를 적용하려고 한다. 그러나 여기에는 두 가지 주요한 도전 과제가 존재한다.

1. 객체(예: 주제)의 개수를 사전에 알 수 없는 경우가 많다(즉, 그래프를 설계할 때 엣지의 개수를 미리 알 수 없음).
2. 다운스트림 노드로 전달되는 입력 상태가 서로 달라야 한다(생성된 각 객체마다 별도의 상태가 필요).

LangGraph는 이러한 문제를 Send API를 통해 해결한다. 조건부 엣지(conditional edge)를 활용하여, Send는 서로 다른 상태(예: 주제)를 노드의 여러 인스턴스(예: 농담 생성)로 분배할 수 있다. 특히, 전송된 상태는 그래프의 핵심 상태와 다를 수 있어 유연하고 동적인 워크플로우 관리가 가능하다.

준비

우선, 필요한 패키지를 설치하자.

pip install -U langchain-openai langgraph

그래프 정의

import operator
from typing import Annotated

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict


from langgraph.types import Send
from langgraph.graph import END, StateGraph, START

from pydantic import BaseModel, Field

load_dotenv()

subjects_prompt = """{topic}과 관련된 2~5개의 예제 목록을 콤마를 구분으로 생성해 줘"""
joke_prompt = """{subject}에 대한 농담을 만들어 줘"""
best_joke_prompt = """아래 내용은 {topic}에 대한 농담이야. 가장 좋은 것을 선택해! 가장 좋은 것의 ID를 반환해줘.

{jokes}"""


class Subjects(BaseModel):
    subjects: list[str]


class Joke(BaseModel):
    joke: str


class BestJoke(BaseModel):
    id: int = Field(description="Index of the best joke, starting with 0", ge=0)


model = ChatOpenAI(model="gpt-4o-mini")


class OverallState(TypedDict):
    topic: str
    subjects: list
    # 개별 노드에서 생성한 모든 농담을 하나의 목록으로 결합하려고 하기 때문에 여기서 operator.add를 사용한다.
    # 이것은 본질적으로 "reduce" 부분이다.
    jokes: Annotated[list, operator.add]
    best_selected_joke: str


# 농담을 생성하기 위해 모든 주제를 "매핑"할 노드의 상태가 될 것이다.
class JokeState(TypedDict):
    subject: str


# 농담의 주제를 생성하는 데 사용할 함수이다.
def generate_topics(state: OverallState):
    prompt = subjects_prompt.format(topic=state["topic"])
    response = model.with_structured_output(Subjects).invoke(prompt)
    return {"subjects": response.subjects}


# 주제에 대한 농담을 생성하는 데 사용할 함수이다.
def generate_joke(state: JokeState):
    prompt = joke_prompt.format(subject=state["subject"])
    response = model.with_structured_output(Joke).invoke(prompt)
    return {"jokes": [response.joke]}


# 여기서 우리는 생성된 주제에 대해 매핑할 로직을 정의한다.
# 우리는 그래프의 엣지로 이것을 사용할 것이다.
def continue_to_jokes(state: OverallState):
    # 'Send' 객체의 리스트를 반환할 것이다.
    # 각 'Send' 객체는 그래프의 노드 이름과 해당 노드로 보낼 상태로 구성된다.
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]


# 최적의 농담을 판단할 것이다.
def best_joke(state: OverallState):
    jokes = "\n\n".join(state["jokes"])
    prompt = best_joke_prompt.format(topic=state["topic"], jokes=jokes)
    response = model.with_structured_output(BestJoke).invoke(prompt)
    return {"best_selected_joke": state["jokes"][response.id]}


graph = StateGraph(OverallState)
graph.add_node("generate_topics", generate_topics)
graph.add_node("generate_joke", generate_joke)
graph.add_node("best_joke", best_joke)
graph.add_edge(START, "generate_topics")
graph.add_conditional_edges("generate_topics", continue_to_jokes, ["generate_joke"])
graph.add_edge("generate_joke", "best_joke")
graph.add_edge("best_joke", END)
app = graph.compile()

LangGraph의 Send API는 특정 노드로 상태(state)를 동적으로 전송할 수 있도록 설계된 객체이다. 주로 맵-리듀스(map-reduce) 같은 패턴에서 활용되며, 조건부 엣지에서 어떤 노드로 보낼 때 사용할 수 있다.

from IPython.display import Image

Image(app.get_graph().draw_mermaid_png(output_file_path="./map-reduce-branches.png"))

그래프 사용

for s in app.stream({"topic": "정치"}):
    print(s)

{'generate_topics': {'subjects': ['정치제도', '정치철학', '정당정치', '국제관계', '선거법']}}
{'generate_joke': {'jokes': ["정치제도가 왜 복잡한지 아세요? 그건 모든 사람이 자기 의견을 의회에 '제출'하려고 하니까요!"]}}
{'generate_joke': {'jokes': ['왜 외교관들은 항상 바다에 가고 싶어할까? \n\n왜냐하면 그들은 항상 "국제 수역"을 원하기 때문이지!']}}
{'generate_joke': {'jokes': ['정치철학자는 왜 항상 혼자 있나요?\\n\\n왜냐하면 그들은 너무 많은 논의를 하느라 친구를 사귈 시간이 없거든요!']}}
{'generate_joke': {'jokes': ["왜 선거법이 항상 혼란스러울까요? \n\n투표할 때마다 '알려진 사실'이 아닌 '알려진 규칙'을 따라야 하니까요!"]}}
{'generate_joke': {'jokes': ["정당 정치를 하던 정치인이 무대에서 넘어졌어요. 관중이 환호했죠. '정당하게 넘어졌군요!'}"]}}
{'best_joke': {'best_selected_joke': '정치철학자는 왜 항상 혼자 있나요?\\n\\n왜냐하면 그들은 너무 많은 논의를 하느라 친구를 사귈 시간이 없거든요!'}}