LangGraph는 LangChain 생태계의 최신 확장으로, 복잡한 다중 에이전트 시스템과 상태 기반 워크플로우를 구축하기 위한 강력한 프레임워크를 제공합니다. 이 도구는 그래프 구조를 활용하여 LLM 기반 애플리케이션의 상태 관리, 조건부 실행, 그리고 에이전트 간 협업을 가능하게 함으로써, 기존 LangChain의 선형적 체인 구조를 넘어선 더욱 정교한 AI 시스템 구축을 지원합니다.
LangGraph의 상태 관리
LangGraph의 핵심 기능 중 하나는 복잡한 워크플로우에서 상태를 효과적으로 관리하는 능력입니다. 이는 `StateGraph` 클래스를 통해 구현되며, 이 클래스는 그래프의 각 노드 간에 상태를 전달하고 유지하는 메커니즘을 제공합니다.
상태 관리의 핵심 요소는 다음과 같습니다:
1. 상태 정의: `TypedDict`를 사용하여 상태의 구조를 명시적으로 정의합니다. 예를 들어:
class GraphState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
이 정의는 메시지 리스트를 상태의 일부로 포함하며, `operator.add`를 사용하여 메시지를 누적하는 방식을 지정합니다.
2. 상태 초기화: 그래프 실행 시 초기 상태를 제공합니다. 이는 워크플로우의 시작점을 설정하는 데 사용됩니다.
3. 상태 업데이트: 각 노드는 현재 상태를 입력으로 받아 처리하고, 업데이트된 상태를 반환합니다. 이 과정에서 상태의 일부 또는 전체를 수정할 수 있습니다.
4. 상태 전파: 그래프 구조에 따라 업데이트된 상태가 다음 노드로 전달됩니다. 이를 통해 워크플로우 전체에 걸쳐 일관된 컨텍스트를 유지할 수 있습니다.
5. 조건부 상태 처리: `add_conditional_edges` 메서드를 사용하여 상태에 기반한 조건부 실행 흐름을 정의할 수 있습니다. 예를 들어:
graph.add_conditional_edges(
"node_a",
lambda state: state["condition"],
{True: "node_b", False: "node_c"}
)
이 코드는 `state["condition"]`의 값에 따라 다음 실행 노드를 결정합니다.
6. 지속성: `SqliteSaver`와 같은 체크포인터를 사용하여 상태를 영구적으로 저장하고 복원할 수 있습니다. 이는 장기 실행 워크플로우나 중단 후 재개가 필요한 시나리오에서 특히 유용합니다.
7. 비동기 상태 관리: LangGraph는 비동기 실행을 지원하며, 이는 `astream` 메서드를 통해 구현됩니다. 이 방법은 상태 업데이트를 실시간으로 스트리밍할 수 있게 해줍니다.
LangGraph의 상태 관리 시스템은 복잡한 AI 워크플로우에서 컨텍스트를 유지하고, 동적인 의사 결정을 가능하게 하며, 장기적인 작업의 일관성을 보장하는 데 중요한 역할을 합니다. 이를 통해 개발자는 더욱 정교하고 적응력 있는 AI 시스템을 구축할 수 있습니다.
LangGraph 컴포넌트 구조
LangGraph의 컴포넌트 구조는 모듈화된 설계를 통해 복잡한 AI 워크플로우를 효율적으로 구현할 수 있게 해줍니다. 주요 컴포넌트들은 다음과 같습니다:
- StateGraph: 전체 워크플로우의 구조를 정의하는 핵심 클래스입니다. 노드와 엣지를 포함하며, 상태 관리를 담당합니다.
- Node: 개별 작업 단위를 나타내며, Python 함수로 구현됩니다. 현재 상태를 입력으로 받아 처리하고 업데이트된 상태를 반환합니다.
- Edge: 노드 간의 연결을 정의하며, 조건부 실행을 지원합니다. `add_conditional_edges` 메서드를 통해 구현됩니다.
- State: TypedDict를 사용하여 정의되는 글로벌 상태 객체입니다. 워크플로우 전체에서 공유되는 데이터를 저장합니다.
- Tools: LangChain의 도구들과 통합되어 외부 API 호출, 데이터 검색 등의 기능을 제공합니다.
- Checkpointer: `SqliteSaver`와 같은 클래스를 사용하여 워크플로우의 상태를 저장하고 복원합니다.
- EventHandler: 워크플로우 실행 중 발생하는 이벤트를 처리하는 컴포넌트입니다. `stream` 및 `astream` 메서드를 통해 실시간 이벤트 처리를 지원합니다.
이러한 컴포넌트들은 유기적으로 연결되어 작동합니다. 예를 들어, StateGraph 내의 Node들은 공유 State를 통해 데이터를 교환하며, Edge를 통해 실행 흐름이 제어됩니다. EventHandler는 워크플로우 전반에 걸쳐 발생하는 이벤트를 모니터링하고 필요한 경우 개입할 수 있습니다.
LangGraph의 이러한 모듈화된 구조는 복잡한 AI 시스템을 더 쉽게 설계, 구현, 유지보수할 수 있게 해줍니다. 특히 대화형 AI 에이전트, 다단계 추론 시스템, 적응형 학습 알고리즘 등의 구현에 효과적으로 활용될 수 있습니다.
LangGraph 구성요소들 정리
LangGraph의 주요 구성요소들은 노드(Node), 엣지(Edge), 그래프(Graph), 그리고 상태(State)입니다. 이들은 복잡한 LLM 기반 워크플로우를 구축하는 데 사용되며, 기초 코스셋에서 다음과 같이 활용됩니다:
1. 노드(Node):
노드는 워크플로우의 개별 작업 단위를 나타냅니다. 각 노드는 Python 함수로 정의되며, 현재 상태를 입력으로 받아 처리하고 업데이트된 상태를 반환합니다.
예시:
def research_node(state: AgentState):
query = state['task']
results = search_engine.search(query)
return {"research_results": results}
graph.add_node("research", research_node)
이 예시에서 `research_node`는 주어진 작업에 대해 검색을 수행하고 결과를 상태에 추가합니다.
2. 엣지(Edge):
엣지는 노드 간의 연결을 정의하며, 데이터 흐름의 방향을 나타냅니다. 조건부 엣지를 사용하여 동적 워크플로우를 구현할 수 있습니다.
예시:
graph.add_edge("research", "write")
graph.add_conditional_edges(
"evaluate",
lambda state: state["quality_score"] > 0.8,
{True: "finalize", False: "revise"}
)
이 코드는 'research' 노드에서 'write' 노드로의 직접 연결과, 'evaluate' 노드에서 품질 점수에 따라 'finalize' 또는 'revise' 노드로 분기하는 조건부 연결을 정의합니다.
3. 그래프(Graph):
그래프는 전체 워크플로우 구조를 나타내며, 노드와 엣지를 포함합니다. `StateGraph` 클래스를 사용하여 정의됩니다.
예시:
from langgraph.graph import StateGraph
graph = StateGraph(AgentState)
# 노드와 엣지 추가
graph.set_entry_point("start_node")
이 코드는 `AgentState` 타입의 상태를 사용하는 새로운 그래프를 생성하고, 시작 노드를 설정합니다.
4. 상태(State):
상태는 워크플로우 전체에서 공유되는 데이터 구조입니다. 일반적으로 `TypedDict`를 사용하여 정의됩니다.
예시:
from typing import TypedDict, Annotated
from langgraph.graph import add_messages
class AgentState(TypedDict):
task: str
messages: Annotated[list[AnyMessage], add_messages]
research_results: list
draft: str
quality_score: float
이 상태 정의는 작업 설명, 메시지 기록, 연구 결과, 초안 내용, 그리고 품질 점수를 포함합니다. `add_messages` 어노테이션은 메시지 리스트에 새 메시지를 누적하도록 지시합니다.
실제 사용 예시:
graph = StateGraph(AgentState)
graph.add_node("research", research_node)
graph.add_node("write", write_node)
graph.add_node("evaluate", evaluate_node)
graph.add_node("revise", revise_node)
graph.add_node("finalize", finalize_node)
graph.add_edge("research", "write")
graph.add_edge("write", "evaluate")
graph.add_conditional_edges(
"evaluate",
lambda state: state["quality_score"] > 0.8,
{True: "finalize", False: "revise"}
)
graph.add_edge("revise", "evaluate")
graph.set_entry_point("research")
app = graph.compile()
final_state = app.invoke({
"task": "Write an essay about artificial intelligence",
"messages": [],
"research_results": [],
"draft": "",
"quality_score": 0.0
})
이 예시는 에세이 작성 워크플로우를 구현합니다. 연구, 작성, 평가, 수정, 최종화 단계를 거치며, 품질 점수에 따라 수정 또는 최종화로 분기합니다. 이러한 구조는 복잡한 AI 작업을 모듈화하고 유연하게 관리할 수 있게 해줍니다.
LangGraph의 주요 구성요소와 그 관계를 시각화한 구조도를 분석하면 다음과 같은 핵심 요소들의 역할과 상호작용을 파악할 수 있습니다:
- 노드(Node):
- 원형으로 표현되며, 각각 독립적인 작업 단위를 나타냅니다.
- 예: '데이터 수집', '분석', '결과 생성' 등의 기능을 수행합니다.
- - 각 노드는 Python 함수로 구현되며, 현재 상태를 입력으로 받아 처리하고 업데이트된 상태를 출력합니다.
- 엣지(Edge):
- 노드를 연결하는 화살표로 표현됩니다.
- 데이터 흐름과 실행 순서를 정의합니다.
- 조건부 엣지를 통해 상태에 따라 다른 경로로 워크플로우를 분기할 수 있습니다.
- 상태(State):
- 중앙의 큰 사각형으로 표현되며, 워크플로우 전체에서 공유되는 데이터 구조입니다.
- TypedDict를 사용하여 정의되며, 모든 노드가 접근하고 수정할 수 있는 글로벌 컨텍스트 역할을 합니다.
- 작업 진행 상황, 중간 결과 등을 저장하여 워크플로우의 일관성을 유지합니다.
- 그래프(Graph):
- 이미지 전체가 하나의 그래프를 나타내며, 노드, 엣지, 상태를 포함하는 전체 워크플로우 구조입니다.
- StateGraph 클래스로 정의되며, 워크플로우의 시작점과 종료 조건을 설정합니다.
LangGraph의 실행 흐름은 다음과 같습니다:
1. 초기화: 그래프의 시작 노드에서 초기 상태를 설정합니다.
initial_state = {"task": "Write an essay", "messages": [], "draft": ""}
2. 노드 실행: 시작 노드가 현재 상태를 입력으로 받아 작업을 수행합니다.
def research_node(state: dict) -> dict:
query = state["task"]
results = search_engine.search(query)
state["research_results"] = results
return state
3. 상태 업데이트: 노드의 작업 결과로 상태가 업데이트됩니다.
4. 엣지 평가: 업데이트된 상태를 기반으로 다음 실행할 노드를 결정합니다.
graph.add_conditional_edges(
"evaluate",
lambda state: state["quality_score"] > 0.8,
{True: "finalize", False: "revise"}
)
5. 다음 노드 실행: 선택된 엣지를 따라 다음 노드로 상태가 전달되고 실행됩니다.
6. 반복: 단계 2-5를 종료 조건이 만족될 때까지 반복합니다.
7. 종료: 최종 상태를 반환하고 워크플로우를 종료합니다.
이러한 구조를 통해 LangGraph는 복잡한 AI 워크플로우를 모듈화하고 유연하게 관리할 수 있습니다. 예를 들어, 에세이 작성 과정에서 연구, 초안 작성, 평가, 수정 등의 단계를 거치며 품질 점수에 따라 수정 또는 최종화로 분기하는 워크플로우를 쉽게 구현할 수 있습니다. 이는 특히 다단계 추론이 필요한 작업 자동화, 복잡한 대화형 AI 시스템 구축, 동적인 의사결정 트리 구현 등에 효과적으로 활용될 수 있습니다.
'IT 공부 > LangChain' 카테고리의 다른 글
| LangChain Agent (10) | 2024.10.30 |
|---|---|
| AI의 LangChain 설명 (15) | 2024.10.28 |