langgraph/how-tos/map-reduce/

[giscus[bot]]

langgraph/how-tos/map-reduce/

Build language agents as graphs

https://langchain-ai.github.io/langgraph/how-tos/map-reduce/

[TaoZC1996]

I used this case, but I got an error on the line "return [Send("generate_joke", {"subject": s}) for s in state['subjects']] " that says "Callable() takes no arguments". How should I solve this?

[hwchase17]

hm i just reran and it worked fine for me. what version of langgraph are you using, whats your environment setup?

[TaoZC1996]

my langgraph version is 0.0.64, and all langchain related dependencies have been upgraded to the latest version, but when I rerun I still encounter this problem... 0.0

[MR-GREEN1337]

I have a question regrading parallel execution, when two nodes or more are executed in parallel, with all the same state, changes are made at the same time to the same state, how do you aggregate the unified the state at the end?

[pcliupc]

I guess the update of the state is lock protected.

[hinthornw]

Does this help? https://langchain-ai.github.io/langgraph/how-tos/branching/#stable-sorting

Basically "it depends" on what your state looks like and how you want parallel operations to be applied. If it's an unordered set, then maybe you're fine just inserting values. If it's a list and order matters, then you'll have to think about what makes sense in your application context.

The updates themselves are applied at the end of the superstep

[MR-GREEN1337]

It helps a lot thanks 👍🏻

[DylanZeigler]

What if I wanted to pause on a parallel state and resume execution due to external factors. How would I reference the parallel state to continue execution? Is that possible now or a future feature?

[etdufay]

Hello,

Thanks for great work !

While returning the output of the instance-processed function (here generate_joke), I got the following error and I don't understand why it does happen:
line 1741, in _apply_writes
for chan, val in pending_writes:
ValueError: too many values to unpack (expected 2)

Here is what I'm returning:
return {"jokes": ["Hello world"]}

Where "jokes" is an attribute of the overall class.

May it be a compatibility issue ? Or do you have any idea ?

Thanks

[MR-GREEN1337]

what does your graph state look like?

[jojohannsen]

That use of conditional edges with 'Send' I'm having trouble understanding. When would you need to use 'Send' in a conditional edge vs. replacing that conditional edge with a node that does the same thing?

def generate_jokes(state):
    jokes = []
    for subject in state["subjects"]:
        response = generate_joke({"subject": subject})
        jokes += response["jokes"]
    return {"jokes": jokes}

joker = StateGraph(OverallState)
joker.add_node('generate_topics', generate_topics)
joker.add_node('generate_jokes', generate_jokes)
joker.add_node('best_joke', best_joke)

joker.set_entry_point('generate_topics')

joker.add_edge('generate_topics', 'generate_jokes')
joker.add_edge('generate_jokes', 'best_joke')
joker.add_edge('best_joke', END)

