Langchain mulltiple edges error, Graph Parallel Processing

Error Message

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

My Graph

def CompilingGraph():
    workflow = StateGraph(AgentState)

    # Make keyword
    workflow.add_node("make_keyword", RunHCXRewrite)  # generate_prompt

    # Data Search
    workflow.add_node("retrieve_api_desc", RetriveAPIDesc)
    workflow.add_node("retrieve_api_title", RetriveAPITitle)
    workflow.add_node("retrieve_dl_title", RetriveDLTitle)
    workflow.add_node("retrieve_dl_desc", RetriveDLDesc)

    workflow.add_node("merge_data", MergeData)


    # Set entry point
    workflow.set_entry_point("make_keyword")

    # Docs Search Graph
    workflow.add_edge("make_keyword", "retrieve_api_desc")
    workflow.add_edge("make_keyword", "retrieve_api_title")
    workflow.add_edge("make_keyword", "retrieve_dl_title")
    workflow.add_edge("make_keyword", "retrieve_dl_desc")

    workflow.add_edge(["retrieve_api_desc", "retrieve_api_title", "retrieve_dl_title", "retrieve_dl_desc"], "merge_data")



    workflow.add_edge("merge_data", END)

    app = workflow.compile()

    return app

make_keyword 에서 4가지 node로 이어지는 그래프를 만들고자 했음 (그래프 병렬처리)

State

class AgentState(TypedDict):
    user_question: Annotated[str, "User Questions"] # 사용자 질문
    # category: str = None
    data_limit: Annotated[int, "data_limit"] = 48

    # =====================================
    #  rewrite using LLM
    # =====================================
    rewrited_question:  Annotated[list, "rewrited_question"] = None
    rewrited_limit:  Annotated[int, "rewrited_limit"] = 3

    keyword: Annotated[list, "keyword"] = None

    # =====================================
    #  Similarity Search
    # =====================================
    api_data_list_1_title: Annotated[list, "api_data_list_1_title"] = None 
    api_data_list_1_desc: Annotated[list, "api_data_list_1_desc"] = None
    api_data_list_2_title: Annotated[list, "api_data_list_2_title"] = None  
    api_data_list_2_desc: Annotated[list, "api_data_list_2_desc"] = None 

    datalake_data_list_1_title: Annotated[list, "datalake_data_list_1_title"] = None 
    datalake_data_list_1_desc: Annotated[list, "datalake_data_list_1_desc"] = None 
    datalake_data_list_2_title: Annotated[list, "datalake_data_list_2_title"] = None 
    datalake_data_list_2_desc: Annotated[list, "datalake_data_list_2_desc"] = None 


    api_data_id_list: Annotated[list, "api_data_id_list"] = None 
    datalake_data_id_list: Annotated[list, "datalake_data_id_list"] = None 

오류원인

오류가 나는 이유는 Multiple edge를 사용하였을 때, "state 가 통합되는 방법을 정의하지 않아서" 이다.

예를 들면
workflow.add_edge(["retrieve_api_desc", "retrieve_api_title", "retrieve_dl_title", "retrieve_dl_desc"], "merge_data")

위 코드에서 4개의 노드들이 merge_data 라는 하나의 노드로 통합되게 되는데,
여기서 state 의 각 요소들은 섞이게 된다.

retrieve_api_desc 노드에서는 A = [1, 2] 인데,
retrieve_api_title 노드에서는 A = [2, 3] 으로 변경되었다고 하자.

그렇게 state 들이 섞였을 때, 어떤 방식으로 최종 state 를 만들어 낼 것인가를 정해줘야 한다.

그래프가 일직선상으로 쭉 놓여있을 때는 상관없지만 (덮어씌워짐), 병렬처리를 할 때는 정의를 해줘야 에러가 안난다.

해결방법

https://langchain-ai.github.io/langgraph/concepts/low_level/?h=reduc#reducers

State 를 관리해주는 reducer 를 추가해주면 됨

custom reducer 를 만들어도 되고, langgraph 에서 제공하는 reducer 를 써도 됨

def reducer(left, right):
    if isinstance(left, int):
        if left <= right:
            return right
        else:
            return left
    elif isinstance(left, (str, list)):
        if len(left) <= len(right):
            return right
        else:
            return left
    elif isinstance(left, float): # Cacluate Time
        if left == right:
            return left
        else: 
            return left + right

.
.
.

    datalake_data_list_title: Annotated[list, reducer] = None 
    datalake_data_list_desc: Annotated[list, reducer] = None 

나 같은 경우는 int, str, list 일 때는 바뀐 값으로 덮어쓰는(None 에서 변경됨) reducer 를 정의했다.

그리고 state class 를 정의할 때 Annotated 를 사용해서 reducer 를 붙여준다. (얘는 이런식으로 변경돼~ 라고 정의)

left : 기존 값
right : 새로 들어온 값

return 한 값이 최종적으로 할당됨