Help me understand Reducers & Parallelization

[apsquared]

After watching the Intro to LangGraph video I thought I understood these concepts, but after building some code I'm confused. Any assistance to clarify would be helpful.

Here's my situation:

I have two fields in my state:

    subreddits: List[str] # Subreddits
    marketing_suggestions: List[str] # Marketing suggestions

I have two functions:

    async def get_subreddits(state: workflow_state):
    async def get_marketing_suggestions(state: workflow_state):

These are called in parallel:

    workflow.add_edge("finalize_competitors", "get_marketing_suggestions")
    workflow.add_edge("finalize_competitors", "get_subreddits")
    workflow.add_edge("get_subreddits", "__END__")
    workflow.add_edge("get_marketing_suggestions", "__END__")

In my first attempt I coded each of the function (similar but with a different key for the other function)

state['subreddits'] = theList
return state

When I run it I got an error:

langgraph.errors.InvalidUpdateError: At key 'appName': Can receive only one value per step. Use an Annotated key to handle multiple values.
For troubleshooting, visit: https://python.langchain.com/docs/troubleshooting/errors/INVALID_CONCURRENT_GRAPH_UPDATE

Since the keys were different of what I modified I thought I was OK, but I guess not.

So next I updated the definition to:

    subreddits: Annotated[List[str], operator.add] # Subreddits

Then changed the return to:

return {"subreddits": response.subreddits}

This caused the state to actually have duplicates of each element in the array.

The only combination that worked was defining without the annotation and return just changed item

    marketing_suggestions: List[str] # Marketing suggestions
....
        return {"marketing_suggestions": response.strategies}

This doesn't really make sense to me from what I read/heard in the video. Could someone help explain?

[vbarda]

Could you actually provide a full example? It's a bit hard to follow what exactly you're trying to return from each node running in parallel (assuming subreddits)

In general, this issue should only arise if both of the nodes running at the same superstep are trying to write an update to the same state key. In this case, LangGraph doesn't know how to apply the updates unless you provide a way for LangGraph to combine the values. TO do so, you can specify a reducer function in the state schema annotation: it tells LangGraph that instead of overwriting the values you need to accumulate them. So state value for the key after the nodes run will be the result of calling reducer(old_key_value, new_key_value).

If you're returning the same values from both nodes and have a reducer with operator.add, it's not surprising that you would see a duplicated list. If you need to only keep distinct values, you would probably need to use a set instead of a list.

I would recommend checking out these document pages:

• https://langchain-ai.github.io/langgraph/troubleshooting/errors/INVALID_CONCURRENT_GRAPH_UPDATE/
• https://langchain-ai.github.io/langgraph/concepts/low_level/#reducers

Hope this helps!

[apsquared]

Hi, I appreciate your comment. All of what you said makes sense but doesn't seem to line up with what I'm seeing

If you check these two files:
https://github.com/apsquared/lg-agents/blob/main/src/agents/marketing_agent/marketing_agent.py
https://github.com/apsquared/lg-agents/blob/main/src/agents/marketing_agent/marketing_schema.py

You will see in the _agent file functions get_marketing_suggestions and get_subreddits some commented code which is what I originally tried and gave an error. You'll also see in the schema file that the reducer on subreddits causes the duplicate.

If you look at get_subreddits I can confirm this function is only called once and the response has 5 elements, but then at the end of the graph it has 10.

Any guidance is really appreciated.

[vbarda]

first off, you shouldn't be manually mutating state inside node functions and return mutated state, this is an antipattern. you don't also need to return the full updated state from the node, only the relevant updates to the state. state updates should only be handled via outputs of node functions (or manual calls to graph.update_state()). e.g. this part

# state["marketing_suggestions"] = response.strategies
# return state

should always be done as

return {"marketing_suggestions": response.strategies}

for your specific issue, if BOTH get_subreddits and get_marketing_suggestions return marketing_suggestions key, then you need a reducer for that key. If both of them return the same list of marketing_suggestions and your reducer is a simple concatenation (e.g. operator.add), then it will be duplicated (so 5 elements will become 10). You need to customize the reducer to handle that if you must return marketing_suggestions from both get_subreddits and from get_marketing_suggestions nodes

i suspect that you were seeing some additional duplication because you were returning the full mutated state instead of only returning the updates

does this help clarify it at all?

[apsquared]

Helpful yes, but I don't think it explains what I'm seeing.

Specifically as the code currently stands get_subreddits is the only function that returns or manipulates subreddits.

    async def get_subreddits(state: workflow_state):
        print("#####GET SUBREDDITS#########")
        prompt = f"""
        Given the following information about {state['appName']} and its competitors, suggest 5 subreddits to monitor or post on for {state['appName']}.
            App description: {state['appDescription']}
            Key features: {state['keyfeatures']}
            Value proposition: {state['value_proposition']}
            Competitors: {state['competitors']}
        """
        llm = get_llm()
        structured_llm = llm.with_structured_output(SubredditList)
        response = structured_llm.invoke(prompt)
        #state["subreddits"] = response.subreddits
        #return state
        print(f"##### Found {len(response.subreddits)} subreddits")
        return {"subreddits": response.subreddits}
        ```

This is only called once and FOUND shows 5.

Yet when the state is printed at the end it has 10.

[vbarda]

Did you see my comments about state mutation / returning full state? Could you replace that everywhere and confirm that you're still seeing an issue?

[apsquared]

You nailed it. The culprit was return state int he end step which caused the reducer on that field to double it.

Thanks!

[vbarda]

Woohoo - i am glad it helped!