langgraph/how-tos/branching/

[giscus[bot]]

langgraph/how-tos/branching/

Build language agents as graphs

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

[HUNZALAMUSHTAQ]

Hi i am implementing this parallel tool calling but it raises exception
openai.BadRequestError: Error code: 400 - {'error': {'message': "An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id'. The following tool_call_ids did not have response messages: call_i1R8a78yTqv2IpDGUve2xPNZ", 'type': 'invalid_request_error', 'param': 'messages', 'code': None}}

mermaid syntax for diagram
graph TD;
start[start]:::startclass;
end[end]:::endclass;
primary_assistant([primary_assistant]):::otherclass;
primary_assistant_tools([primary_assistant_tools]):::otherclass;
enter_order_assistant([enter_order_assistant]):::otherclass;
order_assistant([order_assistant]):::otherclass;
order_assistant_tools([order_assistant_tools]):::otherclass;
leave_skill([leave_skill]):::otherclass;
enter_supply_assistant([enter_supply_assistant]):::otherclass;
supply_assistant([supply_assistant]):::otherclass;
supply_tools([supply_tools]):::otherclass;
start --> primary_assistant;
enter_order_assistant --> order_assistant;
enter_supply_assistant --> supply_assistant;
leave_skill --> primary_assistant;
order_assistant_tools --> order_assistant;
primary_assistant_tools --> primary_assistant;
supply_tools --> supply_assistant;
primary_assistant -.-> enter_order_assistant;
primary_assistant -.-> enter_supply_assistant;
primary_assistant -.-> primary_assistant_tools;
primary_assistant -.-> end;
order_assistant -.-> order_assistant_tools;
order_assistant -.-> leave_skill;
order_assistant -.-> end;
supply_assistant -.-> leave_skill;
supply_assistant -.-> supply_tools;
supply_assistant -.-> end;
classDef startclass fill:#ffdfba;
classDef endclass fill:#baffc9;
classDef otherclass fill:#fad7de;

[hinthornw]

@HUNZALAMUSHTAQ , if you want to fan out then back in and then pass back to an LLM, you need to ensure the message history contains a properly ordered list of messages, including a tool message for each tool call the AI might have.

[ibrahim77gh]

Hi William , thanks for the reply we really appreciate it!

This is how we are maintaining the Message History, we followed the Customer Support Bot example.
We are getting the error on parallel tool call (when the same tool is called multiple times in parallel) by the agent

Can you please provide a little more guidance, We would highly appreciate it,
Thanks.

from typing import Callable, Annotated, Literal, Optional
from langgraph.graph.message import AnyMessage, add_messages
from typing_extensions import TypedDict

def update_dialog_stack(left: list[str], right: Optional[str]) -> list[str]:
"""Push or pop the state."""
if right is None:
return left
if right == "pop":
return left[:-1]
return left + [right]

class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
user_info: str
dialog_state: Annotated[
list[
Literal[
"assistant",
"order_assistant",
"supply_assistant",
]
],
update_dialog_stack,
]

This is how we are maintaining the

[ibrahim77gh]

@hinthornw

[gengyabc]

Thanks for these wonderful guildes. I got a quesiton on the state update.
In the last guides on Create subgraphs, the states got some duplicates, but here there is no duplicates after fan-in even though the reduce function is just "add".

So what is happening under the hood?

[LordO54]

When it comes to executing langchain chains, is it possible to execute parallel calling with these graph structure?

[LordO54]

I have been trying to execute this kind of structure via streaming_events but it does not work. the answer is always waited to be finished, before showing it:

try:
inputs = {"question": data.message, "sessionid": session_id, "user_data":user_data}
response_data = []
async for event in langgraph_app.astream_events(inputs, version="v2"):
print("empezando")
kind= event["event"]
tags = event.get("tags", [])
if "langgraph_node" in event["metadata"]:
if event["metadata"]["langgraph_node"]=="generateCommon":
if "input" in event["data"] and "generation" in event["data"]["input"]:
data = event["data"]["input"]["generation"][0].content
print(data)
if data:
response_data.append(data)
await manager.send_personal_message("".join(response_data), websocket)

