Why is my State not being passed correctly in my LangGraph workflow?

Question

I have a simple series of nodes that are chained together with conditional edges. The first two are shown here:

def email_router(state: TypedDict) -> None:
    node_results, state = get_emails(state)
    if node_results == "New Email":
        state["internal_state"] = "topics_router"
    else:
        state["errors"] = "No New Mail"
        state["internal_state"] = "return_final_status"

def check_internal_state(state: TypedDict) -> str:
    logging.debug(f"Internal State: {state["internal_state"]}")
    return state["internal_state"]

In the email router, I am setting the state value "internal_state" to either "topics_router" or "return_final_status" which are two other nodes. The check_internal_state is there simply to enable the conditional edge which looks as follows:

workflow.add_conditional_edges(
    "email_router",
    check_internal_state,
    {
        "topics_router": "topics_router",
        "return_final_status": "return_final_status",
    },
)

The logging debug statement in the check_internal_state keeps returning an empty string which means that I am not properly carrying the state information across and I cannot figure out what I am doing wrong.

The code is set in classes. the one function compiling the code and saving the output as:

global mailman
mailman = workflow.compile()

I have another function, in the same python file, then invoking mailman and calling functions that all reside in another python file but that are imported. I only bring this up as maybe separating things into multiple functions is causing the issue? Here is the overall workflow:

workflow = StateGraph(GraphState)
workflow.add_node("email_router", email_router)
workflow.add_node("topics_router", topics_router)
workflow.add_node("status_router", status_router)
workflow.add_node("actions_router", actions_router)
workflow.add_node("return_final_status", return_final_status_node)

workflow.set_entry_point("email_router")

workflow.add_conditional_edges(
    "email_router",
    check_internal_state,
    {
        "topics_router": "topics_router",
        "return_final_status": "return_final_status",
    },
)

workflow.add_conditional_edges(
    "topics_router",
    check_internal_state,
    {
        "status_router": "status_router",
        "return_final_status": "return_final_status",
    },
)

workflow.add_conditional_edges(
    "status_router",
    check_internal_state,
    {
        "actions_router": "actions_router",
        "return_final_status": "return_final_status",
    },
)

workflow.add_edge("actions_router", "return_final_status")
workflow.add_edge("return_final_status", END)

Does anybody know what I am doing wrong? Thank you!

Answer 1

This ended up being my lack of understanding of how langgraph works and I figure I would explain things to help anybody else out who gets stuck.

Langgraph is literally creating a memory object for you called state. All nodes must return values matching whatever you want to persist from that node into the state object. For example, you might have "progress_step" in your state object. To keep the value set in your node, you would then return:

return {"progress_step": progress_step}

The returned JSON, which can include as many values as you have in your state object, is automatically used to update the global state used across all nodes.

This should not be confused with conditional routers that require a text output matching the name of any other node in your workflow. I label my nodes with _node and my routing functions with _routing to keep things separate.

I hope that helps anybody else that is confused!