How to pass state that uses add operator to subgraph without duplicating it?

[arko-louisa]

Consider this minimal reproducible example:

# %%
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import END, START, StateGraph

# %%
class State(TypedDict):
    q: Annotated[list[str], add]
    answer: Annotated[list[str], add]
    docs: Annotated[list[str], add]

# %%
def put_docs(state):
    q = state["q"]
    return {
        "q": [],
        "answer": [],
        "docs": ["Some doc"]
    }

# %%
subgraph = StateGraph(State).add_node(put_docs).add_edge(START, "put_docs").add_edge("put_docs", END).compile()

# %%
def answer_node(state: State):
    q = state['q']
    return {
        "q": [],
        "answer": ["Answer"],
        "docs": []
    }

# %%
graph = (
    StateGraph(State)
    .add_node(answer_node)
    .add_node("subgraph", subgraph)
    .add_edge(START, "answer_node")
    .add_edge("answer_node", "subgraph")
    .add_edge("subgraph", END)
    .compile()
)

# %%
q = {"q": ["Hi"]}

print(graph.invoke(q))

This prints
{'q': ['Hi', 'Hi'], 'answer': ['Answer', 'Answer'], 'docs': ['Some doc']}

Is there a way to avoid this duplication?

[vbarda]

yes, you can write a custom reducer (instead of operator.add) that assigns an ID to the answer and then deduplicates based on the IDs, something similar to add_messages reducer https://github.com/langchain-ai/langgraph/blob/main/libs/langgraph/langgraph/graph/message.py#L18

[arko-louisa]

Interesting. Is the need for a custom reducer related to how each subgraph processes and modifies state independently, to avoid interfering with shared state "channels"? And following up: are there any plans to improve upon this in the recent future?

[wei-fh]

I have the same question, if subgraph can be treated as a node, this behavior is somewhat confusing because a plain node won't duplicate the output. Is a deduping reducer the official answer? Or is this an issue LangChain team would want to address?

[hinthornw]

Or the graph can always be called in a function. Example:

async def my_node(state):
    response = await subgraph.ainvoke(state)
    return {"messages": [response['messages'][-1]]}

Subraphs accept the (input) state schema as input and return the final end (output) state schema as a response.
If you don't want the parent to naively update its state based on the final end state of the subgraph, you have to either narrow what data is returned from the child subgraph or make the parent reducer smart enough to recognize and handle duplicates.

A plain node will absolutely duplicate outputs if you return the full state as the output. This is what happens if you put the subgraph directly in a graph. So that graphs have predictable semantics, graphs have the same response output whether they're invoked as a subgraph or as a standalone.

[hinthornw]

Or you can put the subgraph in a function and filter the inputs & outputs:

def handoff(state):
       output = subgraph.invoke(state)
       return {"docs" [output["docs"][-1]]}
StateGraph(State) 
.add_node(answer_node)
    .add_node("subgraph", handoff)
    .add_edge(START, "answer_node")
    .add_edge("answer_node", "subgraph")
    .add_edge("subgraph", END)
    .compile()