from typing import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
class State(TypedDict):
input: str
query: str
results: str | None
def generate_query(state: State) -> State:
state["query"] = "SELECT * FROM sales WHERE date > DATE_SUB(NOW(), INTERVAL 1 MONTH)"
return state
def execute_query(state: State) -> State:
state["results"] = f"Results from executing: {state['query']}"
return state
workflow = StateGraph(State)
workflow.add_node("generate", generate_query)
workflow.add_node("execute", execute_query)
workflow.set_entry_point("generate")
workflow.add_edge("generate", "execute")
workflow.add_edge("execute", END)
app = workflow.compile(checkpointer=MemorySaver(), interrupt_before=["execute"])
config = {"configurable": {"thread_id": "1"}}
app.invoke({"input": "Find recent sales data"}, config)
# Human review
state = app.get_state(config).values
query = state["query"]
if human_approves(query):
results = app.invoke(None, config)["results"]This code shows LangGraph processing input to generate a query, interrupting for human approval via get_state/update if needed, then resuming execution with invoke(None).