LangGraph 런타임

Alpha Notice: These docs cover the v1-alpha release. Content is incomplete and subject to change.

For the latest stable version, see the current LangGraph Python or LangGraph JavaScript docs.

Pregel은 LangGraph의 런타임을 구현하며, LangGraph 애플리케이션의 실행을 관리합니다.

StateGraph를 컴파일하거나 entrypoint를 생성하면 입력과 함께 호출할 수 있는 Pregel 인스턴스가 생성됩니다.

이 가이드는 런타임의 고수준 개념을 설명하고, Pregel을 사용하여 직접 애플리케이션을 구현하는 방법을 제공합니다.

참고: Pregel 런타임의 이름은 그래프를 사용한 대규모 병렬 연산의 효율적인 방법을 설명하는 Google의 Pregel 알고리즘에서 유래했습니다.

​

개요

LangGraph에서 Pregel은 액터와 채널을 하나의 애플리케이션으로 결합합니다. 액터는 채널에서 데이터를 읽고 채널에 데이터를 씁니다. Pregel은 Pregel 알고리즘/대량 동기 병렬 모델을 따라 애플리케이션의 실행을 여러 단계로 구성합니다.

각 단계는 세 가지 단계로 구성됩니다:

• 계획: 이 단계에서 실행할 액터를 결정합니다. 예를 들어, 첫 번째 단계에서는 특수한 입력 채널을 구독하는 액터를 선택하고, 이후 단계에서는 이전 단계에서 업데이트된 채널을 구독하는 액터를 선택합니다.
• 실행: 선택된 모든 액터를 병렬로 실행하며, 모두 완료되거나, 하나가 실패하거나, 타임아웃에 도달할 때까지 계속됩니다. 이 단계 동안 채널 업데이트는 다음 단계까지 액터에게 보이지 않습니다.
• 업데이트: 이 단계에서 액터가 작성한 값으로 채널을 업데이트합니다.

실행할 액터가 선택되지 않거나, 최대 단계 수에 도달할 때까지 반복합니다.

​

액터

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

​

채널

채널은 액터(PregelNode) 간의 통신에 사용됩니다. 각 채널은 값 타입, 업데이트 타입, 그리고 업데이트 함수를 가지고 있습니다. 업데이트 함수는 업데이트 시퀀스를 받아 저장된 값을 수정합니다. 채널은 한 체인에서 다른 체인으로 데이터를 전송하거나, 체인에서 향후 단계의 자신에게 데이터를 전송하는 데 사용할 수 있습니다. LangGraph는 여러 내장 채널을 제공합니다:

• LastValue: 기본 채널로, 채널에 전송된 마지막 값을 저장합니다. 입력 및 출력 값, 또는 한 단계에서 다음 단계로 데이터를 전송하는 데 유용합니다.
• Topic: 구성 가능한 PubSub 토픽으로, 액터 간에 여러 값을 전송하거나 출력을 누적하는 데 유용합니다. 값을 중복 제거하거나 여러 단계에 걸쳐 값을 누적하도록 구성할 수 있습니다.
• BinaryOperatorAggregate: 지속적인 값을 저장하며, 현재 값과 채널에 전송된 각 업데이트에 이진 연산자를 적용하여 업데이트됩니다. 여러 단계에 걸쳐 집계를 계산하는 데 유용합니다. 예: total = BinaryOperatorAggregate(int, operator.add)

​

예제

대부분의 사용자는 StateGraph API 또는 entrypoint 데코레이터를 통해 Pregel과 상호작용하지만, Pregel과 직접 상호작용하는 것도 가능합니다.

다음은 Pregel API의 감각을 얻을 수 있는 몇 가지 예제입니다.

Single node

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)

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