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

​
개요

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

​
액터

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

​
채널

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

​
예제

대부분의 사용자는 StateGraph API 또는 @entrypoint 데코레이터를 통해 Pregel과 상호 작용하지만, Pregel과 직접 상호 작용하는 것도 가능합니다. 다음은 Pregel API를 이해하는 데 도움이 되는 몇 가지 예제입니다.
  • Single node
  • Multiple nodes
  • Topic
  • BinaryOperatorAggregate
  • Cycle
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")

# 그래프를 컴파일합니다.
# 이는 Pregel 인스턴스를 반환합니다.
graph = builder.compile()
컴파일된 Pregel 인스턴스는 노드와 채널 목록과 연결됩니다. 노드와 채널을 출력하여 검사할 수 있습니다.
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