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]
# Initialize KirokuForms interrupt handler
human_review_interrupt = create_kiroku_interrupt_handler(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}
def human_review_node(state: AgentState):
if state.get("needs_human_review"):
print("Requesting human review for the AI response...")
interrupt_config = {
"title": "Review AI Generated Response",
"description": f"Query: '{state['input_query']}'. Please review the AI's response and provide feedback.",
"fields": [
{"type": "textarea", "label": "AI Response (edit if needed)", "name": "ai_edited_response", "defaultValue": state["llm_response"], "required": True, "rows": 5},
{"type": "radio", "label": "Approve this response?", "name": "approved_status", "options": [{"label":"Yes, Approve", "value":"yes"}, {"label":"No, Needs Revision", "value":"no"}], "required": True},
{"type": "textarea", "label": "Reasoning or Correction Notes", "name": "reviewer_notes", "required": False, "rows": 3}
],
"wait_for_completion": True # True = block until human input, False = proceed and check later/webhook
}
# The interrupt_handler pauses the graph, creates a KirokuForms task,
# and returns the state updated with 'human_verification' containing the result.
updated_state_after_review = human_review_interrupt(state, interrupt_config)
# Optionally, process and store the review result more directly in your state
if updated_state_after_review.get("human_verification", {}).get("completed"):
review_data = updated_state_after_review["human_verification"]["result"]
# Example: appending feedback to a list in the state
current_feedback = state.get("human_review_feedback", [])
return {**updated_state_after_review, "human_review_feedback": current_feedback + [review_data], "needs_human_review": False}
return updated_state_after_review
return state # Should only happen if needs_human_review was 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("human_review_node", human_review_node)
workflow.set_entry_point("llm_agent_node")
workflow.add_conditional_edges(
"llm_agent_node",
lambda state: "human_review_node" if state.get("needs_human_review") else END,
)
workflow.add_conditional_edges( # New conditional edge from human_review_node
"human_review_node",
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