joker = joker.compile()```

[jojohannsen]

ok, i think i get it, using Send is more understandable than making the node loop run in parallel. What I find confusing is what the conditional edge function returns. In code examples the conditional edge function is returning strings that map to nodes. In this example, I'm not sure what 'continue_to_jokes' is returning.

[Jaid844]

Hi I have used SEND class but I think it might have a problem with tool, when a different node sends objects to a node with a tool this raises an error
InvalidUpdateError: Invalid update for channel branch:agent:route_tools:tools with values ['agent', 'agent']

[iamq-dev]

I found a case where the model returning the joke index was not using 0 based indexing.

➜ py -m graph.pll-joke-graphv2
2024-08-13 00:29:07 -      LLM setup
2024-08-13 00:29:08 -       Azure Deployment = test
2024-08-13 00:29:08 -       Azure Api = OrderedDict()
2024-08-13 00:29:09 -  asking about birds
2024-08-13 00:29:10 -   generate topics prompt= Generate a comma separated list of between 2 and 5 examples related to: birds.
2024-08-13 00:29:11 -   generate topics response= subjects=['birds']    <---- my topic is passing through as the only subject. Still debugging. 
2024-08-13 00:29:11 -    generate topics return= {'subjects': ['birds']}
2024-08-13 00:29:11 -     cont to jokes return =[Send(node='generate_joke', arg={'subject': 'birds'})]
2024-08-13 00:29:11 -   generate joke prompt= Generate a joke about birds
2024-08-13 00:29:13 -   generate joke response= joke="Why don't birds use Facebook? Because they already tweet enough!"
2024-08-13 00:29:13 -    generate joke return= {'jokes': ["Why don't birds use Facebook? Because they already tweet enough!"]}
2024-08-13 00:29:13 -   best joke prompt= Below are a bunch of jokes about birds. Select the best one! Return the ID of the best one.

Why don't birds use Facebook? Because they already tweet enough!
2024-08-13 00:29:14 -   best joke response= id=1             <------ Here it got a list of 1, responded with id=1, but the array is 0 based so caused an error. 
Traceback (most recent call last):
  File "<frozen runpy>", line 198, in _run_module_as_main
  File "<frozen runpy>", line 88, in _run_code
  File "V:\langchain\graph\pll-joke-graphv2.py", line 135, in <module>
    app.invoke({"topic":"birds"})
  File "V:\langchain\.venv\Lib\site-packages\langgraph\pregel\__init__.py", line 1312, in invoke
    for chunk in self.stream(
  File "V:\langchain\.venv\Lib\site-packages\langgraph\pregel\__init__.py", line 997, in stream
    _panic_or_proceed(done, inflight, loop.step)
  File "V:\langchain\.venv\Lib\site-packages\langgraph\pregel\__init__.py", line 1398, in _panic_or_proceed
    raise exc
  File "V:\langchain\.venv\Lib\site-packages\langgraph\pregel\executor.py", line 60, in done
    task.result()
  File "C:\Users\q\.pyenv\pyenv-win\versions\3.12.0\Lib\concurrent\futures\_base.py", line 449, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "C:\Users\q\.pyenv\pyenv-win\versions\3.12.0\Lib\concurrent\futures\_base.py", line 401, in __get_result
    raise self._exception
  File "C:\Users\q\.pyenv\pyenv-win\versions\3.12.0\Lib\concurrent\futures\thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "V:\langchain\.venv\Lib\site-packages\langgraph\pregel\retry.py", line 25, in run_with_retry
    task.proc.invoke(task.input, task.config)
  File "V:\langchain\.venv\Lib\site-packages\langchain_core\runnables\base.py", line 2876, in invoke
    input = context.run(step.invoke, input, config, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "V:\langchain\.venv\Lib\site-packages\langgraph\utils.py", line 102, in invoke
    ret = context.run(self.func, input, **kwargs)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "V:\langchain\graph\pll-joke-graphv2.py", line 117, in best_joke
    temp= {"best_selected_joke": state["jokes"][response.id]}
                                 ~~~~~~~~~~~~~~^^^^^^^^^^^^^
IndexError: list index out of range

0:29:14
➜ py -m graph.pll-joke-graphv2
2024-08-13 00:35:44 -       the tool add was used.
2024-08-13 00:35:45 -      LLM setup
2024-08-13 00:35:46 -       Azure Deployment = test
2024-08-13 00:35:46 -       Azure Api = OrderedDict()
2024-08-13 00:35:47 -  asking about birds
2024-08-13 00:35:47 -   generate topics prompt= Generate a comma separated list of between 2 and 5 examples related to: birds.
2024-08-13 00:35:49 -   generate topics response= subjects=['birds']
2024-08-13 00:35:49 -    generate topics return= {'subjects': ['birds']}
2024-08-13 00:35:49 -     cont to jokes return =[Send(node='generate_joke', arg={'subject': 'birds'})]
2024-08-13 00:35:49 -   generate joke prompt= Generate a joke about birds
2024-08-13 00:35:51 -   generate joke response= joke="Why do we never see birds working at a computer? Because they're afraid of Windows!"
2024-08-13 00:35:51 -    generate joke return= {'jokes': ["Why do we never see birds working at a computer? Because they're afraid of Windows!"]
}
2024-08-13 00:35:51 -   best joke prompt= Below are a bunch of jokes about birds. Select the best one! Return the ID of the best one as an int u
sing 0 based indexing.     <------  appended the joke prompt with this to fix. 

Why do we never see birds working at a computer? Because they're afraid of Windows!
2024-08-13 00:35:52 -   best joke response= id=0
2024-08-13 00:35:52 -    best joke return= {'best_selected_joke': "Why do we never see birds working at a computer? Because they're afraid of Windows!"}

[Vansh290802]

Can there be 2 map-reduce branches in one Langgraph?

[freshchen]

HI，I want to know if there is any difference between use Send api and runnable.abatch operation in Node.

[mediumreach]

Hi can you show a more simpler example code like the one which you made for human in the loop, i seem to be getting lost in how send works and how graph.add_conditional_edges works in this scenario and can multiple send be used.

[rayshen92]

If there are dependencies between decomposed tasks, how should it be implemented?

[SuperElectron]

Hey all,

Two questions:

1. Why does the graph from this line of python code look different than the example in javascript?
2. I can't get an output with more than one subject. Why?

---

Question 1
Why does the graph from this line of python code look different than the example in javascript?

• In the python example, here, the edges connect directly connect each node, completely linear.

Image(app.get_graph().draw_mermaid_png())

• javascript example of map-reduce
• in the javascript example, link above, one has the same linear line, then dotted lines, i.e. edges, that connect the 'generate topic' node to the other nodes below.

import * as tslab from "tslab";

const representation = app.getGraph();
const image = await representation.drawMermaidPng();
const arrayBuffer = await image.arrayBuffer();

tslab.display.png(new Uint8Array(arrayBuffer));

---

Question 2
I can't get an output with more than one subject. Why?

import operator
from typing import Annotated, TypedDict

from langchain_openai import ChatOpenAI
#from langchain_anthropic import ChatAnthropic

from langgraph.constants import Send
from langgraph.graph import END, StateGraph, START

# NOTE:
# - if you're using langchain-core >= 0.3, you need to use pydantic v2
# - if you're using langchain-core >= 0.2,<0.3, you need to use pydantic v1
from langchain_core import __version__ as core_version
from packaging import version

core_version = version.parse(core_version)
if (core_version.major, core_version.minor) < (0, 3):
    from pydantic.v1 import BaseModel, Field
else:
    from pydantic import BaseModel, Field

model = ChatOpenAI(model="gpt-4o", temperature=0, api_key=OPEN_AI_KEY)

# Model and prompts
# Define model and prompts we will use
subjects_prompt = """Generate a comma separated list of between 2 and 5 examples related to: {topic}."""
joke_prompt = """Generate a joke about {subject}"""
best_joke_prompt = """Below are a bunch of jokes about {topic}. Select the best one! Return the ID of the best one.

{jokes}"""


class Subjects(BaseModel):
    subjects: list[str]


class Joke(BaseModel):
    joke: str


class BestJoke(BaseModel):
    id: int = Field(description="Index of the best joke, starting with 0", ge=0)


# Graph components: define the components that will make up the graph


# This will be the overall state of the main graph.
# It will contain a topic (which we expect the user to provide)
# and then will generate a list of subjects, and then a joke for
# each subject
class OverallState(TypedDict):
    topic: str
    subjects: list
    # Notice here we use the operator.add
    # This is because we want combine all the jokes we generate
    # from individual nodes back into one list - this is essentially
    # the "reduce" part
    jokes: Annotated[list, operator.add]
    best_selected_joke: str


# This will be the state of the node that we will "map" all
# subjects to in order to generate a joke
class JokeState(TypedDict):
    subject: str


# This is the function we will use to generate the subjects of the jokes
def generate_topics(state: OverallState):
    prompt = subjects_prompt.format(topic=state["topic"])
    response = model.with_structured_output(Subjects).invoke(prompt)
    return {"subjects": response.subjects}


# Here we generate a joke, given a subject
def generate_joke(state: JokeState):
    prompt = joke_prompt.format(subject=state["subject"])
    response = model.with_structured_output(Joke).invoke(prompt)
    return {"jokes": [response.joke]}


# Here we define the logic to map out over the generated subjects
# We will use this an edge in the graph
def continue_to_jokes(state: OverallState):
    # We will return a list of `Send` objects
    # Each `Send` object consists of the name of a node in the graph
    # as well as the state to send to that node
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]


# Here we will judge the best joke
def best_joke(state: OverallState):
    jokes = "\n\n".join(state["jokes"])
    prompt = best_joke_prompt.format(topic=state["topic"], jokes=jokes)
    response = model.with_structured_output(BestJoke).invoke(prompt)
    return {"best_selected_joke": state["jokes"][response.id]}


# Construct the graph: here we put everything together to construct our graph
graph = StateGraph(OverallState)
graph.add_node("generate_topics", generate_topics)
graph.add_node("generate_joke", generate_joke)
graph.add_node("best_joke", best_joke)
graph.add_edge(START, "generate_topics")
graph.add_conditional_edges("generate_topics", continue_to_jokes, ["generate_joke"])
graph.add_edge("generate_joke", "best_joke")
graph.add_edge("best_joke", END)
app = graph.compile()

# Call the graph: here we call it to generate a list of jokes
for s in app.stream({"topic": "animals"}):
    print(s)

and the output

{'generate_topics': {'subjects': ['animals']}}
{'generate_joke': {'jokes': ["Why don't elephants use computers? Because they're afraid of the mouse!"]}}
{'best_joke': {'best_selected_joke': "Why don't elephants use computers? Because they're afraid of the mouse!"}}

[atanudasgupta]

change the prompt to something like this
subjects_prompt = """Generate a comma separated list of between 2 and 5 distinct joke subjects eluded in the {topic}. ."""

[itsihsan]

How can the SEND approach be applied in this case?

I have annual reports for companies A and B, each with its dedicated retrieval tool: retrieval-tool-A for Company A and retrieval-tool-B for Company B. When a user asks, "What were the revenues of Company A and B in the past year?" the typical tools-assisted agent splits this into two queries: "What were the revenues for Company A in the past year?" (handled by retrieval-tool-A) and "What were the revenues for Company B in the past year?" (managed by retrieval-tool-B). These answers are then combined to provide the final response.

Can this process be parallelized using the SEND approach? Since the two queries are independent and use separate tools, running them in parallel could significantly reduce response time and enhance the robustness and scalability of tool-assisted agentic methods.

[mjunaidca]

Why not create branches for parallel node execution like this?
https://langchain-ai.github.io/langgraph/how-tos/branching/

And if in your ai-workflow you let an llm to decide the most relevant companies out of the list then using SEND makes more sense. This allows for dynamic decision-making, where the LLM can assess and prioritize which companies to include in the queries, ensuring that your retrieval tools are utilized effectively.

[curiothinker]

I got an error using ollama llama3.1 with chatopenai. But not with a paid openai model. How can I modifiy to overcome this error?

ValidationError Traceback (most recent call last)
Cell In[4], line 2
1 # Call the graph: here we call it to generate a list of jokes
----> 2 for s in app.stream({"topic": "animals"}):
3 print(s)

File ~/miniconda3/envs/Langgraph/lib/python3.10/site-packages/langgraph/pregel/init.py:1298, in Pregel.stream(self, input, config, stream_mode, output_keys, interrupt_before, interrupt_after, debug, subgraphs)
1287 # Similarly to Bulk Synchronous Parallel / Pregel model
1288 # computation proceeds in steps, while there are channel updates
1289 # channel updates from step N are only visible in step N+1
1290 # channels are guaranteed to be immutable for the duration of the step,
1291 # with channel updates applied only at the transition between steps
1292 while loop.tick(
1293 input_keys=self.input_channels,
1294 interrupt_before=interrupt_before_,
1295 interrupt_after=interrupt_after_,
1296 manager=run_manager,
1297 ):
-> 1298 for _ in runner.tick(
1299 loop.tasks.values(),
1300 timeout=self.step_timeout,
1301 retry_policy=self.retry_policy,
1302 get_waiter=get_waiter,
1303 ):
...

ValidationError: 1 validation error for Subjects
subjects
Input should be a valid list [type=list_type, input_value='["dog", "cat", "lion", "elephant"]', input_type=str]
For further information visit https://errors.pydantic.dev/2.9/v/list_type

[alexlyan]

Question:

I want to modify this flow where I add an additional node (retrieve_jokes - retrieving similar jokes) between the "generate_subject" and "generate_joke" nodes. The new node should retrieve related jokes based on the subject and feed them into the joke generation process. Here’s the flow I am envisioning:

Generate Subject → New Node: Retrieve similar jokes from a vector DB by the subject.

Generate Joke for Each Retrieved similar jokes: Generate a new joke for each retrieved document using the subject as context.

Reduce: Choose the best joke from the generated jokes.

# Construct the graph: here we put everything together to construct our graph
graph = StateGraph(OverallState)
graph.add_node("generate_topics", generate_topics)
# new NODE
graph.add_node("retrieve_similar_jokes", retrieve_jokes)
graph.add_node("generate_joke", generate_joke)
graph.add_node("best_joke", best_joke)
graph.add_edge(START, "generate_topics")
graph.add_conditional_edges("generate_topics", continue_to_retrieve, ["retrieve_jokes"])

# new edge
graph.add_edge("retrieve_jokes", "generate_joke")

graph.add_edge("generate_joke", "best_joke")
graph.add_edge("best_joke", END)
app = graph.compile()

I need to extend functionality to save states for every subject (evaluator or self-reflection might be added for node to generate joke). should be dynamic due to uncountable number of subject.

How to modify this flow to get this kindda graph?

[alexlyan]

I solved it by wrapping retrieval and adding other nodes (generate_joke) as sub graph with its own state.

# sub graph
retrieve_sub_graph = StateGraph(RetrieveState)
retrieve_sub_graph.add_node("retrieve_similar_jokes", retrieve_jokes)
retrieve_sub_graph.add_node("generate_joke", generate_joke)

retrieve_sub_graph.add_edge(START, "retrieve_similar_jokes")
retrieve_sub_graph.add_edge("retrieve_similar_jokes", "generate_joke")
retrieve_sub_graph.add_edge("generate_joke", END)

# core graph
graph = StateGraph(OverallState)
graph.add_node("generate_topics", generate_topics)
graph.add_node("retrieve", retrieve_sub_graph.compile())
graph.add_node("best_joke", best_joke)


graph.add_edge(START, "generate_topics")
graph.add_conditional_edges("generate_topics", continue_to_retrieve, ["retrieve"])

graph.add_edge("retrieve", "best_joke")
graph.add_edge("best_joke", END)
app = graph.compile()

[alexazhou]

some body know how to disable parallel, it there a option can control this feature enable or disable ?

In some time, I need branches execute in sequence.

thanks

[ericzon]

Try passing max_concurrency = 1 as parameter in config.

[jamesheavey]

Is there a way to manage the maximum concurrency of the parallel branches? I need to limit it in line with the rate limit of my LLM saas api provider

[eyurtsev]

You can set max_concurrency in the runnable config when invoking the graph.

graph.invoke(..., { "max_concurrency": ...)

You can read more here: https://python.langchain.com/docs/concepts/runnables/#runnableconfig

Alternatively, if you're doing this specifically to rate limit a chat model and use langchain chat models, you can use a rate limiter: https://python.langchain.com/docs/how_to/chat_model_rate_limiting/

[ericzon]

In this particular case, in which of the invokes would be better to add that max_concurrency param, in:

    response = model.with_structured_output(Joke).invoke(prompt)

or directly in .stream method?

for s in app.stream({"topic": "animals"}, { "max_concurrency": 2}):
    print(s)

[VirajVaitha123]

Thanks for this, it's really useful to see how Send API works.

However, I ran the exact same code, yet I only see one joke?

{'generate_topics': {'subjects': ['animals']}}
{'generate_joke': {'jokes': ["Why don't some animals play cards in the wild? Because there are too many cheetahs!"]}}
{'best_joke': {'best_selected_joke': "Why don't some animals play cards in the wild? Because there are too many cheetahs!"}}

Anyone else facing the same issue?