How to handle Human in the loop with Langgraph and FastAPI

Introduction

Learn how to connect LangGraph with FastAPI and a React frontend so humans can review AI outputs before they go live. A step-by-step guide to building safe, production-ready AI workflows.

You can download it and follow it from here, https://github.com/bernatsampera/chatbot/tree/hitl-frontend

This project is build on top of this basic chatbot, take a look about it here, https://samperalabs.com/posts/how-to-create-a-chatbot-with-langgraph-fastapi-and-react-part-1

Create the graph and backend

Backend: hitl_graph.py

This file defines the core HITL workflow using LangGraph. It creates a state machine with two main nodes: chatbot and human_review. When a question comes in, the chatbot node generates a response using a Gemini LLM. Then, the human_review node interrupts the flow to ask a human to validate the response (1 for correct, 0 for incorrect). The final output includes both the LLM's response and whether it was marked as correct or incorrect.

...other imports
from dotenv import load_dotenv
load_dotenv()

class State(TypedDict):
    question: str
    llm_output: str
    final_output: str

# Initialize the LLM
llm = init_chat_model("google_genai:gemini-2.5-flash-lite") # requires GOOGLE_API_KEY in .env

# Define the chatbot node
def chatbot(state: State) -> Command[Literal["human_review"]]:
    response = llm.invoke(state["question"])

    # return {"llm_output": response.content}
    return Command(
        goto="human_review",
        update={"llm_output": response.content},
    )


def human_review(state: State) -> Command[Literal["__end__"]]:
    is_correct = int(interrupt("Is the response correct? (1 for yes, 0 for no): "))

    if is_correct == 1:
        return Command(
            goto=END,
            update={"final_output": f"The response is correct. {state['llm_output']}"},
        )
    else:
        return Command(
            goto=END,
            update={
                "final_output": f"The response is incorrect. {state['llm_output']}"
            },
        )

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("human_review", human_review)
graph_builder.add_edge(START, "chatbot")

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

Backend: main.py

This is the FastAPI server that exposes two key HITL endpoints:

• /start_hitl: Initiates a new HITL conversation with a question

• /resume_hitl: Continues a conversation by providing human feedback

The server manages conversation state using thread IDs and integrates with the LangGraph workflow defined in hitl_graph.py. It handles the API layer between the frontend and the graph execution.

...other imports
from chatbot.hitl_graph import graph

class StartHitlRequest(BaseModel):
    question: str
    conversation_id: str | None = None


class ResumeHitlRequest(BaseModel):
    is_correct: bool
    conversation_id: str


def _run_graph(input_data, thread_id: str):
    """Execute graph with conversation tracking."""
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke(input_data, config)
    return {"response": result, "conversation_id": thread_id}


@app.post("/start_hitl")
def start_hitl(request: StartHitlRequest):
    thread_id = request.conversation_id or str(uuid.uuid4())
    return _run_graph({"question": request.question}, thread_id)


@app.post("/resume_hitl")
def resume_hitl(request: ResumeHitlRequest):
    if not request.conversation_id:
        raise HTTPException(status_code=400, detail="Conversation ID is required")
    return _run_graph(Command(resume=request.is_correct), request.conversation_id)

Testing the endpoints with postman/curl

// Start the graph
curl --location 'http://localhost:8008/start_hitl' \
--header 'Content-Type: application/json' \
--header 'Accept: text/event-stream' \
--data '{
    "question": "What is the capital of Papua new guinea?"
}'

// Resume the graph with user answer
curl --location 'http://localhost:8008/resume_hitl' \
--header 'Content-Type: application/json' \
--header 'Accept: text/event-stream' \
--data '{
    "is_correct": 1,
    "conversation_id": "<conversation_id>" # Find it in the response of the first request
}'

Frontend

Frontend: hitlApi.ts

This service handles all API communication with the backend. It provides two main functions:

• startHitl: Sends a user's question to begin a new HITL conversation

• resumeHitl: Sends the human feedback (parsing "1" as correct, anything else as incorrect)

It also manages conversation IDs and transforms the API responses into the format expected by the UI components.

// API configuration
const API_BASE_URL = "http://localhost:8008";
const ENDPOINTS = {
  START_HITL: `${API_BASE_URL}/start_hitl`,
  RESUME_HITL: `${API_BASE_URL}/resume_hitl`
};

