Using human in the loop in AI application

Text to Review:

Table of contents

Intro

In every application, we should have control over the data regarding how it’s created, updated, and processed. As Gen AI generates a lot of data, how do we govern this? In this article, I will explain how I implemented human-in-the-loop for my application. The application we are going to see is a web app that reviews articles.

What is HITL

HITL - Human-in-the-loop which is basically a step where the process is halted and its state is reviewed by a human for verification and approval. In my application, the LLM reviews the articles that I wrote and corrects the verbiage and submits it back for verification.

How checkpoint helps

To implement HITL we need checkpoints to halt and resume as we covered in the previous article on implementing the checkpoint using Postgres. This application also uses the same Postgres checkpoint to pause/resume based on human feedback.

Building the Graph

The graph consists of the following nodes

• Build Article Review Prompt : Constructs the system prompt instructing the LLM to act as a reviewer and verify verbiage mistakes and return the article highlighting the changes.
• LLM Review - Injects the System message and the prompt which contains the article and the LLM runs the prompt and returns the result to the langgraph state
• Human Review - Reads the state and interrupts the workflow and waits for human review
• Approved - If approved by the user, the process is done.
• Rejected - If rejected, it takes the comments back and goes back to the first step to correct the mistakes.

The graph is shown below

• Implementation of the nodes

def build_prompt_node(state: ArticleReviewState):
    """
    Constructs a prompt node for the article review process based on the provided state.
    """

    extracted_body = extract_description_and_body(state.get("raw_content", ""))
    reviewer_comments = state.get("reviewer_comments", "")
    prompt = build_editorial_messages(extracted_body, reviewer_comments)

    return {
        "extracted_body": extracted_body,
        "reviewer_comments": reviewer_comments,
        "prompt": prompt
    }

async def llm_correction_node(state: ArticleReviewState, config: RunnableConfig):
    """
    Constructs a correction node for the LLM to process based on the provided state.
    On failure, updates article status to 'AI Review Failed' and deactivates the thread.
    """

    configurable = config.get("configurable", {})
    db = configurable.get("db")
    thread_id = configurable.get("thread_id", "")
    prompt = state.get("prompt") or []
    extracted_body = state.get("extracted_body", "")
    reviewer_comments = state.get("reviewer_comments", "") or ""
    article_id = state.get("article_id", "")

    try:
        success, ai_content = await invoke_editorial_llm(prompt, extracted_body, reviewer_comments)

        if not success or not ai_content.strip():
            raise ValueError("LLM returned invalid content. Please check your API keys and configuration.")
        
        update_ai_article_with_status(db, article_id, "AI Review Completed", reviewer_comments, ai_content.strip())

        print("LLM Correction Node: AI Content:", ai_content)
        
        return {
            "ai_content": ai_content.strip(),
            "extracted_body": extracted_body,
            "reviewer_comments": reviewer_comments,
            "prompt": prompt,
            "status": "AI Review Completed"
        }
    except Exception as e:
        print(f"LLM Correction Node failed for article {article_id}: {e}")
        # Update article status to failed
        try:
            update_ai_article_with_status(db, article_id, "AI Review Failed", f"AI Review failed: {str(e)}", "")
        except Exception as db_error:
            print(f"Failed to update article status to 'AI Review Failed': {db_error}")
        # Deactivate the thread so next run creates a new one
        try:
            if thread_id:
                deactivate_thread(db, thread_id)
        except Exception as thread_error:
            print(f"Failed to deactivate thread {thread_id}: {thread_error}")
        raise

def human_review_node(state: ArticleReviewState):
    """
    Constructs a human review node for the article review process based on the provided state.
    """
   
    status = state.get("status", "")
    reviewer_comments = state.get("reviewer_comments", "")

    payload = {
        "status": status,
        "reviewer_comments": reviewer_comments,
    }

    #print("Human Review Node: Payload:", payload)

    #Execution pauses here for human input, which is simulated by the interrupt function. In a real-world scenario, this would involve user interaction.
    human_input =  interrupt(payload)

    return {
        "status": human_input.get("status", status),
        "reviewer_comments": human_input.get("reviewer_comments", reviewer_comments),
    }

def route_decision_node(state: ArticleReviewState):
    """
    Constructs a routing decision node to determine the next step in the article review process based on the provided state.
    """
    status = state.get("status", "")
    #print("Routing Decision Node: Current Status:", status)
    if status == "Approved":
        return "APPROVED"
    elif status == "Rejected":
        return "REJECTED_WITH_FEEDBACK"  # loops back toward build_prompt_node
    else:
        raise ValueError(f"Unexpected human_decision value: {status!r}")

