[LangGraph] 워크플로우(workflow)

 

langgraph 버전: 0.6.4

내용 출처 : LangGrapn Docs

LangGraph

 

1. 개요

LangGraph는 워크플로우를 구성할 때 두 가지 방식을 제공합니다.

• Graph API: 노드와 엣지를 명시적으로 정의하는 방식
• Functional API: 함수 호출 흐름 중심으로 간결하게 작성하는 방식

 

Graph API는 복잡한 흐름 제어, 조건 분기, 노드 재사용에 유리하고,

Functional API는 간단한 직렬 실행 흐름을 빠르게 작성하는 데 적합합니다.

 

비교 표

비교 항목	Graph API	Functional API
선언 방식	노드/엣지를 명시적으로 정의	함수 하나로 전체 흐름 구성
구조 복잡도	복잡한 조건 분기, 병렬 실행에 적합	단순 직렬 실행에 적합
확장성	높음	제한적
코드 길이	비교적 길고 구조적	짧고 직관적
학습 곡선	약간 높음	낮음

 

---

2. Graph API

2.1. Graph API 개념

• 노드(Node)와 엣지(Edge)를 명시적으로 추가하여 실행 흐름을 구성합니다.
• 각 노드는 함수, LLM 호출, 툴 실행 등 다양한 동작을 수행할 수 있습니다.
• 조건 분기, 병렬 실행, 서브그래프 등 복잡한 구조에 적합합니다.

 

---

2.2 예시코드, 코드설명

1. State 정의: 그래프 전체에서 공유되는 상태 구조를 TypedDict로 정의
2. 노드 함수: 입력 상태를 받아 새로운 상태를 반환
3. StateGraph 생성: 상태 타입을 기반으로 그래프 객체 생성
4. add_node: 노드 이름과 처리 함수를 등록
5. add_edge: 노드 간의 실행 순서를 정의
6. compile: 그래프를 실행 가능한 형태로 변환
7. invoke: 입력 데이터를 기반으로 그래프 실행

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# 1. State 정의
class State(TypedDict):
    text: str
    result: str

# 2. 노드 함수 정의
def process_text(state: State) -> State:
    return {"result": state["text"].upper()}

# 3. 그래프 생성
graph = StateGraph(State)

# 4. 노드 추가
graph.add_node("process", process_text)

# 5. 엣지 추가
graph.add_edge(START, "process")
graph.add_edge("process", END)

# 6. 그래프 컴파일
app = graph.compile()

# 7. 실행
result = app.invoke({"text": "hello langgraph"})
print(result)

 

---

3. Functional API

3.1. Functional API 개념

• 노드와 엣지를 별도로 선언하지 않고, 함수형 스타일로 흐름을 구성합니다.
• 실행 순서가 단순할 때 코드가 짧고 직관적입니다.
• 복잡한 조건 분기에는 적합하지 않지만, 빠른 프로토타이핑에 유리합니다.

 

---

3.2 예시코드, 코드설명

1. State 정의: Graph API와 동일하게 상태 구조를 정의
2. @entrypoint: 워크플로우 진입 함수를 지정
3. 함수 본문: 실행 로직을 단일 함수 안에 작성
4. invoke: 바로 실행 가능

from typing import TypedDict
from langgraph.func import entrypoint

# 1. State 정의
class State(TypedDict):
    text: str
    result: str

# 2. Functional API로 워크플로우 작성
@entrypoint()
def workflow(state: State) -> State:
    result = state["text"].upper()
    return {"result": result}

# 3. 실행
output = workflow.invoke({"text": "hello functional api"})
print(output)

 

---

4. 코드

https://colab.research.google.com/drive/1GluDhXKMvN_nq5KK9r6avsnS-waFp2bj?usp=sharing

Untitled0.ipynb