Unexpected behaviour in conditional edges

[CyberKnight1803]

I am using LangGraph but trying to avoid using LangChain. Given below is my code that I am just experimenting -

Code

src/data_types.py

from typing import Union, TypedDict
from pydantic import BaseModel

from src.config import settings

class HumanMessage(BaseModel):
    content: str 

class AIMessage(BaseModel):
    content: str

class SummaryMessage(BaseModel):
    content: str

AnyMessage = Union[HumanMessage, AIMessage, SummaryMessage]

class Config(TypedDict):
    thread_id: str
    window_size: int = settings.SHORT_TERM_MEMORY_WINDOW_SIZE

src/state.py

from typing import List, Tuple, TypedDict, Annotated
from src.data_types import AnyMessage

def update_window(
    messages: List[AnyMessage], 
    updates: List[AnyMessage] | Tuple[List[AnyMessage], int]
) -> List[AnyMessage]:
    
    if isinstance(updates, list):
        print(f"MESSAGES: {messages}")
        print(f"UPDATES: {updates}")

        messages.extend(updates)
        return messages
    
    elif isinstance(updates, tuple):
        updated_messages, window_size = updates
        return [updated_messages[0]] + updated_messages[(-window_size + 1):]

    else:
        raise NotImplementedError("Invalid type for updates")

class NodeState(TypedDict):
    query: str
    short_term_memory: Annotated[List[AnyMessage], update_window] = []

src/graph.py

from typing import Dict, Any
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.runnables import RunnableConfig
from openai import OpenAI

from src.state import NodeState
from src.prompts import prompts
from src.data_types import HumanMessage, AIMessage, SummaryMessage, Config
from src.config import env

class Workflow:
    def __init__(self):
        self.openai_client = OpenAI(api_key=env.OPENAI_API_KEY)
        self.memory_saver = MemorySaver()

        # Build Graph
        graph_builder = StateGraph(NodeState)
        
        # Add Nodes
        graph_builder.add_node("preprocessor_node", self.preprocessor_node)
        graph_builder.add_node("chatbot_node", self.chatbot_node)
        graph_builder.add_node("summarize_conversation_node", self.summarize_conversation_node)
        
        # Add Edges
        graph_builder.add_edge(START, "preprocessor_node")
        graph_builder.add_edge("preprocessor_node", "chatbot_node")

        graph_builder.add_conditional_edges(
            "chatbot_node", 
            self.route_to_summarize_conversation_node,
            {
                "summarize_conversation_node": "summarize_conversation_node",
                END: END
            }
        )
        graph_builder.add_edge("summarize_conversation_node", END)

        
        
        # Compile graph
        self.graph = graph_builder.compile(checkpointer=self.memory_saver)

    def preprocessor_node(self, state: NodeState, config: RunnableConfig) -> Dict[str, Any]:
        if len(state["short_term_memory"]) == 0:
            return {
                "short_term_memory": [SummaryMessage(content="")]
            }
    
    def chatbot_node(self, state: NodeState, config: RunnableConfig) -> Dict[str, Any]:
        response = self.openai_client.chat.completions.create(
            messages=[
                {
                    "role": "system",
                    "content": prompts.get_system_prompt(),
                },
                {
                    "role": "user",
                    "content": prompts.get_user_prompt(
                        user_query=state["query"], 
                        conversation_history=state["short_term_memory"]
                    ),
                }
            ],
            model=env.LLM_MODEL
        )

        chatbot_response = response.choices[0].message.content

        return {
            "short_term_memory": [
                HumanMessage(content=state["query"]),
                AIMessage(content=chatbot_response)
            ]
        }

    def summarize_conversation_node(self, state: NodeState, config: RunnableConfig) -> Dict[str, Any]:

        if len(state["short_term_memory"]) > config["configurable"]["window_size"]:
            # The first message in short_term_memory is SummaryMessage
            summary_message = state['short_term_memory'][0]

            # The next two messages are the messages to summarize i.e. pair of (HumanMessage, AIMessage) which will be removed from short_term_memory afterwards.
            conv_to_summarize = state['short_term_memory'][1:3]

            # Summarize the conversation
            response = self.openai_client.chat.completions.create(
                messages=[
                    {
                        "role": "system",
                        "content": prompts.get_system_prompt(),
                    },
                    {
                        "role": "user",
                        "content": prompts.get_summary_prompt(
                            existing_summary=summary_message,
                            new_conversation=conv_to_summarize
                        ),
                    }
                ],
                model=env.LLM_MODEL
            )

            # Update SummaryMessage
            updated_summary = response.choices[0].message.content
            summary_message.content = updated_summary
            state['short_term_memory'][0] = summary_message

            return {
                "short_term_memory": (
                    state["short_term_memory"],
                    config["configurable"]["window_size"]
                )
            } 
    
    def route_to_summarize_conversation_node(self, state: NodeState, config: Config) -> str:
        if len(state["short_term_memory"]) == config["configurable"]["window_size"]:
            return "summarize_conversation_node"
        
        return END

    def run(self, query: str, config: Config) -> NodeState:
        state = NodeState(query=query)
        return self.graph.invoke(state, config={
            "configurable": config
        })
   

Graph

Issue

When I run the following code -

query = "Hi, Who are you?"
state = workflow.run(query, config)
print(state)

I see the following output -

MESSAGES: []
UPDATES: [SummaryMessage(content='')]

MESSAGES: [SummaryMessage(content='')]
UPDATES: [HumanMessage(content='Hi, Who are you?'), AIMessage(content="Hi there! I'm your friendly chatbot, here to help with any questions or topics you'd like to discuss. How can I assist you today?")]

MESSAGES: [SummaryMessage(content=''), HumanMessage(content='Hi, Who are you?'), AIMessage(content="Hi there! I'm your friendly chatbot, here to help with any questions or topics you'd like to discuss. How can I assist you today?")]
UPDATES: [HumanMessage(content='Hi, Who are you?'), AIMessage(content="Hi there! I'm your friendly chatbot, here to help with any questions or topics you'd like to discuss. How can I assist you today?")]

{'query': 'Hi, Who are you?',
 'short_term_memory': [SummaryMessage(content=''),
  HumanMessage(content='Hi, Who are you?'),
  AIMessage(content="Hi there! I'm your friendly chatbot, here to help with any questions or topics you'd like to discuss. How can I assist you today?"),
  HumanMessage(content='Hi, Who are you?'),
  AIMessage(content="Hi there! I'm your friendly chatbot, here to help with any questions or topics you'd like to discuss. How can I assist you today?")]}

The window_size is 5 so, it doesnt go in summarize_conversation_node.

As you can see the reducer function update_window has been called three times instead of two... Why is this happening?
Why is there an extra state update for short_term_memory happening?