해당 글은 유튜버 테디노트님의 강의를 보고 공부한 내용을 정리한 것입니다.
출처: 테디노트의 LangGraph 개념 완전 정복 몰아보기(3시간)
이번 시간에는 저번 시간에 이어서 LangGraph의 세부 기능에 대해 알아보고자 한다.
1. State(상태)
상태(state)
: 노드와 노드 간에 정보를 전달할 때 상태(state) 객체에 담아 전달한다.
- TypedDict: 일반 파이썬 dict에 타입힌팅을 추가한 개념이지만, 쉽게 dictionary로 생각하면 됨
- 모든 값을 다 채우지 않아도 됨
- 새로운 노드에서 값을 덮어쓰기(Overwrite) 방식으로 채움
- Reducer (add_messages 혹은 operator.add): 자동으로 list에 메시지를 추가해주는 기능
위 코드를 보면서 알아보자.
먼저 위와 같은 state는 노드와 노드 사이에 정보를 전달할 때 거기서 사용되는 키들을 미리 사전에 정의한 것들이다.
따라서 처음에 내가 노드를 만들 때 사용자가 질문하는 노드를 만들었다면 굳이 다른 키들의 값을 채우지 않고 question 값만 채우고 다음 노드로 넘겨주어도 상관없다.
TypedDict는 Python의 typing 모듈에서 제공하는 기능으로, 딕셔너리의 키와 값의 타입을 미리 정의할 수 있게 해준다.
마치 클래스를 정의하듯이 딕셔너리를 더 명확하게 사용할 수 있게 도와주는 타입 힌트이다.
우리가 일반적인 딕셔너리에 키와 값을 쌍으로 넣어주듯이 TypedDict도 마찬가지이다.
question이라는 키에는 list 형식의 값이 들어갈 수 있고
context라는 키에는 str 형식의 값이 들어갈 수 있다.
Annotated
앞 부분에 보면 Annotated라는 것을 볼 수 있는데 이것은 말 그대로 주석이다.
위 코드를 보면 context: Annotated[str, "Context"]가 적혀있는데 이는 context라는 키에는 str 형식의 값이 들어가면 되는데 이때 들어가는 값은 그냥 "Context"가 들어가면 된다는 의미이다.
add_messages
add_messages의 경우 리스트에 새 메시지를 추가할 수 있도록 한다.
즉 list에 저장될 데이터를 계속해서 추가할 수 있다는 의미로 add_messages의 경우 계속 추가하면서 늘려나갈 수 있는 list 형식에만 사용할 수 있다.
add_messages에 대해서 추가적으로 설명하자면
우리가 Human_msg가 있고 AI_msg가 있을 때 사람이 질문하고 AI가 답변하고를 반복하게 되면 AI가 답변할 때 이전 대화 내용을 알아야하는 경우가 존재한다.
이때 우리는 리스트에 .append를 사용해서 리스트에 새로운 내용을 추가하고 했는데, add_messages를 사용하게 되면 따로 .append를 사용할 필요 없이 반환 값에 리스트를 주기만 하면 Annotated[list, add_messages]가 붙어있기만 하면 자동으로 기존 리스트에 추가가 되는 개념이다.
랭그래프에서 이를 따로 만들어둔 이유는 쉽게 만들기 위함으로,
앞으로 너가 question이라는 키를 반환한다면 계속해서 리스트에 추가하도록 해줄게!
를 의미하게 된다.
만약 add_messages가 없고 Annotated[list] 이렇게 구성되어 있다면 다음과 같은 상황에서 리스트가 바로 교체되게 된다.
예를 들어 이전 리스트에 [ ( 1 ) , ( 2 ) ] 이 내용이 저장되어 있고, [ ( 3 ) ] 내용을 저장하려고 시도하게 되면 이전 리스트에 있던 [ ( 1 ) , ( 2 ) ] 내용이 사라지고 [ ( 3 ) ] 내용이 새롭게 교체되게 된다.
하지만 Annotated[list, add_messages]로 구성되어 있다면 [ ( 1 ) , ( 2 ) ] 상태에서 [ ( 3 ) ] 내용을 저장하게 되면 [ ( 1 ) , ( 2 ) , ( 3 ) ] 이렇게 수정된다.
상태에 대해서 조금 더 쉽게 말하면 위 사진을 보면 된다.
우리는 각 노드들이 개별적으로 움직이기 때문에 노드 1에서 진행한 과정에 대해서 노드 2, 3, 4는 전혀 알지 못한다.
따라서 노드 1이 진행하고 난 이후 결과를 위에서 말한 state를 사용해서 다음 노드에게 전달해주게 된다.
처음에는 모든 값이 비어있지만 노드 1에서 진행한 후 time과 llm 키의 값을 채우고 노드 2로 전달하게 된다.
노드 2도 마찬가지로 작동하고 끝난 결과를 state에 정리해서 넣어주고 이를 노드 3에게 전달한다.
그렇게 되면 결국 마지막 노드 4가 노드 1, 2, 3의 결과가 종합된 state를 받아서 처리하는 것이 가능해진다.
그럼 위에서 말한 사례를 RAG로 바꿔서 확인해보자.
위 사진은 state의 내용을 RAG로 바꾼 것으로
- 노드 1에서는 사용자 질문이 상태 값에 저장
- 노드 2에서는 노드 1에서 state에 저장한 질문을 활용해 리트리버를 실행하고, 검색된 문서를 상태 값에 저장
- 노드 3에서는 노드 2에서 state에 저장한 질문, 문서를 LLM에게 전달했을 때의 답변을 상태 값에 저장
- 마지막 노드 4에서는 노드 3에서 state에 문서, 질문, 답변을 가지고 평가를 진행하여, 결과를 score에 저장,
- 이때 score의 값이 BAD이기 때문에 조건부 엣지를 통해 노드 1, 2, 3 중 하나로 돌아가 다시 진행하도록 설정
2. 노드(Node)
- 함수로 정의
- 입력인자: 상태(State) 객체
- 반환(return)
- 대부분 상태(State) 객체
- Conditional Edge의 경우 다를 수 있음
우리가 2번 노드로 설명을 하자면 2번 노드는 1번 Question 노드가 끝나고 2번 노드에세 전달한 상태를 그대로 받아서 2번 노드를 실행하고 2번 노드의 결과를 상태에 다시 적게 된다.
그리고 저장된 상태를 다시 3번 노드에게 전달하게 되는데 이때 2번 노드에서 진행되는 과정은 모두 함수로 작성이 되며, 함수에서 다른 함수로 단순히 저장된 상태를 전달해준다고 생각하면 된다.
입력: 상태
출력: 상태
입력도 상태이고 출력도 상태인 것이 노드의 기본 형태이다.
노드는 입력으로 받은 상태에서 필요한 정보만을 꺼내서 사용하고, 이후에 결과를 다시 상태에 저장해서 다음 노드에게 전달하는 역할을 하게 된다.
노드 추가 방법
우리가 노드를 추가할 때는 add_node를 사용하면 된다.
사전에 구성한 2개의 함수(노드)가 존재한다는 가정하에
retrieve_document 함수, llm_answer 함수 총 2개가 존재한다고 가정
workflow = StateGraph(GraphState)
workflow.add_node("retrieve", retrieve_document)
workflow.add_node("llm_answer", llm_answer)
# workflowd.add_node("함수와 관련된 문자열(최대한 영어로)", 함수이름)위와 같은 방식으로 추가해주면 된다.
3. 엣지(Edge)
-
노드에서 노드간의 연결
-
add_edge("노드이름", "노드이름")
- from > to
위와 같이 우리가 노드를 만들게 되면
위 이미지와 같이 네모 모양의 노드들만 생성이 되고, 이제 우리는 이를 이어줄 수 있는 edge를 생성해주어야 한다.
엣지를 생성해주는 것도 어렵지 않다.
단순 .add_edge("from", "to") # from에서 to로 연결 방식을 활용하면 된다.
예를 들어
workflow.add_edge("retrieve", "llm_answer") # 검색 -> 답변
workflow.add_edge("llm_answer", "relevance_check") # 답변 -> 관련성 체크위와 같은 코드를 구성해주면 위 이미지에서 3개의 노드를 연결해준 엣지를 만들어준 것이다.
4. 조건부 엣지(Conditional Edge)
- 노드에 조건부 엣지를 추가하여 분기를 수행할 수 있다.
- add_conditional_edge("노드이름", 조건부 판단 함수, dict로 다음 단계 결정)
흐름
"relevance_check" 노드에서 나온 결과를 is_relevant 함수에 입력
반환된 값은 "grounded", "notGrounded", "notSure" 중 하나
- value에 해당하는 값이 END면 Graph 실행 종료
- "llm_answer"와 같이 노드이름이면 해당 노드로 연결
위 내용을 말로 풀어서 설명하면
"relevance_check" 함수를 실행한 결과를 is_relevant 함수에 입력한다.
is_relevant 함수에서 반환된 값이 "grounded"라면 END
반환된 값이 "notGrounded"라면 "llm_answer"노드로 이동
반환된 값이 "notSure"라면 "llm_answer" 노드로 이동이렇게 정리할 수 있다.
5. 시작점 지정
- set_entry_point("노드이름")
- 지정한 시작점부터 Graph가 시작함
# 시작점을 설정합니다.
workflow.set_entry_point("retrieve")위와 같이 코드를 구성하게 되면
사진과 같이 처음 시작 노드를 "retrieve" 노드로 구성해줄 수 있다.
6. 체크포인터(memory)
- Checkpointer: 각 노드간 실행결과를 추적하기 위한 메모리(대화에 대한 기록과 유사 개념)
- 체크포인터를 활용하여 특정 시점(Snapshot)으로 되돌리기 기능도 가능
- compile(checkpointer=memory) 지정하여 그래프 생성
# 기록을 위한 메모리 저장소를 설정
memory = MemorySaver()
# 그래프를 컴파일
app = workflow.compile(checkpointer=memory)위에서 사용된 메모리(MemorySaver())는 기억 저장 용도로 이 체크포인터라는 것을 활용해야 이전으로 돌아가던가, Snapshot(특정시점)으로 되돌아가는 것이 가능하다.
7. 그래프 시각화
- get_graph(xray=True).draw_mermaid_png()
- 생성한 그래프 시각화
8. 그래프 실행
- RunnableConfig
- recursion_limit: 최대 노드 실행 개수를 지정. (13인 경우: 총 13개의 노드까지 실행)
- thread_id: 그래프 실행 아이디를 기록하고, 추후 추적하기 위한 목적으로 활용
- 상태(State)로 시작
- 여기서 "question"에 질문만 입력하고 상태를 첫 번째 노드에게 전달
- invoke(상태, config) 전달하여 실행
위 코드를 예시로 보면 우리가 노드를 순환 구조로 만들었을 때, 즉 조건을 만족하지 못했을 경우 계속해서 다른 노드들로 이동하면서 반복되는 구조로 만들었을 때
RunnableConfig에서 recursion_limit을 설정해두게 되면 최대로 이동할 수 있는 노드의 갯수를 적어놓고 무한 반복이 되지 않도록 설정할 수 있다.
다음으로 configurable에서는 thread_id를 전달해줄 수 있는데, 이는 나중에 멀티 턴 대화를 할 때 이 thread_id 별로 따로 대화를 저장할 수 있도록 도와준다.
9. 결과 확인
- 출력된 결과로 최종 확인
- 출력 결과 역시 상태(State)에 담겨 있음
우리가 위에서 이야기 했듯이 노드는 state를 입력으로 전달 받고 출력 또한 state로 진행한다.
따라서 마지막 노드가 출력하는 결과 또한 state이기 때문에 우리는 이를 확인해서 결과로 보면 된다.
우리는 마지막 노드가 출력한 state를 받아서 state에 저장된 키를 살펴보게 되면은 결과를 확인할 수 있게 되는 것이다.
여기까지 가볍게 langgraph를 통해 노드를 구성하고 연결하고 확인할 수 있는 과정을 이론으로 알아보았다.
이제 다음에는 실제 코드를 통해서 확인해보는 과정을 가져보도록 하자.
