Workflow with multi branch and a final end node problem

[Fnck]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

def _prepare_next_tasks(
    checkpoint: Checkpoint,
    processes: Mapping[str, PregelNode],
    channels: Mapping[str, BaseChannel],
    managed: ManagedValueMapping,
    config: RunnableConfig,
    step: int,
    *,
    for_execution: bool,
    manager: Union[None, ParentRunManager, AsyncParentRunManager] = None,
) -> tuple[Checkpoint, Union[list[PregelTaskDescription], list[PregelExecutableTask]]]:
...

Description

we are using langgraph v0.0.55, and here is the problem:

• we have a workflow with 2 branches just like start -> A -> B -> end, start -> C -> D -> end, the branches share a same end node here.
• when the workflow is completed, the final end node will be run twice
• I just check the internal func _prepare_next_tasks in venv/Lib/site-packages/langgraph/pregel/init.py line 1540, there seems have no depency check when the code prepare the next task
  So the here is the problem: where can I find the dependency related logic in the code, and is the scenario of end node were run twice is expected?
  Thank you

System Info

langgraph v0.0.55

[dosubot[bot]]

Hey there, @Fnck! I'm here to help you with any bugs, questions, or contributions. Feel free to ask me anything while you wait for a human maintainer.

The issue you're encountering with the end node being executed twice in a workflow with multiple branches converging into a single end node is likely related to how dependencies are managed and checked within the PregelNode and TaskExecutor classes.

1. Dependency Management: The PregelNode class handles task dependencies through the TaskExecutor class, specifically in the check_dependency method. This method checks if all dependencies of a given task are either completed or not failed/pending:

   def check_dependency(self, task: Task) -> bool:
       for dep_id in task.dep:
           if dep_id == -1:
               continue
           dep_task = self.id_task_map[dep_id]
           if dep_task.failed() or dep_task.pending():
               return False
       return True

2. Task Execution: To prevent a task from being executed multiple times, the TaskExecutor class checks if a task is pending and if its dependencies are met before running it:

   def run(self) -> str:
       for task in self.tasks:
           print(f"running {task}")  # noqa: T201
           if task.pending() and self.check_dependency(task):
               self.update_args(task)
               task.run()
       if self.completed():
           self.status = "completed"
       elif self.failed():
           self.status = "failed"
       else:
           self.status = "pending"
       return self.status

3. Unit Tests: There are existing unit tests that cover workflows with multiple branches converging into a single end node. For example, the test_schema_complex_seq function demonstrates a workflow where multiple branches converge into a single end node:

   def test_schema_complex_seq() -> None:
       prompt1 = ChatPromptTemplate.from_template("what is the city {person} is from?")
       prompt2 = ChatPromptTemplate.from_template(
           "what country is the city {city} in? respond in {language}"
       )
   
       model = FakeListChatModel(responses=[""])
   
       chain1: Runnable = RunnableSequence(
           prompt1, model, StrOutputParser(), name="city_chain"
       )
   
       assert chain1.name == "city_chain"
   
       chain2: Runnable = (
           {"city": chain1, "language": itemgetter("language")}
           | prompt2
           | model
           | StrOutputParser()
       )
   
       assert chain2.input_schema.schema() == {
           "title": "RunnableParallel<city,language>Input",
           "type": "object",
           "properties": {
               "person": {"title": "Person", "type": "string"},
               "language": {"title": "Language"},
           },
       }
   
       assert chain2.output_schema.schema() == {
           "title": "StrOutputParserOutput",
           "type": "string",
       }

4. Potential Causes: Specific conditions within the _call function could lead to a task being scheduled more than once. The get_next_task method generates new tasks based on the result of the executed task, and these new tasks are added to the task_list. If get_next_task generates a task that already exists in the task_list, it could lead to the same task being scheduled more than once:

   new_tasks = self.get_next_task(
       result,
       task["task_name"],
       objective,
       callbacks=_run_manager.get_child(),
   )
   for new_task in new_tasks:
       self.task_id_counter += 1
       new_task.update({"task_id": self.task_id_counter})
       self.add_task(new_task)
   self.task_list = deque(
       self.prioritize_tasks(
           this_task_id, objective, callbacks=_run_manager.get_child()
       )
   )

By ensuring that dependencies are correctly checked and managed, and by verifying that tasks are not added multiple times to the task list, you can prevent the end node from being executed twice ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[Fnck]

Since this is an AI generated answer, I will post some discover here:

1. The functions which were mentioned by the answer are not exist, I just searched the langgraph code;
2. there should have no mechanisms for dependency check in PregelNode as I know

So this discussion should still be open.

[Fnck]

It's closed.