โNode๊ฐ ํ๋์, Edge๊ฐ ํ๋ฆ์ ๋ง๋ ๋ค.โ
LangGraph์ ํต์ฌ์ ๋ฐ๋ก โ์ฐ๊ฒฐโ์ด๋ค.
LangGraph๋ StateGraph๋ผ๋ ๊ตฌ์กฐ๋ฅผ ์ค์ฌ์ผ๋ก ๋์ํฉ๋๋ค.
๊ทธ ๋ด๋ถ๋ ํฌ๊ฒ ๋
ธ๋(Node) ์ ์ฃ์ง(Edge) ๋ก ๊ตฌ์ฑ๋ฉ๋๋ค.
| ๊ตฌ์ฑ ์์ | ์ญํ | ์์ |
|---|---|---|
| ๐ฆ Node | ์คํ ๋จ์ (ํจ์, ์ฒด์ธ, LLM ํธ์ถ ๋ฑ) | โ๋ต๋ณ ์์ฑโ, โ๊ฒ์ ์ํโ |
| ๐ถ Edge | ํ๋ฆ ์ ์ด (๋ค์ ๋ ธ๋๋ก ์ด๋) | โ๋ต๋ณ ์ ๋ขฐ๋ < 0.8 โ ์ฌ์๋ ๋ ธ๋โ |
| ๐ง State | ๊ทธ๋ํ ๋ด ๋ฐ์ดํฐ (์ปจํ ์คํธ) | ์ง๋ฌธ, ๋ต๋ณ, ๊ฒ์ ๊ฒฐ๊ณผ ๋ฑ |
LangGraph๋ ๊ฒฐ๊ตญ โ์ด ์ธ ๊ฐ์ง๊ฐ ์ฐ๊ฒฐ๋ ๊ตฌ์กฐ์ฒดโ๋ผ๊ณ ๋ณด๋ฉด ๋ฉ๋๋ค.
์๋๋ ๋ ๊ฐ์ ๋
ธ๋๋ฅผ ๊ฐ์ง ๊ฐ์ฅ ๊ธฐ๋ณธ์ ์ธ ๊ทธ๋ํ ์์์
๋๋ค.
์ง๋ฌธ์ ๋ฐ๊ณ , ๋ต๋ณ์ ์์ฑํ๋ ๊ฐ๋จํ ํ๋ฆ์ด์์.
from langgraph.graph import StateGraph
from langchain_openai import ChatOpenAI
# ์ํ ์ ์
class GraphState:
question: str
answer: str
# ๋
ธ๋ 1: ์
๋ ฅ ๋
ธ๋
def input_node(state: GraphState):
print(f"์ฌ์ฉ์ ์ง๋ฌธ: {state.question}")
return state
# ๋
ธ๋ 2: ๋ต๋ณ ๋
ธ๋
def answer_node(state: GraphState):
llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke(f"Q: {state.question}\nA:")
state.answer = response.content
print(f"๋ต๋ณ: {state.answer}")
return state
# ๊ทธ๋ํ ๊ตฌ์ฑ
graph = StateGraph(GraphState)
graph.add_node("input", input_node)
graph.add_node("answer", answer_node)
# ๋
ธ๋ ๊ฐ ์ฐ๊ฒฐ (์ฃ์ง)
graph.add_edge("input", "answer")
# ์ง์
์ ๊ณผ ์ข
๋ฃ์ ์ง์
graph.set_entry_point("input")
graph.set_finish_point("answer")
# ์คํ
app = graph.compile()
result = app.invoke({"question": "LangGraph์ ๋
ธ๋๋ ๋ฌด์์ธ๊ฐ?"})
LangGraph๋ ๋จ์ํ ํจ์๋ค์ ์ด์ด๋ถ์ด๋ ๊ฒ ์๋๋๋ค.
๊ฐ ๋
ธ๋๋ ๋ฐ๋์ ์ํ(state)๋ฅผ ๋ฐ๊ณ , ์ํ๋ฅผ ๋ฐํํด์ผ ํฉ๋๋ค.
์ฆ, ๋ชจ๋ ๋ ธ๋๋ ๋ค์ ํํ๋ฅผ ๊ธฐ๋ณธ์ผ๋ก ํฉ๋๋ค:
def node_name(state: GraphState) -> GraphState:
# ์ํ ์์ ๋๋ ์ฐ์ฐ ์ํ
return state
โ
์ด ๊ตฌ์กฐ๋ฅผ ์งํค๋ฉด, ๊ทธ๋ํ ๋ด ๋ฐ์ดํฐ ํ๋ฆ์ด ๋ช
์์ ์ผ๋ก ์ถ์ ๊ฐ๋ฅํด์ง๋๋ค.
LLM์ reasoning ๊ฒฐ๊ณผ๋ API ํธ์ถ ๊ฒฐ๊ณผ๋ ๋ชจ๋ ์ํ๋ฅผ ํตํด ์ ๋ฌ๋ฉ๋๋ค.
LangGraph์ ์ง๊ฐ๋ ๋ฐ๋ก ์กฐ๊ฑด๋ถ ํ๋ฆ ์ ์ด์
๋๋ค.
์๋ฅผ ๋ค์ด โ๋ต๋ณ ์ ๋ขฐ๋ < 0.8์ด๋ฉด ์ฌ์๋ ๋
ธ๋๋ก ์ด๋โ ๊ฐ์ ๋ก์ง์ ๋ง๋ค ์ ์์ฃ .
def check_confidence(state: GraphState):
# ๊ฐ์ : LLM ๋ต๋ณ ์ ๋ขฐ๋๋ฅผ ๊ณ์ฐํ๋ค๊ณ ์น์
confidence = 0.6
state.confidence = confidence
return state
def retry_node(state: GraphState):
print("์ ๋ขฐ๋๊ฐ ๋ฎ์ ์ฌ์๋ํฉ๋๋ค.")
return state
def success_node(state: GraphState):
print("๋ต๋ณ ์ ๋ขฐ๋๊ฐ ์ถฉ๋ถํฉ๋๋ค.")
return state
graph = StateGraph(GraphState)
graph.add_node("check", check_confidence)
graph.add_node("retry", retry_node)
graph.add_node("success", success_node)
# ์กฐ๊ฑด๋ถ ์ฃ์ง ์ฐ๊ฒฐ
def confidence_edge(state):
if state.confidence < 0.8:
return "retry"
return "success"
graph.add_conditional_edges("check", confidence_edge)
graph.set_entry_point("check")
graph.set_finish_point("success")
app = graph.compile()
app.invoke({})
๐ ์คํ ํ๋ฆ (Mermaid ์๊ฐํ):

์ด์ฒ๋ผ, LangGraph์ ์ฃ์ง๋ โ์กฐ๊ฑด ํจ์โ๋ฅผ ํตํด ๋์ ์ผ๋ก ๊ฒฐ์ ๋ฉ๋๋ค.
LangGraph๋ state๋ฅผ ์ค์ฌ์ผ๋ก ๋
ธ๋ ๊ฐ ๋ฐ์ดํฐ๋ฅผ ์ ๋ฌํฉ๋๋ค.
์๋ฅผ ๋ค์ด ํ ๋
ธ๋์์ ์์ฑํ answer๊ฐ ๋ค์ ๋
ธ๋์ ์
๋ ฅ์ผ๋ก ์ ๋ฌ๋์ฃ .
์ด๊ฒ์ด โํ๋กฌํํธ ๊ธฐ๋ฐ ์ฌ๊ณ โ์์ ํฐ ์ฐจ์ด์ ์ ๋๋ค.
| ์ ๊ทผ ๋ฐฉ์ | ๋ฐ์ดํฐ ์ ๋ฌ ๋ฐฉ์ | ์ฅ์ |
|---|---|---|
| ํ๋กฌํํธ ๊ธฐ๋ฐ | ๋ฌธ์์ด ๊ฒฐํฉ (๋น๊ตฌ์กฐ์ ) | ๋จ์ํ์ง๋ง ํ์ฅ์ฑ ๋ฎ์ |
| ๊ทธ๋ํ ๊ธฐ๋ฐ | state ๊ฐ์ฒด ์ ์ด (๊ตฌ์กฐ์ ) | ์ถ์ , ์กฐ๊ฑด ๋ถ๊ธฐ, ๋ก๊น ๊ฐ๋ฅ |
class GraphState:
step: int
message: str
def node_a(state: GraphState):
state.step = 1
state.message = "์ฒซ ๋ฒ์งธ ๋
ธ๋์
๋๋ค."
return state
def node_b(state: GraphState):
state.step += 1
state.message += " โ ๋ ๋ฒ์งธ ๋
ธ๋ ์คํ ์๋ฃ."
print(state.message)
return state
graph = StateGraph(GraphState)
graph.add_node("A", node_a)
graph.add_node("B", node_b)
graph.add_edge("A", "B")
graph.set_entry_point("A")
graph.set_finish_point("B")
app = graph.compile()
app.invoke({})
์ถ๋ ฅ ๊ฒฐ๊ณผ:
์ฒซ ๋ฒ์งธ ๋
ธ๋์
๋๋ค. โ ๋ ๋ฒ์งธ ๋
ธ๋ ์คํ ์๋ฃ.
์ด์ฒ๋ผ, ๋ชจ๋ ๋ฐ์ดํฐ๋ ์ํ(State)๋ฅผ ํตํด ์ ๋ฌ๋๊ณ ์งํํฉ๋๋ค.
LangGraph๋ ๊ธฐ๋ณธ์ ์ผ๋ก graph.visualize() ๋ฉ์๋๋ฅผ ์ ๊ณตํฉ๋๋ค.
graph.visualize("flow.png")
๋๋ Velog / Notion ๋ฑ ๋ฌธ์ ๋ด์์๋ ์๋์ฒ๋ผ Mermaid ๋ค์ด์ด๊ทธ๋จ์ผ๋ก ์๊ฐํํ ์ ์์ต๋๋ค.

๐ 3ํธ: ์ํ ๊ด๋ฆฌ์ ํผ๋๋ฐฑ โ Reducer & MessageGraph
์ฌ๊ธฐ์๋ LangGraph์ Reducer, Memory, MessageGraph ๊ตฌ์กฐ๋ฅผ ํ์ฉํด
๋ํ์ ๋งฅ๋ฝ์ด ๋์ ๋๋ โ์ง์ํ ๊ทธ๋ํโ๋ฅผ ์ค๊ณํด๋ณผ ์์ ์
๋๋ค.
(์ฆ, ๋จ์ ์ง์์๋ต โ ์ง์์ ๋ํ ํ๋ฆ์ผ๋ก ๋ฐ์ )
| ์ฃผ์ | ํ์ต ์ด์ | ์ถ์ฒ ํ์ต ๋ฐฉํฅ |
|---|---|---|
| Graph Theory (๊ทธ๋ํ ์ด๋ก ) | ์กฐ๊ฑด๋ถ ๋ถ๊ธฐ์ ์ํ ๊ตฌ์กฐ ์ค๊ณ์ ์ํ์ ๊ธฐ๋ฐ | Directed Graph, DAG, Cycle |
| LangChain Runnable ๊ตฌ์กฐ ์ฌํ | ๊ฐ ๋ ธ๋์์ LLM, Chain, Tool์ ๋์ ์ผ๋ก ํธ์ถ | RunnableLambda, RunnableParallel |
| State ๋ณํฉ ์ ๋ต (Reducer) | ์ฌ๋ฌ ๋ ธ๋์ ๊ฒฐ๊ณผ๋ฅผ ํ๋์ ์ํ๋ก ํตํฉ | ์ํ ๋ณํฉ ํจ์, accumulator ํจํด |
| LangGraph Visualization API | ๋ํ ๊ทธ๋ํ ๋๋ฒ๊น ๋ฐ ๋ฌธ์ํ | graph.visualize(), Mermaid ์๋ ์์ฑ |
| ๋น๋๊ธฐ ๊ทธ๋ํ ์ ์ด | ToolNode ๋ณ๋ ฌ ์คํ, LLM ๋์ ํธ์ถ | async def, await app.ainvoke() |
| LangGraph + LangServe | ๊ทธ๋ํ๋ฅผ API ํํ๋ก ๋ฐฐํฌ | LangServe ๋๋ FastAPI ์ฐ๋ |
- LangGraph์ ์ต์ ๋จ์๋ Node์ Edge๋ค.
- Node๋ โ๋์โ, Edge๋ โํ๋ฆโ์ ๋ด๋นํ๋ค.
- ์ํ(State)๋ฅผ ์ค์ฌ์ผ๋ก ๋ฐ์ดํฐ๊ฐ ์ ์ด๋๋ฉฐ, ์กฐ๊ฑด๋ถ ๋ถ๊ธฐ(Conditional Edge)๋ก ๊ทธ๋ํ๊ฐ ์ด์ ์์ง์ธ๋ค.
๐ ๋ค์ ๋จ๊ณ๋ ์ด ํ๋ฆ์ โ๊ธฐ์ต(Memory)โ๊ณผ โํผ๋๋ฐฑ(Feedback)โ์ ์ฃผ์ ํ๋ ๊ฒ์ด๋ค.