[zetyquickly]

May I ask regarding the "Parallel node fan-out and fan-in with extra steps", what is the point of having parallel paths if both branches share the same state? I mean if "C" sees not only "A", but also the change made in "B". I mean in here:

Adding I'm A to []
Adding I'm B to ["I'm A"]
Adding I'm C to ["I'm A"]
Adding I'm B2 to ["I'm A", "I'm B", "I'm C"]
...

[jacopo-chevallard]

@rlancemartin I tried to modify the map-reduce example to introduce a further step (refine_jokes) along the branch before the reduce operation. The goal would be to be able to branch out (using Send) and have several steps in each branch (executed in parallel) before the reduce operation. I guess that this can be achieved using sub-graphs, but perhaps there is a simpler solution?

The error I'm getting is

    raise InvalidUpdateError(
langgraph.errors.InvalidUpdateError: At key 'joke': Can receive only one value per step. Use an Annotated key to handle multiple values.

Here my code

import operator
from typing import Annotated, TypedDict

from langchain_anthropic import ChatAnthropic

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

from pydantic import BaseModel, Field

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

refine_joke_prompt = """Refine the following joke to make it even funnier: {joke}."""


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)


model = ChatAnthropic(model="claude-3-5-sonnet-20240620")

# 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
    joke: str


# This is the function we will use to generate the subjects of the jokes
def generate_topics(state: OverallState) -> 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) -> JokeState:
    prompt = joke_prompt.format(subject=state["subject"])
    response = model.with_structured_output(Joke).invoke(prompt)
    return {"joke": [response.joke]}

def refine_joke(state: JokeState) -> OverallState:
    prompt = refine_joke_prompt.format(joke=state["joke"])
    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) -> 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]}


def compile_graph():
    # 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("refine_joke", refine_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", "refine_joke")
    graph.add_edge("refine_joke", "best_joke")
    graph.add_edge("best_joke", END)
    app = graph.compile()
    return app


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

[TomTom101]

Have you been able to solve this? I'm still struggling with exactly the same problem. Thanks!

[EliezerIsrael]

Also looking at the same issue. It's clear that "refine_joke" is acting as the sink - there's nothing in the graph structure to distinguish it from "best_joke" - but it's not clear how to keep the second step in each mapped branch isolated.

[EliezerIsrael]

If I understand correctly, the add_edge method adds an edge to a single node identified by its name. In your case, you want a different node on each branch. I believe that you can use the Send method again, at the second step, to create that node dynamically. In other words, the pattern currently used in continue_to_jokes will work for the edge between generate_joke and refine_joke.

Remove the explicit edge
graph.add_edge("generate_joke", "refine_joke")
and replace it with
graph.add_conditional_edges("generate_joke", continue_to_refine, ["refine_joke"])

Add a function:

def continue_to_refine(state: JokeState):
    return Send("refine_joke", state)

[EliezerIsrael]

In my case, I want each mapped branch to use tools with something like a ReAct architecture. I haven't cracked that one yet.

[tianyi-fca]

I found an interesting unexpected behaviour:

In the basic fan-out and fan-in case, the method 1 and method 2 shown below do exactly the same thing.

// add nodes to the graph
workflow
  .addNode("A", nodeA)
  .addNode("B", nodeB)
  .addNode("C", nodeC)
  .addNode("D", nodeD)
  // add edges to the graph

  // *** fan out from Node A to B and C and then fan in to D ***
  .addEdge(START, "A")
  .addEdge("A", "B")
  .addEdge("A", "C")

  // >>> method 1 <<<
  .addEdge("B", "D")
  .addEdge("C", "D")

  // >>> method 2 <<<
  // .addEdge(["B", "C"], "D")

  .addEdge("D", END);

Output from method 1 and method 2 are identical

Adding A to 
Adding B to A
Adding C to A
Adding D to A,B,C
{ aggregate: [ 'A', 'B', 'C', 'D' ] }

However, in the fan-out and fan-in with extra steps, method 1 and method 2 (below) give different output