// Type definitions for API responses
interface HitlApiResponse {
  conversation_id: string;
  response: {
    llm_output?: string;
    final_output?: string;
  };
}

/**
 * Extracts the latest text message from a thread of messages
 */
export const getLatestUserMessage = (
  messages: readonly ThreadMessage[]
): string => {
  const lastMessage = messages.at(-1);
  const firstContent = lastMessage?.content?.[0];
  return firstContent?.type === "text" ? firstContent.text ?? "" : "";
};

/**
 * Makes a POST request to the HITL API with JSON payload
 * Handles common error cases and type safety
 */
async function makeApiRequest<T>(
  url: string,
  body: unknown,
  signal: AbortSignal
): Promise<T> {
  const response = await fetch(url, {
    method: "POST",
    headers: {"Content-Type": "application/json"},
    body: JSON.stringify(body),
    signal
  });

  if (!response.ok) {
    throw new Error(
      `HITL API request failed: ${response.status} ${response.statusText}`
    );
  }
  return response.json() as Promise<T>;
}

/**
 * Initiates a new HITL conversation or continues an existing one
 */
export async function startHitl(
  userMessage: string,
  conversationIdRef: React.RefObject<string | null>,
  abortSignal: AbortSignal
): Promise<ChatModelRunResult> {
  const data = await makeApiRequest<HitlApiResponse>(
    ENDPOINTS.START_HITL,
    {question: userMessage, conversation_id: conversationIdRef.current},
    abortSignal
  );

  if (data.conversation_id) conversationIdRef.current = data.conversation_id;

  return {
    content: [
      {
        type: "text",
        text:
          data.response.llm_output ||
          "No response received from the chat service"
      }
    ]
  };
}

/**
 * Resumes HITL conversation after user feedback (1 = correct, other = incorrect)
 */
export async function resumeHitl(
  userMessage: string,
  conversationIdRef: React.RefObject<string | null>,
  abortSignal: AbortSignal
): Promise<ChatModelRunResult> {
  const isCorrect = userMessage.includes("1");

  const data = await makeApiRequest<HitlApiResponse>(
    ENDPOINTS.RESUME_HITL,
    {is_correct: isCorrect, conversation_id: conversationIdRef.current},
    abortSignal
  );

  if (data.conversation_id) conversationIdRef.current = data.conversation_id;

  return {
    content: [
      {
        type: "text",
        text:
          data.response.final_output ||
          "No response received from the chat service"
      }
    ]
  };
}

Frontend: HitlChatbot.tsx

This is the main React component that renders the chat interface. It uses the @assistant-ui/react library to manage the chat UI, you can read more about it here, and integrates with hitlApi.ts to handle the conversation flow. It maintains the conversation state using a conversationIdRef and decides whether to start a new conversation or resume an existing one based on this ID.

import {
  AssistantRuntimeProvider,
  useLocalRuntime,
  type ChatModelAdapter,
  type ChatModelRunResult
} from "@assistant-ui/react";
import {Thread} from "@/components/assistant-ui/thread";
import {useRef} from "react";
import {getLatestUserMessage, startHitl, resumeHitl} from "@/services/hitlApi";

function HitlChatbot() {
  const conversationIdRef = useRef<string | null>(null);

  const createChatModelAdapter = (): ChatModelAdapter => ({
    async run({messages, abortSignal}) {
      try {
        const userMessage = getLatestUserMessage(messages);

        // decide based on conversationId, not hitlResponse
        return conversationIdRef.current
          ? await resumeHitl(userMessage, conversationIdRef, abortSignal)
          : await startHitl(userMessage, conversationIdRef, abortSignal);
      } catch (error) {
        console.error("Chat API error:", error);
        return {
          content: [{type: "text", text: "Failed to connect to chat service"}]
        } as ChatModelRunResult;
      }
    }
  });

  const chatRuntime = useLocalRuntime(createChatModelAdapter());

  return (
    <div className="h-screen w-screen">
      <AssistantRuntimeProvider runtime={chatRuntime}>
        <Thread />
      </AssistantRuntimeProvider>
    </div>
  );
}

export default HitlChatbot;