에이전트나 워크플로에서 도구 호출을 검토하고 편집하며 승인하려면 LangGraph의 human-in-the-loop 기능을 사용하세요.
동적 인터럽트 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'}
그래프가 초기 상태와 함께 실행됩니다.
그래프가 인터럽트에 도달하면 페이로드와 메타데이터가 포함된 인터럽트 객체를 반환합니다.
3. 그래프는 Command(resume=...)를 통해 재개되며, 사용자의 입력을 주입하고 실행을 계속합니다.
다음은 LangGraph API 서버에서 실행할 수 있는 그래프 예제입니다.
자세한 내용은 LangSmith 빠른 시작 을 참조하세요. 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()
interrupt(...)는 human_node에서 실행을 일시 중지하고, 주어진 페이로드를 사용자에게 표시합니다.JSON 직렬화 가능한 모든 값을 interrupt 함수에 전달할 수 있습니다. 여기서는 수정할 텍스트가 포함된 dict를 전달합니다.
재개되면 interrupt(...)의 반환 값은 사용자가 제공한 입력이 되며, 이는 상태를 업데이트하는 데 사용됩니다.
LangGraph API 서버가 실행 중이면 LangGraph SDK 를 사용하여 상호작용할 수 있습니다. 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'}
그래프가 초기 상태와 함께 실행됩니다.
그래프가 인터럽트에 도달하면 페이로드와 메타데이터가 포함된 인터럽트 객체를 반환합니다.
3. 그래프는 Command(resume=...)를 통해 재개되며, 사용자의 입력을 주입하고 실행을 계속합니다.
정적 인터럽트 정적 인터럽트(정적 중단점이라고도 함)는 노드가 실행되기 전이나 후에 트리거됩니다.
정적 인터럽트는 human-in-the-loop 워크플로에 권장되지 않습니다 . 디버깅과 테스트에 가장 적합합니다.
컴파일 시 interrupt_before와 interrupt_after를 지정하여 정적 인터럽트를 설정할 수 있습니다:
graph = graph_builder.compile( # (1)! interrupt_before = [ "node_a" ], # (2)! interrupt_after = [ "node_b" , "node_c" ], # (3)! )
중단점은 compile 시점에 설정됩니다.
interrupt_before는 노드가 실행되기 전에 실행을 일시 중지할 노드를 지정합니다.interrupt_after는 노드가 실행된 후에 실행을 일시 중지할 노드를 지정합니다.
또는 런타임에 정적 인터럽트를 설정할 수도 있습니다:
await client.runs.wait( # (1)! thread_id, assistant_id, inputs = inputs, interrupt_before = [ "node_a" ], # (2)! interrupt_after = [ "node_b" , "node_c" ] # (3)! )
client.runs.wait가 interrupt_before 및 interrupt_after 매개변수와 함께 호출됩니다. 이는 런타임 구성이며 매 호출마다 변경할 수 있습니다.interrupt_before는 노드가 실행되기 전에 실행을 일시 중지할 노드를 지정합니다.interrupt_after는 노드가 실행된 후에 실행을 일시 중지할 노드를 지정합니다. 다음 예제는 정적 인터럽트를 추가하는 방법을 보여줍니다:
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)! )
첫 번째 중단점에 도달할 때까지 그래프가 실행됩니다.
입력에 None을 전달하여 그래프를 재개합니다. 이렇게 하면 다음 중단점에 도달할 때까지 그래프가 실행됩니다.
더 알아보기 
