에이전트나 워크플로우에서 tool 호출을 검토, 편집 및 승인하려면 LangGraph의 human-in-the-loop 기능을 사용하세요.

​
Dynamic interrupts

  • Python
  • JavaScript
  • cURL
from langgraph_sdk import get_client
from langgraph_sdk.schema import Command
client = get_client(url=<DEPLOYMENT_URL>)

# Using the graph deployed with the name "agent"
assistant_id = "agent"

# create a thread
thread = await client.threads.create()
thread_id = thread["thread_id"]

# Run the graph until the interrupt is hit.
result = await client.runs.wait(
    thread_id,
    assistant_id,
    input={"some_text": "original text"}   # (1)!
)

print(result['__interrupt__']) # (2)!
# > [
# >     {
# >         'value': {'text_to_revise': 'original text'},
# >         'resumable': True,
# >         'ns': ['human_node:fc722478-2f21-0578-c572-d9fc4dd07c3b'],
# >         'when': 'during'
# >     }
# > ]


# Resume the graph
print(await client.runs.wait(
    thread_id,
    assistant_id,
    command=Command(resume="Edited text")   # (3)!
))
# > {'some_text': 'Edited text'}
  1. 그래프가 초기 state로 호출됩니다.
  2. 그래프가 interrupt에 도달하면 payload와 metadata가 포함된 interrupt 객체를 반환합니다. 3. 그래프는 Command(resume=...)로 재개되며, 사람의 입력을 주입하고 실행을 계속합니다.
이것은 LangGraph API server에서 실행할 수 있는 예제 그래프입니다. 자세한 내용은 LangSmith quickstart를 참조하세요.
from typing import TypedDict
import uuid

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command

class State(TypedDict):
    some_text: str

def human_node(state: State):
    value = interrupt( # (1)!
        {
            "text_to_revise": state["some_text"] # (2)!
        }
    )
    return {
        "some_text": value # (3)!
    }


# Build the graph
graph_builder = StateGraph(State)
graph_builder.add_node("human_node", human_node)
graph_builder.add_edge(START, "human_node")

graph = graph_builder.compile()
  1. interrupt(...)는 human_node에서 실행을 일시 중지하고, 주어진 payload를 사람에게 표시합니다.
  2. JSON 직렬화 가능한 모든 값을 interrupt function에 전달할 수 있습니다. 여기서는 수정할 텍스트가 포함된 dict를 전달합니다.
  3. 재개되면 interrupt(...)의 반환 값은 사람이 제공한 입력이며, 이는 state를 업데이트하는 데 사용됩니다.
실행 중인 LangGraph API server가 있으면 LangGraph SDK를 사용하여 상호 작용할 수 있습니다.
  • Python
  • JavaScript
  • cURL
from langgraph_sdk import get_client
from langgraph_sdk.schema import Command
client = get_client(url=<DEPLOYMENT_URL>)

# Using the graph deployed with the name "agent"
assistant_id = "agent"

# create a thread
thread = await client.threads.create()
thread_id = thread["thread_id"]

# Run the graph until the interrupt is hit.
result = await client.runs.wait(
    thread_id,
    assistant_id,
    input={"some_text": "original text"}   # (1)!
)

print(result['__interrupt__']) # (2)!
# > [
# >     {
# >         'value': {'text_to_revise': 'original text'},
# >         'resumable': True,
# >         'ns': ['human_node:fc722478-2f21-0578-c572-d9fc4dd07c3b'],
# >         'when': 'during'
# >     }
# > ]


# Resume the graph
print(await client.runs.wait(
    thread_id,
    assistant_id,
    command=Command(resume="Edited text")   # (3)!
))
# > {'some_text': 'Edited text'}
  1. 그래프가 초기 state로 호출됩니다.
  2. 그래프가 interrupt에 도달하면 payload와 metadata가 포함된 interrupt 객체를 반환합니다. 3. 그래프는 Command(resume=...)로 재개되며, 사람의 입력을 주입하고 실행을 계속합니다.

​
Static interrupts

Static interrupts(정적 중단점이라고도 함)는 노드가 실행되기 전이나 후에 트리거됩니다.
Static interrupts는 human-in-the-loop 워크플로우에 권장되지 않습니다. 디버깅 및 테스트에 가장 적합합니다.
컴파일 시 interrupt_before와 interrupt_after를 지정하여 static interrupts를 설정할 수 있습니다: 
graph = graph_builder.compile( # (1)!
    interrupt_before=["node_a"], # (2)!
    interrupt_after=["node_b", "node_c"], # (3)!
)
  1. breakpoint는 compile 시점에 설정됩니다.
  2. interrupt_before는 노드가 실행되기 전에 실행을 일시 중지해야 하는 노드를 지정합니다.
  3. interrupt_after는 노드가 실행된 후에 실행을 일시 중지해야 하는 노드를 지정합니다.
또는 런타임에 static interrupts를 설정할 수 있습니다:
  • Python
  • JavaScript
  • cURL
await client.runs.wait( # (1)!
    thread_id,
    assistant_id,
    inputs=inputs,
    interrupt_before=["node_a"], # (2)!
    interrupt_after=["node_b", "node_c"] # (3)!
)
  1. client.runs.wait이 interrupt_before 및 interrupt_after 매개변수와 함께 호출됩니다. 이것은 런타임 구성이며 모든 호출마다 변경할 수 있습니다.
  2. interrupt_before는 노드가 실행되기 전에 실행을 일시 중지해야 하는 노드를 지정합니다.
  3. interrupt_after는 노드가 실행된 후에 실행을 일시 중지해야 하는 노드를 지정합니다.
다음 예제는 static interrupts를 추가하는 방법을 보여줍니다:
  • Python
  • JavaScript
  • cURL
from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>)

# Using the graph deployed with the name "agent"
assistant_id = "agent"

# create a thread
thread = await client.threads.create()
thread_id = thread["thread_id"]

# Run the graph until the breakpoint
result = await client.runs.wait(
    thread_id,
    assistant_id,
    input=inputs   # (1)!
)

# Resume the graph
await client.runs.wait(
    thread_id,
    assistant_id,
    input=None   # (2)!
)
  1. 그래프는 첫 번째 breakpoint에 도달할 때까지 실행됩니다.
  2. 그래프는 입력에 None을 전달하여 재개됩니다. 이렇게 하면 다음 breakpoint에 도달할 때까지 그래프가 실행됩니다.

​
더 알아보기

  • Human-in-the-loop 개념 가이드: LangGraph human-in-the-loop 기능에 대해 자세히 알아보세요.
  • 일반적인 패턴: 작업 승인/거부, 사용자 입력 요청, tool 호출 검토, 사람 입력 검증과 같은 패턴을 구현하는 방법을 알아보세요.
Edit the source of this page on GitHub.
Connect these docs programmatically to Claude, VSCode, and more via MCP for real-time answers.
⌘I