Integrating KirokuForms with LangGraph

Leverage KirokuForms for seamless human-in-the-loop (HITL) workflows within your LangGraph applications. This guide covers integrating KirokuForms using direct node manipulation for HITL tasks and LangGraph's native interrupt capabilities.

Installation

First, ensure you have the KirokuForms LangGraph integration package. This package provides the necessary tools, such as the KirokuFormsHITL client and the create_kiroku_interrupt_handler.

Install KirokuForms LangGraph Package

pip install kirokuforms-langgraph

API Key

You'll need a KirokuForms

API key

to authenticate with the service. Ensure it's kept secure and has the appropriate permissions for creating and managing HITL tasks.

Direct Node Integration for HITL

One approach is to use the KirokuFormsHITL client to explicitly create HITL tasks within your LangGraph nodes. This method gives you fine-grained control over when and how human intervention is requested and processed.

The workflow typically involves:

A node that determines the need for human review and uses create_verification_task.
The graph pausing (implicitly, as LangGraph awaits the next step, or explicitly if using checkpoints and awaiting external triggers).
A human completing the form via the provided KirokuForms URL.
A subsequent node (often triggered by a webhook or a polling mechanism) that fetches the result using get_task_result and continues the workflow.

HITL using KirokuFormsHITL Client in Nodes

from langgraph.graph import StateGraph, END
from langgraph.checkpoint import LocalStateCheckpointRegistry # Stores workflow state
from kirokuforms import KirokuFormsHITL # KirokuForms client for HITL tasks

# Initialize the KirokuForms client
kiroku_client = KirokuFormsHITL(api_key="YOUR_KIROKU_API_KEY")

# Define node that requires human input
def require_human_verification(state):
    # Extract data that needs verification
    data = state["data"]
    
    # Create a form for human verification
    task = kiroku_client.create_verification_task(
        title="Verify Information",
        description="Please verify the accuracy of this information",
        data=data, # Data to pre-fill or display in the form
        fields=[
            {"label": "Is this information correct?", "type": "radio", "name": "is_correct", 
             "options": [{"label": "Yes", "value": "yes"}, {"label": "No", "value": "no"}]},
            {"label": "Comments", "type": "textarea", "name": "comments"}
        ]
    )
    
    # Return state with task info to pause and wait for human
    return {**state, "task_id": task.task_id, "form_url": task.form_url, "human_review_pending": True}

# Define node to process human feedback
def process_human_feedback(state):
    task_id = state["task_id"]
    
    # Get the human response from KirokuForms
    response = kiroku_client.get_task_result(task_id)
    
    # Update state based on human feedback
    return {
        **state,
        "verified": response["is_correct"] == "yes",
        "feedback": response["comments"],
        "human_review_pending": False, # Mark review as complete
        "human_verified": True
    }

# Define conditional routing
def should_proceed(state):
    if state.get("human_review_pending"):
        return "pause_for_human" # Edge not explicitly defined, graph pauses due to no outgoing edge from require_human_verification until resumed
    if "human_verified" in state and state["human_verified"]:
        return "verified" if state["verified"] else "rejected"
    else:
        return "need_verification" # Initial routing to human verification step

# Create the graph
workflow_builder = StateGraph()

# Add nodes
workflow_builder.add_node("start_node", lambda state: state) # Starting point
workflow_builder.add_node("verification_needed_node", require_human_verification)
workflow_builder.add_node("process_feedback_node", process_human_feedback)
workflow_builder.add_node("final_verified_node", lambda state: {**state, "result": "verified"})
workflow_builder.add_node("final_rejected_node", lambda state: {**state, "result": "rejected"})

# Add edges
workflow_builder.set_entry_point("start_node")
workflow_builder.add_conditional_edges(
    "start_node",
    should_proceed,
    {
        "need_verification": "verification_needed_node",
        # No direct path to verified/rejected from start if human_verified is not set
    }
)
# After human verification, LangGraph needs to be re-invoked with the state
# The 'pause_for_human' is conceptual; LangGraph with checkpoints will pause.
# This example implies manual re-invocation or a webhook to continue to process_feedback_node.
# For a more integrated flow, an interrupt handler or webhook driven resumption is better.
workflow_builder.add_edge("verification_needed_node", "process_feedback_node") # This edge is followed upon re-invocation after HITL
workflow_builder.add_conditional_edges(
    "process_feedback_node",
    should_proceed,
    {
        "verified": "final_verified_node",
        "rejected": "final_rejected_node"
    }
)
workflow_builder.add_edge("final_verified_node", END)
workflow_builder.add_edge("final_rejected_node", END)


