LangGraph Integration - KirokuForms | KirokuForms by Chelsea AI

KirokuForms provides native integration with LangGraph for human-in-the-loop workflows. This guide shows how to use KirokuForms with LangGraph's checkpointing and interrupt capabilities.

Installation

First, install the KirokuForms LangGraph integration package:

pip install langgraph-kirokuforms

Basic Integration

Here's a basic example of using KirokuForms as a human-in-the-loop provider in LangGraph:

from langgraph.graph import StateGraph, END
from langgraph.checkpoint import LocalStateCheckpointRegistry
from kirokuforms import KirokuFormsHITL

# Initialize the KirokuForms client
kiroku_client = KirokuFormsHITL(api_key="your_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,
        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
    return {**state, "task_id": task.task_id, "form_url": task.form_url}

# Define node to process human feedback
def process_human_feedback(state):
    task_id = state["task_id"]
    
    # Get the human response
    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_verified": True
    }

# Define conditional routing
def should_proceed(state):
    if "human_verified" in state and state["human_verified"]:
        return "verified" if state["verified"] else "rejected"
    else:
        return "need_verification"

# Create the graph
workflow = StateGraph()

# Add nodes
workflow.add_node("start", lambda state: state)
workflow.add_node("need_verification", require_human_verification)
workflow.add_node("process_feedback", process_human_feedback)
workflow.add_node("verified", lambda state: {**state, "result": "verified"})
workflow.add_node("rejected", lambda state: {**state, "result": "rejected"})

# Add edges
workflow.add_edge("start", should_proceed)
workflow.add_edge("need_verification", "process_feedback")
workflow.add_edge("process_feedback", should_proceed)
workflow.add_edge("verified", END)
workflow.add_edge("rejected", END)

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

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

# Run the workflow
result = app.invoke({"data": {"name": "Acme Corp", "revenue": 1500000}})

# Output the result
print(result)

Using with Interrupts

LangGraph's interrupt capability is perfect for HITL workflows. Here's how to use it with KirokuForms:

from langgraph.graph import StateGraph, END
from langgraph.checkpoint import LocalStateCheckpointRegistry
from langgraph.prebuilt import ToolNode
from kirokuforms import create_kiroku_interrupt_handler

# Initialize the KirokuForms interrupt handler
interrupt_handler = create_kiroku_interrupt_handler(api_key="your_api_key")

# Define a graph with an interrupt
workflow = StateGraph()

# Add nodes
workflow.add_node("start", lambda state: state)
workflow.add_node("process", ToolNode(tools=[...]))  # Your processing logic

# Add edges
workflow.add_edge("start", "process")
workflow.add_edge("process", END)

# Create checkpoint registry
registry = LocalStateCheckpointRegistry()

# Compile the graph with checkpointing and interrupt
app = workflow.compile(
    checkpointer=registry,
    interrupt_before=["process"],
    interrupt_handlers={
        "human_verification": interrupt_handler
    }
)

# When your agent needs human verification, it can trigger the interrupt:
# agent.interrupt("human_verification", 
#                 title="Verify Data", 
#                 description="Please verify this information",
#                 fields=[...])

Advanced Features

Webhook Integration

For production workflows, you can configure a webhook for immediate notification when human input is received:

# Configure webhook for task completion
kiroku_client = KirokuFormsHITL(
    api_key="your_api_key",
    webhook_url="https://your-server.com/webhooks/kiroku",
    webhook_secret="your_webhook_secret"
)

Custom Form Fields

You can create complex forms with various field types:

task = kiroku_client.create_task(
    title="Data Verification",
    fields=[
        {"type": "text", "label": "Company Name", "name": "company", "required": True},
        {"type": "number", "label": "Revenue", "name": "revenue"},
        {"type": "select", "label": "Industry", "name": "industry", 
         "options": [
             {"label": "Technology", "value": "tech"},
             {"label": "Healthcare", "value": "health"},
             {"label": "Finance", "value": "finance"}
         ]},
        {"type": "radio", "label": "Verified Status", "name": "status",
         "options": [
             {"label": "Verified", "value": "verified"},
             {"label": "Needs Correction", "value": "correction"},
             {"label": "Rejected", "value": "rejected"}
         ]}
    ]
)

Example Projects

Check out these example projects:

KirokuForms LangGraph Integration