Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions langgraph_plugin/functional_api/human_in_the_loop/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,17 +22,19 @@ Demonstrates pausing an entrypoint with LangGraph's `interrupt()` and waiting in
Prerequisites: `uv sync --group langgraph` and a running Temporal dev server (`temporal server start-dev`).

```bash
# Terminal 1
# Terminal 1: start the worker
uv run langgraph_plugin/functional_api/human_in_the_loop/run_worker.py

# Terminal 2
# Terminal 2: start the workflow (polls for the draft, then prompts you for feedback)
uv run langgraph_plugin/functional_api/human_in_the_loop/run_workflow.py
```

When the draft is ready, you'll be prompted at the terminal. Type `approve` to accept it as-is, or type revision feedback and the draft will be revised incorporating your notes.

## Files

| File | Description |
|------|-------------|
| `workflow.py` | `@task` functions, `@entrypoint`, and `ChatbotFunctionalWorkflow` |
| `run_worker.py` | Registers tasks and entrypoint with `LangGraphPlugin`, starts worker |
| `run_workflow.py` | Starts workflow, polls draft via query, sends approval via signal |
| `run_workflow.py` | Starts workflow, polls draft via query, prompts for human feedback, sends it via signal |
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,11 @@ async def main() -> None:

print(f"Draft for review: {draft}")

# Send approval via signal
await handle.signal(ChatbotFunctionalWorkflow.provide_feedback, "approve")
# Prompt for human feedback instead of auto-approving.
feedback = await asyncio.to_thread(
input, "Enter 'approve' to accept, or type revision feedback: "
)
await handle.signal(ChatbotFunctionalWorkflow.provide_feedback, feedback)

result = await handle.result()
print(f"Final response: {result}")
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@


