Hello langgraph team,
i have a tool node which calls 2 downstream functions in parallel. In these parallel functions, i am trying different ways of ‘natural language analysis’ which i eventually want to save into a file.
The problem lies when i want to adapt the output directory (to which the files will be saved), based on which parallel branch called the ‘save file’ function.
The file saving function is common which will be called by both branches eventually. (this has the output_dir parameter which needs to be adapted based on which upstream parallel branch called it)
def save_clusters_to_files(state: dict, output_dir: str = "clusters_created") -> dict:
I need to understand ‘the upstream node that called the save file function’ so that i am able to pass/adapt the appropriate output directory parameter value.
OPTION 1 TRIED :
i have tried keeping a parameter in the state to track the last_tool. But this merges the ‘last used tool name’ from from both branches parallelly, hence it is not in sequence. So it is difficult for me to know which was the last node that was called.
class State(MessagesState):
last_tool: Annotated[list[str], operator.add]
if i dont use the operator.add, since both the branch functions are making some update to last_tool in state, i get the following error
langgraph.errors.InvalidUpdateError: At key ‘last_tool’: Can receive only one value per step. Use an Annotated key to handle multiple values.
OPTION 2 TRIED:
The messages in AIMessage are also not in sequence as they are merged from both the branches. So i cant decode this based on last message also