async def mark_approved_node(state: ArticleReviewState, config: RunnableConfig):
    """
    Constructs a node to mark the article as approved based on the provided state.
    """
    configurable = config.get("configurable", {})
    db = configurable.get("db")
    status = "Approved"
    reviewer_comments = state.get("reviewer_comments", "") or ""
    article_id = state.get("article_id", "")
    ai_content = state.get("ai_content", "") or ""
    
    #print("Mark Approved Node: Article ID:", article_id, "Status:", status, "Reviewer Comments:", reviewer_comments)

    update_ai_article_with_status(db, article_id, status, reviewer_comments, ai_content.strip())

    return {
        "status": status,
        "reviewer_comments": state.get("reviewer_comments", ""),
    }

def capture_feedback_node(state: ArticleReviewState, config: RunnableConfig):
    """
    Constructs a node to capture feedback for the article based on the provided state.
    """
    configurable = config.get("configurable", {})
    db = configurable.get("db")
    status = "Rejected"
    reviewer_comments = state.get("reviewer_comments", "") or ""
    article_id = state.get("article_id", "")
    ai_content = state.get("ai_content", "") or ""

    #print("Capture Feedback Node: Article ID:", article_id, "Status:", status, "Reviewer Comments:", reviewer_comments)

    update_ai_article_with_status(db, article_id, status, reviewer_comments, ai_content.strip())


    return {
        "status": status,
        "reviewer_comments": state.get("reviewer_comments", ""),
        "iteration": state.get("iteration", 0) + 1,
    }

• Building the graph

builder = StateGraph(ArticleReviewState)

builder.add_node("BUILD_ARTICLE_REVIEW_PROMPT", build_prompt_node)
builder.add_node("LLM_REVIEW", llm_correction_node)
builder.add_node("HUMAN_REVIEW", human_review_node)
builder.add_node("APPROVED", mark_approved_node)
builder.add_node("REJECTED_WITH_FEEDBACK", capture_feedback_node)

builder.add_edge(START, "BUILD_ARTICLE_REVIEW_PROMPT")
builder.add_edge("BUILD_ARTICLE_REVIEW_PROMPT", "LLM_REVIEW")
builder.add_edge("LLM_REVIEW", "HUMAN_REVIEW")

builder.add_conditional_edges("HUMAN_REVIEW", route_decision_node,{
 "APPROVED": "APPROVED",
 "REJECTED_WITH_FEEDBACK": "REJECTED_WITH_FEEDBACK",
})

builder.add_edge("APPROVED", END)
builder.add_edge("REJECTED_WITH_FEEDBACK", "BUILD_ARTICLE_REVIEW_PROMPT") 

• Invoking the graph : We are using ainvoke as we are using async methods in some of the nodes.

async def run_article_review_workflow(article: ArticleResponse, db:Session):
 """
 Executes the article review workflow for a given article.
 """
 config = {"configurable": {"thread_id": article.id, "db": db}}
 initial_state: ArticleReviewState = {
 "article_id": str(article.id),
 "raw_content": article.content,
 
 "status": article.status,
 "iteration": 0,
 }
 async with AsyncPostgresSaver.from_conn_string(raw_db_url) as checkpointer:
 graph = builder.compile(checkpointer=checkpointer)
 await graph.ainvoke(initial_state, config=config)

• Resuming the graph : The operation gets resumed by the Command resume

async def resume_article_review_workflow(article: StatusUpdateRequest, db:Session):
 """
 Resumes the article review workflow for a given article.
 """
 config = {"configurable": {"thread_id": article.articleId, "db": db}}
 initial_state: ArticleReviewState = {
 "article_id": str(article.articleId),
 
 
 "status": article.status,
 "iteration": 0,
 }
 async with AsyncPostgresSaver.from_conn_string(raw_db_url) as checkpointer:
 graph = builder.compile(checkpointer=checkpointer)
 result = await graph.ainvoke(
 Command(
 resume={
 "status": article.status,
 "reviewer_comments": article.comments if article.comments else "",
 }
 ),
 config=config,
 )

 return result

Conclusion

This application demonstrates a simple checkpoint and HITL using Postgres and langgraph. If you want to see the source code of the above example, refer to the repository here.