@task
def generate_draft(message: str) -> str:
async def generate_draft(message: str) -> str:
"""Generate a draft response. Replace with an LLM call in production."""
return (
f"Here's my response to '{message}': "
Expand All @@ -26,16 +26,16 @@ def generate_draft(message: str) -> str:


@task
def request_human_review(draft: str) -> str:
"""Pause execution to request human review of the draft."""
async def request_human_review(draft: str) -> str:
"""Present draft to human for review via interrupt; revise on feedback."""
feedback = interrupt(draft)
if feedback == "approve":
return draft
return f"[Revised] {draft} (incorporating feedback: {feedback})"


@entrypoint()
async def chatbot_entrypoint(user_message: str) -> dict:
async def chatbot_entrypoint(user_message: str) -> dict[str, Any]:
"""Chatbot entrypoint: generate a draft, get human review, return result."""
draft = await generate_draft(user_message)
final_response = await request_human_review(draft)
Expand Down
2 changes: 1 addition & 1 deletion langgraph_plugin/graph_api/human_in_the_loop/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ uv run langgraph_plugin/graph_api/human_in_the_loop/run_worker.py
uv run langgraph_plugin/graph_api/human_in_the_loop/run_workflow.py
```

When the draft is ready, you'll be prompted at the terminal. Type `approve` to accept it as-is, or type revision feedback and the draft will be regenerated by an LLM incorporating your notes.
When the draft is ready, you'll be prompted at the terminal. Type `approve` to accept it as-is, or type revision feedback and the draft will be revised incorporating your notes.

## Files

Expand Down
23 changes: 9 additions & 14 deletions langgraph_plugin/graph_api/human_in_the_loop/workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@

from datetime import timedelta

from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import START, StateGraph
Expand All @@ -21,25 +20,21 @@ class State(TypedDict):


async def generate_draft(state: State) -> dict[str, str]:
"""Generate a draft response with an LLM."""
response = await init_chat_model("claude-sonnet-4-6").ainvoke(
f"Please respond concisely to: {state['value']}"
)
return {"value": str(response.content)}
"""Generate a draft response. Replace with an LLM call in production."""
return {
"value": (
f"Here's my response to '{state['value']}': "
"The answer is 42. Let me know if this helps!"
)
}


async def human_review(state: State) -> dict[str, str]:
"""Present draft to human for review via interrupt; revise with LLM on feedback."""
"""Present draft to human for review via interrupt; revise on feedback."""
feedback = interrupt(state["value"])
if feedback == "approve":
return {"value": state["value"]}
response = await init_chat_model("claude-sonnet-4-6").ainvoke(
"Revise the following draft according to the reviewer's feedback. "
"Output only the revised draft, with no preamble.\n\n"
f"Draft:\n{state['value']}\n\n"
f"Feedback:\n{feedback}"
)
return {"value": str(response.content)}
return {"value": f"[Revised] {state['value']} (incorporating feedback: {feedback})"}


def make_chatbot_graph() -> StateGraph:
Expand Down
50 changes: 50 additions & 0 deletions tests/langgraph_plugin/functional_human_in_the_loop_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,16 @@
)


@pytest.fixture(autouse=True)
def _restore_tasks():
original_funcs = [getattr(t, "func") for t in all_tasks]
try:
yield
finally:
for t, orig in zip(all_tasks, original_funcs):
setattr(t, "func", orig)


async def test_functional_human_in_the_loop_approve(client: Client) -> None:
task_queue = f"functional-hitl-test-{uuid.uuid4()}"
plugin = LangGraphPlugin(
Expand Down Expand Up @@ -56,3 +66,43 @@ async def test_functional_human_in_the_loop_approve(client: Client) -> None:
result = await handle.result()

assert result["response"] == draft


async def test_functional_human_in_the_loop_revise(client: Client) -> None:
task_queue = f"functional-hitl-revise-test-{uuid.uuid4()}"
plugin = LangGraphPlugin(
entrypoints={"chatbot": chatbot_entrypoint},
tasks=all_tasks,
activity_options=activity_options,
)

async with Worker(
client,
task_queue=task_queue,
workflows=[ChatbotFunctionalWorkflow],
plugins=[plugin],
):
handle = await client.start_workflow(
ChatbotFunctionalWorkflow.run,
"test message",
id=f"functional-hitl-revise-{uuid.uuid4()}",
task_queue=task_queue,
)

# Poll for draft
draft = None
for _ in range(40):
await asyncio.sleep(0.25)
draft = await handle.query(ChatbotFunctionalWorkflow.get_draft)
if draft is not None:
break
assert draft is not None

# Send revision feedback
await handle.signal(
ChatbotFunctionalWorkflow.provide_feedback, "please be more concise"
)
result = await handle.result()

assert "[Revised]" in result["response"]
assert "please be more concise" in result["response"]
131 changes: 51 additions & 80 deletions tests/langgraph_plugin/human_in_the_loop_test.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
import asyncio
import sys
import uuid
from unittest.mock import patch

import pytest
from temporalio.client import Client
Expand All @@ -19,59 +18,36 @@
)


class _FakeMessage:
def __init__(self, content: str) -> None:
self.content = content


class _EchoModel:
"""Stand-in for a chat model that echoes the prompt back as its response."""

async def ainvoke(self, prompt: str) -> _FakeMessage:
return _FakeMessage(prompt)


def _fake_init_chat_model(*args: object, **kwargs: object) -> _EchoModel:
return _EchoModel()


_patch_llm = lambda: patch(
"langgraph_plugin.graph_api.human_in_the_loop.workflow.init_chat_model",
_fake_init_chat_model,
)


async def test_human_in_the_loop_approve(client: Client) -> None:
task_queue = f"hitl-test-{uuid.uuid4()}"
plugin = LangGraphPlugin(graphs={"chatbot": make_chatbot_graph()})

with _patch_llm():
async with Worker(
client,
async with Worker(
client,
task_queue=task_queue,
workflows=[ChatbotWorkflow],
plugins=[plugin],
):
handle = await client.start_workflow(
ChatbotWorkflow.run,
"test message",
id=f"hitl-{uuid.uuid4()}",
task_queue=task_queue,
workflows=[ChatbotWorkflow],
plugins=[plugin],
):
handle = await client.start_workflow(
ChatbotWorkflow.run,
"test message",
id=f"hitl-{uuid.uuid4()}",
task_queue=task_queue,
)

# Poll for draft to be ready
draft = None
for _ in range(40):
await asyncio.sleep(0.25)
draft = await handle.query(ChatbotWorkflow.get_draft)
if draft is not None:
break
assert draft is not None
assert "test message" in draft

# Approve
await handle.signal(ChatbotWorkflow.provide_feedback, "approve")
result = await handle.result()
)

# Poll for draft to be ready
draft = None
for _ in range(40):
await asyncio.sleep(0.25)
draft = await handle.query(ChatbotWorkflow.get_draft)
if draft is not None:
break
assert draft is not None
assert "test message" in draft

# Approve
await handle.signal(ChatbotWorkflow.provide_feedback, "approve")
result = await handle.result()

assert result == draft # approved draft returned as-is

Expand All @@ -80,36 +56,31 @@ async def test_human_in_the_loop_revise(client: Client) -> None:
task_queue = f"hitl-revise-test-{uuid.uuid4()}"
plugin = LangGraphPlugin(graphs={"chatbot": make_chatbot_graph()})

with _patch_llm():
async with Worker(
client,
async with Worker(
client,
task_queue=task_queue,
workflows=[ChatbotWorkflow],
plugins=[plugin],
):
handle = await client.start_workflow(
ChatbotWorkflow.run,
"test message",
id=f"hitl-revise-{uuid.uuid4()}",
task_queue=task_queue,
workflows=[ChatbotWorkflow],
plugins=[plugin],
):
handle = await client.start_workflow(
ChatbotWorkflow.run,
"test message",
id=f"hitl-revise-{uuid.uuid4()}",
task_queue=task_queue,
)

# Poll for draft
draft = None
for _ in range(40):
await asyncio.sleep(0.25)
draft = await handle.query(ChatbotWorkflow.get_draft)
if draft is not None:
break
assert draft is not None

# Send revision feedback
await handle.signal(
ChatbotWorkflow.provide_feedback, "please be more concise"
)
result = await handle.result()

# The revision node feeds the draft and feedback into the LLM; the echo
# stand-in returns the revision prompt, which contains both.
)

# Poll for draft
draft = None
for _ in range(40):
await asyncio.sleep(0.25)
draft = await handle.query(ChatbotWorkflow.get_draft)
if draft is not None:
break
assert draft is not None

# Send revision feedback
await handle.signal(ChatbotWorkflow.provide_feedback, "please be more concise")
result = await handle.result()

assert "[Revised]" in result
assert "please be more concise" in result
assert "test message" in result
Loading