Langraph fan out simultaneously - End one of the two branches if the exit condition has been met.

[kpratik41]

I have a complex graph and at one point in the graph I want the execution to happen simultaneously in 2 branches (I am aware of send api example in documentation). I am successfully able to execute things in both the branches and I am able to merge them properly as well to a single sink node. The state dictionary on both the branches have same keys and one of the keys is of type Annotated[list[str], operator.add].

The above part works fine.

I am trying another scenario where if one of the branches finds a solution then instead of going to the sink node it should go to the end node and the answer should be returned. I have implemented a conditional edge for this purpose and have tried all possible approaches but the execution never Ends instead waits for both the paths to complete.

In one scenario I made both the branches go towards END path instead of sink (indicating both paths found the right answer). Even for this scenario the execution is not stopping until both the path finish execution and reach END.

Has anyone come across an example/resource where graph execution stops and exits even when some other path in the graph was still running since the exit condition was met in one of the paths?

[kpratik41]

@vbarda Any suggestions?
Thank you

[hinthornw]

@kpratik41 so a bit like hedged requests?

We had discussed this a bit in the community slack

If you're streaming outputs, you can surface the completed node's output to the user while the other branch is still executing (so the user has a response even though the graph hasn't fully terminated)

We don't currently have a way in the langgraph DSL to do this in separate nodes.

You can do this in a single node using task cancellation, however:

import asyncio
import random


async def mock_api_request(name: str) -> str:
    delay = random.uniform(1, 5)
    await asyncio.sleep(delay)
    return f"Response from {name} after {delay:.2f} seconds"


async def main() -> None:
    names = ["API1", "API2", "API3"]
    tasks = [asyncio.create_task(mock_api_request(name), name=name) for name in names]

    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)

    for task in done:
        result = task.result()
        print(f"First completed task.  {result = }")

    for task in pending:
        task.cancel()
        print(f"{task.get_name()} is cancelled")

    await asyncio.gather(*pending, return_exceptions=True)


asyncio.run(main())

[TJKlein]

If you're streaming outputs, you can surface the completed node's output to the user while the other branch is still executing (so > the user has a response even though the graph hasn't fully terminated)

@hinthornw: I am exactly interested in this showing of some intermediate output of the faster branch, while the other branch is still running, and then potentially do the final generation when all branches are completed. How could I achieve this? Some example would be highly appreciated.