# Create checkpoint registry (for persisting state during human verification)
registry = LocalStateCheckpointRegistry()

# Compile the graph with checkpointing
app = workflow_builder.compile(checkpointer=registry)

# Run the workflow (example assumes data is present for verification)
# In a real scenario, you might invoke, get form_url, wait for human, then resume
initial_state = {"data": {"name": "Acme Corp", "revenue": 1500000}}
# result_after_verification_request = app.invoke(initial_state) 
# print(f"Form URL: {result_after_verification_request.get('form_url')}")
# After human submits:
# state_after_human_input = app.get_state(result_after_verification_request['config']) # Get current state
# state_after_human_input.values['task_id'] = result_after_verification_request['task_id'] # ensure task_id is there for process_feedback
# final_result = app.invoke(None, state_after_human_input['config']) # Resume
# print(final_result)

Checkpointing and Resumption

For workflows that pause for human input,

LangGraph's checkpointing

is crucial. The example shows how to set up a LocalStateCheckpointRegistry. Resuming the graph after human input typically involves re-invoking the graph with the appropriate thread configuration.

Using LangGraph Interrupts

LangGraph's interrupt capability provides a more integrated way to handle HITL. You can configure your graph to pause before certain nodes and delegate the HITL task creation to KirokuForms via a registered interrupt handler.

The create_kiroku_interrupt_handler simplifies this by providing a ready-to-use handler that:

Is called by LangGraph when an interrupt is triggered.
Creates a KirokuForm based on parameters you define or pass dynamically.
Manages the pausing and eventual resumption of the graph.

HITL with KirokuForms Interrupt Handler

from langgraph.graph import StateGraph, END
from langgraph.checkpoint import LocalStateCheckpointRegistry
from langgraph.prebuilt import ToolNode # If you use tools
from kirokuforms import create_kiroku_interrupt_handler # KirokuForms interrupt factory

# Initialize the KirokuForms interrupt handler
# This handler will be called by LangGraph when an interrupt is triggered.
interrupt_handler = create_kiroku_interrupt_handler(
    api_key="YOUR_KIROKU_API_KEY"
)

# Define a graph (example nodes)
workflow_builder = StateGraph()

workflow_builder.add_node("start_node", lambda state: {**state, "message": "Workflow started"})
workflow_builder.add_node("processing_node", lambda state: {**state, "processed_data": "some_data"}) # Your core logic
# workflow_builder.add_node("tool_node", ToolNode(tools=[...])) # Example if using tools

workflow_builder.set_entry_point("start_node")
workflow_builder.add_edge("start_node", "processing_node")
workflow_builder.add_edge("processing_node", END) # Or to another node

# Create checkpoint registry
registry = LocalStateCheckpointRegistry()

# Compile the graph with checkpointing and the interrupt handler
# The interrupt will pause the graph BEFORE the "processing_node" executes.
app = workflow_builder.compile(
    checkpointer=registry,
    interrupt_before=["processing_node"], # Trigger interrupt before this node
    # interrupt_handlers={ # This was an older way, create_kiroku_interrupt_handler is self-contained
    #     "human_verification": interrupt_handler 
    # }
)

# To run and trigger the interrupt:
initial_state = {"input_data": "example"}
thread_config = {"configurable": {"thread_id": "my-thread-1"}} # Required for interrupts

# First invocation might run up to the interrupt point
app.invoke(initial_state, thread_config) 

# At this point, LangGraph would have called the interrupt_handler.
# The interrupt_handler (from KirokuForms) would create a HITL task.
# You'd get a form URL, send it to a human.
# After human submission, you would resume the graph:
# resumed_state = app.invoke(None, thread_config) # Pass None for input to resume
# print(resumed_state)

# If your agent logic (e.g., inside a tool or LLM call within a node)
# determines it needs human verification, it can directly call:
# interrupt_handler(current_state, {
#     "title": "Verify Data Manually",
#     "description": "Please verify this information from the agent.",
#     "fields": [...] # Kiroku form fields
# })
# This would then create the KirokuForm task and manage the LangGraph interrupt.

