how to send details from one node to multiple nodes in parallel

[arvindkrbhatt]

In LangGraph we add nodes and edges so that flow should be directed from one particular node to another particular node. Conditional edge takes the decision that which node to direct. Is it possible in LangGraph to direct details from one node to multiple nodes in parallel? how we can do it?

code

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

from langchain_openai import AzureChatOpenAI
# from langchain_community.utilities import OpenWeatherMapAPIWrapper
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 = AzureChatOpenAI(deployment_name="deployment",azure_endpoint="https://azope.openai.azure.com", openai_api_key="2e", api_version="2024-02-15-p",temperature=0, streaming=True)


def function_1(str1:str)->str:
    return str1

def function_2(str2:str)->str:
    return str2

def function_3(str3:str)->str:
    return str3

# def function_4(str4:str)->str:
#     return str4 + "final"

def function_4(str2:str, str3:str)->str:
    return str2+ " - " + str3

# Define a Langchain graph
workflow = Graph()

#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")

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

print(response)

[hinthornw]

Hello! It looks like you're almost there - the thing you'd want to make sure you handle well is that when the updates are returned from nodes 2 and 3 in your graph, the results don't overwrite each other (if they're writing to the same state key). So you typically either want to write to different keys or to make sure that the key in the state is annotated with a reducer that will handle the updates correctly.

You can read more about branching here: https://langchain-ai.github.io/langgraph/how-tos/branching/

A related API is the "Send" API which makes it easier to do map-reduce like work (where each parallel branch can see a different view/copy of the state), but this doesn't seem necessary for your intent.

[arvindkrbhatt]

Hi @hinthornw

Thanks a lot for your inputs.
I have some more doubts regarding the ask.

1. For given link it shows Node A is connected with Node B and Node C and further both Node B and Node C are connected with Node D. My question is does Node D wait for completion of both Node B and Node C or Node D will be executed if any of Node B or Node C executed successfully.
2. What will happen if any of Node B or Node C prompts error? Will Node D execute?
3. For a below code base that is implemented in LangChain (using Sequential chain) link how can I implement the same in LangGraph? What would be the best way to handle scenarios where some node is depending on inputs from other nodes?

I'll appreciate your valuable suggestions.

Link for [1] - https://langchain-ai.github.io/langgraph/how-tos/branching/#parallel-node-fan-out-and-fan-in
Code Link for [3] - https://www.analyticsvidhya.com/blog/2023/10/a-comprehensive-guide-to-using-chains-in-langchain/

from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.prompts import ChatPromptTemplate
from langchain.chains import SequentialChain

llm = OpenAI(temperature=0.7, openai_api_key=OPENAI_API_KEY)
Review = "Les ordinateurs portables GamersTech impressionne par ses
 performances exceptionnelles et son design élégant. De sa configuration 
 matérielle robuste à un clavier RVB personnalisable et un système de 
 refroidissement efficace, il établit un équilibre parfait entre prouesses 
 de jeu et portabilité."

# prompt template 1: translate to English

first_prompt = ChatPromptTemplate.from_template(
    "Translate the following review to english:"
    "\n\n{Review}"
)

# chain 1: input= Review and output= English_Review

chain_one = LLMChain(llm=llm, prompt=first_prompt,
                     output_key="English_Review"                     
                    )

# prompt template 2: Summarize the English review

second_prompt = ChatPromptTemplate.from_template(
    "Can you summarize the following review in 1 sentence:"
    "\n\n{English_Review}"
)

# chain 2: input= English_Review and output= summary

chain_two = LLMChain(llm=llm, prompt=second_prompt,
                     output_key="summary"
                    )

# prompt template 3: translate to English

third_prompt = ChatPromptTemplate.from_template(
    "What language is the following review:\n\n{Review}"
)

# chain 3: input= Review and output= language

chain_three = LLMChain(llm=llm, prompt=third_prompt,
                       output_key="language"
                      )

# prompt template 4: follow up message

fourth_prompt = ChatPromptTemplate.from_template(
    "Write a follow up response to the following "
    "summary in the specified language:"
    "\n\nSummary: {summary}\n\nLanguage: {language}"
)

# chain 4: input= summary, language and output= followup_message

chain_four = LLMChain(llm=llm, prompt=fourth_prompt,
                      output_key="followup_message"
                     )

overall_chain = SequentialChain(
    chains=[chain_one, chain_two, chain_three, chain_four],
    input_variables=["Review"],
    output_variables=["English_Review", "summary","followup_message"],
    verbose=True
)
overall_chain(Review)

[arvindkrbhatt]

@ All
please help to answer above query.

[hinthornw]

For given link it shows Node A is connected with Node B and Node C and further both Node B and Node C are connected with Node D. My question is does Node D wait for completion of both Node B and Node C or Node D will be executed if any of Node B or Node C executed successfully.

• What will happen if any of Node B or Node C prompts error? Will Node D execute?

Does this help? (from the link above)

For a below code base that is implemented in LangChain (using Sequential chain) link how can I implement the same in LangGraph? What would be the best way to handle scenarios where some node is depending on inputs from other nodes?

The state is updated after every step, so by definition, nodes that are executed later will have access to the outputs of previous steps so long as:

1. Either the nodes write to different keys (aka channels) in the state
2. If they are writing to the same key, then your annotations/reducers make it so it is "append only" (similar to what we do in messages graph)

To get the same behavior as the linked code, you'd have a state with keys "English_Review", "summary", "followup_message", and each node would write to those separate keys.

[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']