from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator
# Define your agent state if using typing
class AgentState(TypedDict):
input_query: str
llm_response: str
needs_human_review: bool
# Store human feedback, e.g., as a list of dictionaries
human_review_feedback: Annotated[List[dict], operator.add]
# One client for the whole graph.
client = KirokuFormsHITL(api_key="YOUR_KIROKU_API_KEY")
def llm_agent_node(state: AgentState):
# Your agent's logic
response = f"AI response to: {state['input_query']}" # Replace with actual LLM call
# Determine if human review is needed based on your criteria
needs_review = True # Example: always review for now
return {"llm_response": response, "needs_human_review": needs_review}
# The review step. interrupt() suspends the graph here: this process is free
# while the reviewer decides, and the run resumes later with their answer.
human_review_node = create_kiroku_interrupt_node(
client,
name="response-review",
title="Review AI Generated Response",
description="Please review the response below and approve it or ask for a revision.",
data_key="review_payload",
)
def prepare_review(state: AgentState):
"""Put this run's values where the review node can turn them into fields.
fields= is fixed when the node is built, so anything that changes per run
goes through data_key instead.
"""
return {
**state,
"review_payload": {
"input_query": state["input_query"],
"llm_response": state["llm_response"],
},
}
def record_review(state: AgentState):
"""Runs after the graph resumes, with the human's answer in the state."""
review = (state.get("human_verification") or {}).get("result") or {}
return {
"human_review_feedback": [review],
"needs_human_review": False,
}
def route_after_review(state: AgentState):
last_review = state.get("human_review_feedback", [{}])[-1]
if last_review.get("approved_status") == "yes":
print("Human approved. Ending workflow.")
return END
else:
print("Human requested revision. Potentially loop back or send to another handler.")
# Example: Could loop back to llm_agent_node or a specific rework_node
return "llm_agent_node" # Or another state
# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("llm_agent_node", llm_agent_node)
workflow.add_node("prepare_review", prepare_review)
workflow.add_node("human_review_node", human_review_node)
workflow.add_node("record_review", record_review)
workflow.set_entry_point("llm_agent_node")
workflow.add_conditional_edges(
"llm_agent_node",
lambda state: "prepare_review" if state.get("needs_human_review") else END,
)
workflow.add_edge("prepare_review", "human_review_node")
workflow.add_edge("human_review_node", "record_review")
workflow.add_conditional_edges( # after the human has answered
"record_review",
route_after_review
)
app = workflow.compile()
# Example invocation:
# initial_state = {"input_query": "Explain quantum computing simply.", "human_review_feedback": []}
# for event in app.stream(initial_state, {"recursion_limit": 10}):
# print("\n--- Event ---")
# for key, value in event.items():
# print(f"{key}: {value}")
# final_state = event.get(list(event.keys())[-1]) # Get the state from the last event part