Interrupts and State

When using interrupts, LangGraph relies on checkpointing to save the state. Ensure your thread_config is consistent across invocations to correctly resume the interrupted workflow. The KirokuForms interrupt handler is designed to work with this system.

Advanced Features

Webhook Integration

For asynchronous HITL workflows, especially in production, configure KirokuForms to send a webhook notification to your server upon task completion. This allows your LangGraph application to resume processing immediately without polling.

Configuring Webhooks

# Configure KirokuForms client with webhook details
# This is typically done when initializing KirokuFormsHITL or the interrupt handler
from kirokuforms import KirokuFormsHITL, create_kiroku_interrupt_handler

# Option 1: For KirokuFormsHITL (if using direct node approach)
kiroku_client_webhook = KirokuFormsHITL(
    api_key="YOUR_KIROKU_API_KEY",
    webhook_url="https://your-server.com/webhooks/kiroku-hitl",
    webhook_secret="YOUR_SECURE_WEBHOOK_SECRET" # Used to verify webhook authenticity
)

# Option 2: For create_kiroku_interrupt_handler
interrupt_handler_webhook = create_kiroku_interrupt_handler(
    api_key="YOUR_KIROKU_API_KEY",
    webhook_url="https://your-server.com/webhooks/kiroku-interrupt",
    webhook_secret="YOUR_SECURE_WEBHOOK_SECRET"
)

# Your webhook endpoint (e.g., in a FastAPI/Flask app) would then receive
# notifications from KirokuForms when a task is completed.
# You'd use the task_id from the webhook payload to find the paused LangGraph
# workflow (via its thread_id/config) and resume it.

Your server-side webhook endpoint will need to verify the webhook signature (using the provided webhook_secret) and then use the task_id from the payload to identify and resume the correct LangGraph workflow instance.

Custom Form Fields

KirokuForms allows you to define a rich set of form fields to capture precisely the information you need from humans. This includes text inputs, selections, file uploads, and more.

Defining Complex Form Fields

# Example of defining various fields for a KirokuForms HITL task
# This would be part of the 'fields' array passed to create_verification_task
# or the interrupt_handler.

task_fields = [
    {"type": "text", "label": "Company Name", "name": "company", "required": True, "defaultValue": "Acme Corp"},
    {"type": "number", "label": "Revenue (USD)", "name": "revenue", "defaultValue": 100000},
    {"type": "textarea", "label": "Additional Notes", "name": "notes", "placeholder": "Enter any relevant notes..."},
    {"type": "email", "label": "Contact Email", "name": "contact_email"},
    {"type": "date", "label": "Founded Date", "name": "founded_date"},
    {"type": "checkbox", "label": "Is Publicly Traded?", "name": "is_public"},
    {
        "type": "select", 
        "label": "Industry", 
        "name": "industry",
        "options": [
            {"label": "Select an industry...", "value": ""},
            {"label": "Technology", "value": "tech"},
            {"label": "Healthcare", "value": "health"},
            {"label": "Finance", "value": "finance"},
            {"label": "Other", "value": "other"}
        ],
        "defaultValue": "tech"
    },
    {
        "type": "radio", 
        "label": "Verification Status", 
        "name": "status",
        "required": True,
        "options": [
            {"label": "Verified", "value": "verified"},
            {"label": "Needs Correction", "value": "correction"},
            {"label": "Rejected", "value": "rejected"}
        ]
    },
    {
        "type": "file", # Allows users to upload files
        "label": "Upload Supporting Document",
        "name": "support_doc"
    },
    {
        "type": "staticHTML", # Display read-only HTML content
        "name": "instructions_html",
        "content": "<h3>Review Instructions</h3><p>Please carefully review all fields before submitting.</p>"
    }
]

# Usage:
# kiroku_client.create_verification_task(title="Data Entry", fields=task_fields, ...)
# interrupt_handler(state, {"title": "Data Entry", "fields": task_fields, ...})

Refer to the KirokuForms documentation for a complete list of supported field types and their configuration options.

Next Steps

Now that you've seen how to integrate KirokuForms with LangGraph, explore these resources:

Review practical HITL Examples with LangGraph to see these concepts in action for various use cases.
Dive deeper into KirokuForms HITL API documentation for detailed information on task creation and management.
Explore the official LangGraph documentation for more on checkpointing, interrupts, and graph construction.