🦜🕸️ LangGraph

LangGraph는 LangChain 생태계의 일부로, 복잡한 AI 워크플로우를 구축하기 위한 강력한 프레임워크
유연하고 확장 가능한 AI 애플리케이션을 비교적 쉽게 만들 수 있음

✅ 주요 특징

• 상태 기반 그래프: 복잡한 로직을 명확하게 표현
• 조건부 흐름: 동적인 의사 결정 프로세스 구현
• 모듈성: 재사용 가능한 컴포넌트로 워크플로우 구성
• 유지보수성: 모듈화된 구조로 개별 컴포넌트 수정 용이
• 확장성: 새로운 노드나 조건을 쉽게 추가 가능
• 디버깅: 각 단계별 상태 확인이 용이

✅ 워크플로우 구축 과정

1. 상태 정의
2. 노드 (함수) 정의
3. 그래프 생성
4. 노드 추가
5. 엣지 (연결) 정의
6. 시작점 설정
7. 그래프 컴파일

✅ 상세 구현 예제

1. 상태 정의

from typing import TypedDict, Annotated, Sequence, List
from langchain_core.documents.base import Document

class AgentState(TypedDict):
    actions: str
    context: Sequence[Document]
    answer: str
    question: str
    groundness_status: str

• AgentState는 워크플로우의 각 단계에서 사용되는 데이터를 정의
• 각 필드는 워크플로우의 특정 부분에서 필요한 정보를 나타냄

2. 노드 (함수) 정의

• 각 노드는 특정 작업을 수행하는 함수로 정의
• 이 글에는 노드 구현부 없음

def extract_actions(state: AgentState) -> AgentState:
    """액션 추출"""
    actions = "질문에서 추출된 액션들"
    return AgentState(actions=actions)

def is_exist_actions(state: AgentState) -> str:
    """액션 존재 여부 체크"""
    # is_exist_actions, is_not_exist_actions
    is_exist_actions = "is_exist_actions"
    return is_exist_actions

def is_exist_docs(state: AgentState) -> str:
    """검색 문서 존재 여부 체크"""
    # is_exist_docs, is_not_exist_docs
    is_exist_docs = "is_exist_docs"
    return is_exist_docs

def retrieve(state: AgentState) -> AgentState:
    """문서 검색"""
    docs = [Document(page_content="검색된 문서 1"), Document(page_content="검색된 문서 2")]
    return AgentState(context=docs)

def invoke_llm_with_context(state: AgentState) -> AgentState:
    """context와 함께 LLM 호출"""
    answer = "LLM 답변"
    return AgentState(answer=answer)

def invoke_llm(state: AgentState) -> AgentState:
    """LLM 호출"""
    answer = "LLM 답변"
    return AgentState(answer=answer)
    
def check_groundedness(state: AgentState) -> AgentState:
    """검색 문서와 LLM 답변의 적합성 확인"""
    # grounded, notGrounded, notSure
    groundness_status = "grounded" 
    return AgentState(groundness_status=groundness_status)
    
def judgement(state: AgentState) -> str:
    """노드 재호출 혹은 종료 판단"""
    # retry_retrieve, retry_invoke_llm_with_context, END
    judgement = END 
    return judgement

3. 그래프 생성

from langgraph.graph import StateGraph

workflow = StateGraph(AgentState)

4. 노드 추가

workflow.add_node("extract_actions", extract_actions)
workflow.add_node("retrieve", retrieve)
workflow.add_node("invoke_llm", invoke_llm)
workflow.add_node("invoke_llm_with_context", invoke_llm_with_context)
workflow.add_node("check_groundedness", check_groundedness)

5. 엣지 (연결) 정의

workflow.add_conditional_edges(
    "extract_actions",
    is_exist_actions,
    {
        "is_exist_actions": "retrieve",
        "is_not_exist_actions": "invoke_llm",
    }
)
workflow.add_conditional_edges(
    "retrieve",
    is_exist_docs,
    {
        "is_exist_docs": "invoke_llm_with_context",
        "is_not_exist_docs": "invoke_llm",
    }
)
workflow.add_edge("invoke_llm_with_context", "check_groundedness")
workflow.add_edge("invoke_llm", END)
workflow.add_conditional_edges(
    "check_groundedness",
    judgement,
    {
        "retry_retrieve": "retrieve",
        "retry_invoke_llm_with_context": "invoke_llm_with_context",
        END: END
    }
)

• 조건부 엣지를 사용하여 동적인 흐름 제어가 가능

6. 시작점 설정

workflow.set_entry_point("extract_actions")

7. 그래프 컴파일

app = workflow.compile()

✅ 액티비티 다이어그램

✅ 실행결과

app.invoke(AgentState(question="내 질문"))