서버 API를 이용한 Human-in-the-loop

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

출처: 문서

본문

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

동적 interrupt

Python

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'}

그래프가 초기 상태로 호출됩니다.
그래프가 interrupt에 도달하면 페이로드와 메타데이터가 있는 interrupt 객체를 반환합니다.
Command(resume=...)로 그래프가 재개되어 인간의 입력을 주입하고 실행을 계속합니다.

JavaScript

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });

// Using the graph deployed with the name "agent"
const assistantID = "agent";

// create a thread
const thread = await client.threads.create();
const threadID = thread["thread_id"];

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

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

// Resume the graph
console.log(await client.runs.wait(
    threadID,
    assistantID,
    { command: { resume: "Edited text" }}   # (3)!
));
# > {'some_text': 'Edited text'}

그래프가 초기 상태로 호출됩니다.
그래프가 interrupt에 도달하면 페이로드와 메타데이터가 있는 interrupt 객체를 반환합니다.
{ resume: ... } 커맨드 객체로 그래프가 재개되어 인간의 입력을 주입하고 실행을 계속합니다.

cURL

스레드를 만듭니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'

interrupt에 도달할 때까지 그래프를 실행합니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/wait \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"some_text\": \"original text\"}
}"

그래프를 재개합니다:

curl --request POST \
 --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/wait \
 --header 'Content-Type: application/json' \
 --data "{
   \"assistant_id\": \"agent\",
   \"command\": {
     \"resume\": \"Edited text\"
   }
 }"

확장 예제: `interrupt` 사용하기

Agent Server에서 실행할 수 있는 예시 그래프입니다. 자세한 내용은 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에서 실행을 일시 중지하고 주어진 페이로드를 사람에게 표면화합니다.
interrupt 함수에는 JSON 직렬화 가능한 어떤 값이든 전달할 수 있습니다. 여기서는 수정할 텍스트를 담은 dict를 전달합니다.
재개되면 interrupt(...)의 반환 값은 인간이 제공한 입력이며, 이를 사용해 상태를 업데이트합니다.

실행 중인 Agent Server가 있으면 LangGraph SDK로 상호작용할 수 있습니다.

Python

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'}

그래프가 초기 상태로 호출됩니다.
그래프가 interrupt에 도달하면 페이로드와 메타데이터가 있는 interrupt 객체를 반환합니다.
Command(resume=...)로 그래프가 재개되어 인간의 입력을 주입하고 실행을 계속합니다.

JavaScript

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });

// Using the graph deployed with the name "agent"
const assistantID = "agent";

// create a thread
const thread = await client.threads.create();
const threadID = thread["thread_id"];

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

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

// Resume the graph
console.log(await client.runs.wait(
    threadID,
    assistantID,
    { command: { resume: "Edited text" }}   # (3)!
));
# > {'some_text': 'Edited text'}

그래프가 초기 상태로 호출됩니다.
그래프가 interrupt에 도달하면 페이로드와 메타데이터가 있는 interrupt 객체를 반환합니다.
{ resume: ... } 커맨드 객체로 그래프가 재개되어 인간의 입력을 주입하고 실행을 계속합니다.

cURL

스레드를 만듭니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'

interrupt에 도달할 때까지 그래프를 실행합니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/wait \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"some_text\": \"original text\"}
}"

그래프를 재개합니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/wait \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"command\": {
    \"resume\": \"Edited text\"
  }
}"

정적 interrupt

정적 interrupt(정적 중단점이라고도 함)는 노드가 실행되기 전이나 후에 트리거됩니다.

경고: 정적 interrupt는 human-in-the-loop 워크플로우에는 권장되지 않습니다. 디버깅과 테스트에 가장 적합합니다.

컴파일 시 interrupt_before와 interrupt_after를 지정해 정적 interrupt를 설정할 수 있습니다:

graph = graph_builder.compile( # (1)!
    interrupt_before=["node_a"], # (2)!
    interrupt_after=["node_b", "node_c"], # (3)!
)

중단점은 compile 시점에 설정됩니다.
interrupt_before는 노드가 실행되기 전에 실행이 일시 중지되어야 하는 노드를 지정합니다.
interrupt_after는 노드가 실행된 후에 실행이 일시 중지되어야 하는 노드를 지정합니다.

또는 런타임에 정적 interrupt를 설정할 수 있습니다:

Python

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는 노드가 실행된 후에 실행이 일시 중지되어야 하는 노드를 지정합니다.

JavaScript

await client.runs.wait( // (1)!
    threadID,
    assistantID,
    {
    input: input,
    interruptBefore: ["node_a"], // (2)!
    interruptAfter: ["node_b", "node_c"] // (3)!
    }
)

client.runs.wait가 interruptBefore 및 interruptAfter 매개변수와 함께 호출됩니다. 이는 런타임 구성이며 호출마다 변경될 수 있습니다.
interruptBefore는 노드가 실행되기 전에 실행이 일시 중지되어야 하는 노드를 지정합니다.
interruptAfter는 노드가 실행된 후에 실행이 일시 중지되어야 하는 노드를 지정합니다.

cURL

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/wait \
--header 'Content-Type: application/json' \
--data "{
    \"assistant_id\": \"agent\",
    \"interrupt_before\": [\"node_a\"],
    \"interrupt_after\": [\"node_b\", \"node_c\"],
    \"input\": <INPUT>
}"

다음 예시는 정적 interrupt를 추가하는 방법을 보여줍니다:

Python

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을 전달해 그래프가 재개됩니다. 그러면 다음 중단점에 도달할 때까지 그래프가 실행됩니다.

JavaScript

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });

// Using the graph deployed with the name "agent"
const assistantID = "agent";

// create a thread
const thread = await client.threads.create();
const threadID = thread["thread_id"];

// Run the graph until the breakpoint
const result = await client.runs.wait(
  threadID,
  assistantID,
  { input: input }   # (1)!
);

// Resume the graph
await client.runs.wait(
  threadID,
  assistantID,
  { input: null }   # (2)!
);

그래프는 첫 번째 중단점에 도달할 때까지 실행됩니다.
입력으로 null을 전달해 그래프가 재개됩니다. 그러면 다음 중단점에 도달할 때까지 그래프가 실행됩니다.

cURL

스레드를 만듭니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'

중단점까지 그래프를 실행합니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/wait \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": <INPUT>
}"

그래프를 재개합니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/wait \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\"
}"

더 알아보기

Human-in-the-loop 개념 가이드: LangGraph human-in-the-loop 기능에 대해 자세히 알아보세요.
일반적인 패턴: 동작 승인/거부, 사용자 입력 요청, 도구 호출 검토, 인간 입력 검증 같은 패턴 구현 방법을 알아보세요.

더 알아보기

Human-in-the-loop 기능 전반은 Interrupts 문서를 참고하세요.
일반적인 human-in-the-loop 패턴은 Common patterns 문서를 확인해 보세요.

서버 API를 이용한 Human-in-the-loop