Pregel은 LangGraph의 runtime을 구현하며, LangGraph 애플리케이션의 실행을 관리합니다. StateGraph를 컴파일하거나 @entrypoint를 생성하면 입력과 함께 호출할 수 있는 Pregel 인스턴스가 생성됩니다. 이 가이드는 runtime을 높은 수준에서 설명하고 Pregel을 사용하여 애플리케이션을 직접 구현하는 방법을 제공합니다.
참고: Pregel runtime은 그래프를 사용한 대규모 병렬 계산의 효율적인 방법을 설명하는 Google의 Pregel 알고리즘의 이름을 따서 명명되었습니다.

​
개요

LangGraph에서 Pregel은 actors와 channels를 단일 애플리케이션으로 결합합니다. Actors는 channels에서 데이터를 읽고 channels에 데이터를 씁니다. Pregel은 Pregel Algorithm/Bulk Synchronous Parallel 모델을 따라 애플리케이션의 실행을 여러 단계로 구성합니다. 각 단계는 세 가지 단계로 구성됩니다:
  • Plan: 이 단계에서 실행할 actors를 결정합니다. 예를 들어, 첫 번째 단계에서는 특수 input channels를 구독하는 actors를 선택하고, 후속 단계에서는 이전 단계에서 업데이트된 channels를 구독하는 actors를 선택합니다.
  • Execution: 선택된 모든 actors를 병렬로 실행하며, 모두 완료되거나, 하나가 실패하거나, 타임아웃에 도달할 때까지 실행합니다. 이 단계에서 channel 업데이트는 다음 단계까지 actors에게 보이지 않습니다.
  • Update: 이 단계에서 actors가 작성한 값으로 channels를 업데이트합니다.
실행할 actors가 선택되지 않거나 최대 단계 수에 도달할 때까지 반복합니다.

​
Actors

actor는 PregelNode입니다. channels를 구독하고, 데이터를 읽고, 데이터를 씁니다. Pregel 알고리즘의 actor로 생각할 수 있습니다. PregelNodes는 LangChain의 Runnable 인터페이스를 구현합니다.

​
Channels

Channels는 actors(PregelNodes) 간의 통신에 사용됩니다. 각 channel은 값 타입, 업데이트 타입, 업데이트 함수를 가지며, 업데이트 함수는 일련의 업데이트를 받아 저장된 값을 수정합니다. Channels는 한 chain에서 다른 chain으로 데이터를 보내거나, chain에서 미래 단계의 자신에게 데이터를 보내는 데 사용할 수 있습니다. LangGraph는 여러 내장 channels를 제공합니다:
  • LastValue: 기본 channel로, channel에 전송된 마지막 값을 저장하며, 입력 및 출력 값이나 한 단계에서 다음 단계로 데이터를 전송하는 데 유용합니다.
  • Topic: 구성 가능한 PubSub Topic으로, actors 간에 여러 값을 전송하거나 출력을 누적하는 데 유용합니다. 값을 중복 제거하거나 여러 단계에 걸쳐 값을 누적하도록 구성할 수 있습니다.
  • BinaryOperatorAggregate: 지속적인 값을 저장하며, 현재 값과 channel에 전송된 각 업데이트에 이진 연산자를 적용하여 업데이트됩니다. 여러 단계에 걸쳐 집계를 계산하는 데 유용합니다. 예: total = BinaryOperatorAggregate(int, operator.add)

​
예제

대부분의 사용자는 StateGraph API 또는 @entrypoint decorator를 통해 Pregel과 상호 작용하지만, Pregel과 직접 상호 작용하는 것도 가능합니다. 다음은 Pregel API에 대한 이해를 돕기 위한 몇 가지 예제입니다.
  • 단일 노드
  • 여러 노드
  • Topic
  • BinaryOperatorAggregate
  • 순환
from langgraph.channels import EphemeralValue
from langgraph.pregel import Pregel, NodeBuilder

node1 = (
    NodeBuilder().subscribe_only("a")
    .do(lambda x: x + x)
    .write_to("b")
)

app = Pregel(
    nodes={"node1": node1},
    channels={
        "a": EphemeralValue(str),
        "b": EphemeralValue(str),
    },
    input_channels=["a"],
    output_channels=["b"],
)

app.invoke({"a": "foo"})

{'b': 'foofoo'}

​
고수준 API

LangGraph는 Pregel 애플리케이션을 생성하기 위한 두 가지 고수준 API를 제공합니다: StateGraph (Graph API)와 Functional API입니다.
  • StateGraph (Graph API)
  • Functional API
StateGraph (Graph API)는 Pregel 애플리케이션 생성을 단순화하는 더 높은 수준의 추상화입니다. 노드와 엣지의 그래프를 정의할 수 있습니다. 그래프를 컴파일하면 StateGraph API가 자동으로 Pregel 애플리케이션을 생성합니다.
from typing import TypedDict

from langgraph.constants import START
from langgraph.graph import StateGraph

class Essay(TypedDict):
    topic: str
    content: str | None
    score: float | None

def write_essay(essay: Essay):
    return {
        "content": f"Essay about {essay['topic']}",
    }

def score_essay(essay: Essay):
    return {
        "score": 10
    }

builder = StateGraph(Essay)
builder.add_node(write_essay)
builder.add_node(score_essay)
builder.add_edge(START, "write_essay")
builder.add_edge("write_essay", "score_essay")

# Compile the graph.
# This will return a Pregel instance.
graph = builder.compile()
컴파일된 Pregel 인스턴스는 노드와 channels 목록과 연결됩니다. 노드와 channels를 출력하여 검사할 수 있습니다.
print(graph.nodes)
다음과 같은 내용이 표시됩니다:
{'__start__': <langgraph.pregel.read.PregelNode at 0x7d05e3ba1810>,
 'write_essay': <langgraph.pregel.read.PregelNode at 0x7d05e3ba14d0>,
 'score_essay': <langgraph.pregel.read.PregelNode at 0x7d05e3ba1710>}

print(graph.channels)
다음과 같은 내용이 표시됩니다:
{'topic': <langgraph.channels.last_value.LastValue at 0x7d05e3294d80>,
 'content': <langgraph.channels.last_value.LastValue at 0x7d05e3295040>,
 'score': <langgraph.channels.last_value.LastValue at 0x7d05e3295980>,
 '__start__': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e3297e00>,
 'write_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e32960c0>,
 'score_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e2d8ab80>,
 'branch:__start__:__self__:write_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e32941c0>,
 'branch:__start__:__self__:score_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e2d88800>,
 'branch:write_essay:__self__:write_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e3295ec0>,
 'branch:write_essay:__self__:score_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e2d8ac00>,
 'branch:score_essay:__self__:write_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e2d89700>,
 'branch:score_essay:__self__:score_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e2d8b400>,
 'start:write_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e2d8b280>}
Edit the source of this page on GitHub.
Connect these docs programmatically to Claude, VSCode, and more via MCP for real-time answers.
⌘I