LangGraph multiple edges / For multiple edges, use StateGraph with an annotated state key.

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

 

LangGraph에서 add_edge로 node를 연결하다가 아래와 같은 오류가 발생하였다.

 

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

 

아무리 검색하고 찾아봐도 답이 보이지 않았지만 2시간 만에 문제를 해결하게 되었다. 

 

 

 

현재 이렇게 만들어야하는데

 

병렬 노드 실행을 위한 브랜치를 만드는데 무엇인가 잘못된것일까? 의심하였다.

아무리 찾아도 답이 없어서 LangGraph 공식 페이지에 들어가 보았다.

 

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

Create branches for parallel execution

 

 

예시코드가 있다. 그래도 모르겠다. 

 

import operator
from typing import Annotated, Any

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END


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']}")
        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("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", "d")
builder.add_edge("c", "d")
builder.add_edge("d", END)
graph = builder.compile()

 

이런 저런 고민을 하다가 

내 코드에는 Annotated를 사용하지 않았고 

 

Annotated state key라고 하였으니 State class에 Annotated를 추가해보기로 하였다. 

 

수정 전 

from typing import TypedDict


# GraphState 상태를 저장하는 용도로 사용합니다.
class GraphState(TypedDict):
    filepath: str  # path
    filetype: str  # pdf
    page_numbers: list[int]  # page numbers
    batch_size: int  # batch size
    split_filepaths: list[str]  # split files
    analyzed_files: list[str]  # analyzed files
    page_elements: dict[int, dict[str, list[dict]]]  # page elements
    page_metadata: dict[int, dict]  # page metadata
    page_summary: dict[int, str]  # page summary
    images: list[str]  # image paths
    image_summary: list[str]  # image summary
    tables: list[str]  # table
    table_summary: dict[int, str]  # table summary
    table_markdown: dict[int, str]  # table markdown
    texts: list[str]  # text
    text_summary: dict[int, str]  # text summary
    table_summary_data_batches: list[dict]  # table summary data batches
    language: str  # language

 

수정 후 

from typing import TypedDict, Annotated
import operator

# GraphState 상태를 저장하는 용도로 사용합니다.
class GraphState(TypedDict):
    filepath: Annotated[str, "filepath"]  # path
    filetype: Annotated[str, "filetype"]  # pdf
    page_numbers: Annotated[list[str], operator.add]  # page numbers
    batch_size: Annotated[int, "batch_size"]  # batch size
    split_filepaths: Annotated[list[str], "split_filepaths"]  # split files
    analyzed_files: Annotated[list[str], "analyzed_files"]  # analyzed files
    page_elements: Annotated[dict[int, dict[str, list[dict]]], "page_elements"]  # page elements
    page_metadata: Annotated[dict[int, dict], "page_metadata"]  # page metadata
    page_summary: Annotated[dict[int, str], "page_summary"]  # page summary
    images: Annotated[list[str], "images"]  # image paths
    image_summary: Annotated[list[str], "image_summary"]  # image summary
    tables: Annotated[list[str], "tables"]  # table
    table_summary: Annotated[dict[int, str], "table_summary"]  # table summary
    table_markdown: Annotated[dict[int, str], "table_markdown"]  # table markdown
    texts: Annotated[list[str], "texts"]  # text
    text_summary: Annotated[dict[int, str], "text_summary"]  # text summary
    table_summary_data_batches: Annotated[list[dict], "table_summary_data_batches"]  # table summary data batches
    language: Annotated[str, "language"]  # language

 

 

해결!!