Why Branching requires Annotation in state in LangGraph

[arvindkrbhatt]

I am trying to achieve branching in LangGraph. When I create StateGraph without Annotating any number then it prompts below error

ValueError: Already found path for node 'node1'.
For multiple edges, use StateGraph with an annotated state key.

But the moment when I make any state graph member Annotated it works. Please help me to understand why Annotation requires in Branching? If I don't apply branching then it works without Annotation and no error as above,

Below is code

from langgraph.graph import Graph, StateGraph, END
from langchain_core.messages import HumanMessage

from langchain_openai import AzureChatOpenAI
from dotenv import load_dotenv
import os   

from langchain_core.utils.function_calling import convert_to_openai_function
from langchain_community.tools.openweathermap import OpenWeatherMapQueryRun
from langchain_core.utils.function_calling import convert_to_openai_function

from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage

from langgraph.prebuilt import ToolInvocation
import json
from langchain_core.messages import FunctionMessage
from langgraph.prebuilt import ToolExecutor


from langchain.globals import set_debug
from langchain.globals import set_verbose
 
# set_debug(True)
# set_verbose(True)

# model 

model = AzureChatOpenAI(deployment_name="deployment4T",azure_endpoint="https://azope.openai.azure.com", openai_api_key="27a", api_version="2024-02-15-preview",temperature=0, streaming=True)

class AgentState(TypedDict):
    str1: str #Annotated[str, operator.add]
    str2: str #Annotated[str, operator.add]
    str3: str #Annotated[str, operator.add]
    str4: str #Annotated[str, operator.add]
    # aggregate: Annotated[list, operator.add]


def function_1(str1:str)->str:
   
    return {"str1": "function_1"}
 


def function_2(str2:str)->str:
  
    return {"str2": "function_2"}
   
def function_3(str3:str)->str:
    
    return {"str3": "function_3"}
 

def function_4(str4:str)->str:
    
    return {"str4": "function_4"}
    
 

workflow = StateGraph(AgentState)

#calling node 1 as agent
workflow.add_node("node1", function_1)
workflow.add_node("node2", function_2)
workflow.add_node("node3", function_3)
workflow.add_node("node4", function_4)

workflow.add_edge('node1', 'node2')
workflow.add_edge('node1', 'node3')
workflow.add_edge('node2', 'node4')
workflow.add_edge('node3', 'node4')
# workflow.add_edge('node4', END)

workflow.set_entry_point("node1")
workflow.set_finish_point("node4")

app = workflow.compile()

# response = app.invoke("testseq")
# response = app.invoke({"aggregate": ["testseq"]})








# inputs = {"messages": ["what is the temperature in las vegas"]}
# response = app.invoke(inputs)

# print(response)

[hinthornw]

This is a great question! I'll try to update our docs in a couple places to better explain this, since it does seem non-obvious.

The short answer is that in general annotations of state keys (aka channels) define reducers, reducers let the graph know how to apply an update to the state. By default we don't really know the preferred way and so overwrite any channel when it is present in an update.

If no reducer is provided, and the graph has no branching, this is OK, since there is only ever one update to be applied in a given superstep.

However if there is branching, the update becomes undefined, since the StateGraph contract does not guarantee an ordering of the updates - if both branches write to the same key, we don't guarantee that the same ultimate state would result.

If you define a reducer, you can ID values or do whatever other logic you'd like to ensure consistent ordering.

Seems like we have a couple of TODOs from this:

1. Improve the docs to better explain this - maybe improve the error message as well
2. Consider API improvements to make it easier to handle this type of scenario

[YanSte]

@arvindkrbhatt

If you wish to proceed with your development without encountering any blockages.

in the class StateGraph(Graph):. You can simply comment out this code bellow:

class StateGraph(Graph):
    ...
    def __init__(self, schema: Type[Any]) -> None:
       ...

        # Here
        #if any(isinstance(c, BinaryOperatorAggregate) for c in self.channels.values()): 
        self.support_multiple_edges = True
        ...

[hinthornw]

Thanks for the rec! Though to anyone reading this, I would recommend against modifying the source code for anything but local experimentation.

[GonyRosenman]

where can i learn more about the logic of state updating?
i encountered this error when i tried to build a small graph that has an entry node that connects to two other nodes.

there is no chance that these nodes will update the same state parameter at the same time, but i tried to look deeper into the code and got confused by the many options there.

[jw782cn]

i wonder how you finally solve this? thx

[duanzhihua]

I also encountered a similar problem using Langgraph's StateGraph. I was planning to implement parallel connections for multiple nodes after one node, but an error occurred during runtime. Please help solve this problem.

Traceback (most recent call last):
  File "d:\LangGraph\main.py", line 32, in <module>    
    app = WorkFlow().app
  File "d:\LangGraph\src\genai_graph\graph.py", line 50, in __init__
    workflow.add_edge('generate_code', 'review_code')
  File "E:\ProgramData\anaconda3\envs\crewAI\lib\site-packages\langgraph\graph\state.py", line 349, in add_edge    
    return super().add_edge(start_key, end_key)
  File "E:\ProgramData\anaconda3\envs\crewAI\lib\site-packages\langgraph\graph\graph.py", line 187, in add_edge    
    raise ValueError(
ValueError: Already found path for node 'generate_code'.
For multiple edges, use StateGraph with an annotated state key.