workflow
  .addNode("a", nodeA.call.bind(nodeA))
  .addNode("b", nodeB.call.bind(nodeB))
  .addNode("b2", nodeB2.call.bind(nodeB2))
  .addNode("c", nodeC.call.bind(nodeC))
  .addNode("d", nodeD.call.bind(nodeD))
  .addEdge(START, "a")
  .addEdge("a", "b")
  .addEdge("b", "b2")
  .addEdge("a", "c")

  // >>> method 1 (right way) <<<
  .addEdge(["b2", "c"], "d")

  // method 2 (wrong way) - The following will add one extra `I'm D` to the aggregate
  // .addEdge("b2", "d")
  // .addEdge("c", "d")

  .addEdge("d", END);

The output:

// from method 1
Adding I'm A to 
Adding I'm B to I'm A
Adding I'm C to I'm A
Adding I'm B2 to I'm A,I'm B,I'm C
Adding I'm D to I'm A,I'm B,I'm C,I'm B2
{ aggregate: [ "I'm A", "I'm B", "I'm C", "I'm B2", "I'm D" ] }

// From method 2
Adding I'm A to 
Adding I'm B to I'm A
Adding I'm C to I'm A
Adding I'm B2 to I'm A,I'm B,I'm C
Adding I'm D to I'm A,I'm B,I'm C
Adding I'm D to I'm A,I'm B,I'm C,I'm B2,I'm D
{
  aggregate: [ "I'm A", "I'm B", "I'm C", "I'm B2", "I'm D", "I'm D" ]
}

[freshchen]

In Parallel node fan-out and fan-in with extra steps。

It seems that b2 needs to be executed after c is completed

I expect the execution of b and b2 to be independent of c. Should I use subgraph?

[Huarong]

I used subgraph and got real parallel successfully.

[freshchen]

Thanks. it seems that multi steps parallel only subgraph can be used.
I feel a little uncomfortable splitting the graph for technical reasons rather than business reasons

[ibrahim-omer-621]

Is it possible to define how each Node transforms the state rather than defining it globally within the global Graph State?

[ashantanu]

another interesting behaviour, if i do

builder.add_edge("b2", "d")
builder.add_edge("c", "d")

instead of

builder.add_edge(["b2", "c"], "d")

then d is executed twice and doesn't wait for both to complete

[khw11044]

I have solved this problem.

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

https://hyundoil.tistory.com/383

[ganeshghag]

parallel paths for execution entails, any given instance of workflow may have 2 or more current nodes of execution and therefore 2 or more rejoin-points,
if each parallel branch had a human-in-loop, then to continue the workflow instance, how will we specify which parallel branch to continue?
does agentexecutor invoke function support specifying branch-id?

[djprawns]

Correct me if im wrong, but I tested out the below code with some time delay to see whether the parallel nodes (which do run parallely) run in a fashion that if one finishes then it moves ahead to the next node. But I observed that the nodes at a particular parallel layer wait until both finished. Why is this happening. You can observe this in the following example. How can make two branches run in a way that they run parallely and "asynchronously".

import time
import random
import operator
from typing import Annotated, Any

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END

import operator
from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph


class State(TypedDict):
    # The operator.add reducer fn makes this append-only
    aggregate: Annotated[list, operator.add]


class ReturnNodeValue:
    def __init__(self, node_secret: str):
        self._value = node_secret

    def __call__(self, state: State) -> Any:
        print(f"Adding {self._value} to {state['aggregate']}")
        time.sleep(random.random())
        return {"aggregate": [self._value]}


builder = StateGraph(State)
builder.add_node("a", ReturnNodeValue("I'm A"))
builder.add_edge(START, "a")
builder.add_node("b", ReturnNodeValue("I'm B"))
builder.add_node("b2", ReturnNodeValue("I'm B2"))
builder.add_node("c", ReturnNodeValue("I'm C"))
builder.add_node("d", ReturnNodeValue("I'm D"))
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "b2")
builder.add_edge(["b2", "c"], "d")
builder.add_edge("d", END)
graph = builder.compile()

graph.invoke({"aggregate": []})