๐Ÿงญ LangGraph ์™„์ „ ์ •๋ณต โ‘ก โ€” ๋…ธ๋“œ์™€ ์—ฃ์ง€: LangGraph์˜ ์ตœ์†Œ ๋‹จ์œ„

okorionยท2025๋…„ 10์›” 4์ผ

โ€œNode๊ฐ€ ํ–‰๋™์„, Edge๊ฐ€ ํ๋ฆ„์„ ๋งŒ๋“ ๋‹ค.โ€
LangGraph์˜ ํ•ต์‹ฌ์€ ๋ฐ”๋กœ โ€˜์—ฐ๊ฒฐโ€™์ด๋‹ค.


๐Ÿงฉ 1. LangGraph์˜ ๊ธฐ๋ณธ ๊ตฌ์„ฑ์š”์†Œ

LangGraph๋Š” StateGraph๋ผ๋Š” ๊ตฌ์กฐ๋ฅผ ์ค‘์‹ฌ์œผ๋กœ ๋™์ž‘ํ•ฉ๋‹ˆ๋‹ค.
๊ทธ ๋‚ด๋ถ€๋Š” ํฌ๊ฒŒ ๋…ธ๋“œ(Node) ์™€ ์—ฃ์ง€(Edge) ๋กœ ๊ตฌ์„ฑ๋ฉ๋‹ˆ๋‹ค.

๊ตฌ์„ฑ ์š”์†Œ์—ญํ• ์˜ˆ์‹œ
๐ŸŸฆ Node์‹คํ–‰ ๋‹จ์œ„ (ํ•จ์ˆ˜, ์ฒด์ธ, LLM ํ˜ธ์ถœ ๋“ฑ)โ€œ๋‹ต๋ณ€ ์ƒ์„ฑโ€, โ€œ๊ฒ€์ƒ‰ ์ˆ˜ํ–‰โ€
๐Ÿ”ถ Edgeํ๋ฆ„ ์ œ์–ด (๋‹ค์Œ ๋…ธ๋“œ๋กœ ์ด๋™)โ€œ๋‹ต๋ณ€ ์‹ ๋ขฐ๋„ < 0.8 โ†’ ์žฌ์‹œ๋„ ๋…ธ๋“œโ€
๐Ÿง  State๊ทธ๋ž˜ํ”„ ๋‚ด ๋ฐ์ดํ„ฐ (์ปจํ…์ŠคํŠธ)์งˆ๋ฌธ, ๋‹ต๋ณ€, ๊ฒ€์ƒ‰ ๊ฒฐ๊ณผ ๋“ฑ

LangGraph๋Š” ๊ฒฐ๊ตญ โ€œ์ด ์„ธ ๊ฐ€์ง€๊ฐ€ ์—ฐ๊ฒฐ๋œ ๊ตฌ์กฐ์ฒดโ€๋ผ๊ณ  ๋ณด๋ฉด ๋ฉ๋‹ˆ๋‹ค.


โš™๏ธ 2. ๊ธฐ๋ณธ ๊ตฌ์กฐ ์ฝ”๋“œ ์˜ˆ์‹œ

์•„๋ž˜๋Š” ๋‘ ๊ฐœ์˜ ๋…ธ๋“œ๋ฅผ ๊ฐ€์ง„ ๊ฐ€์žฅ ๊ธฐ๋ณธ์ ์ธ ๊ทธ๋ž˜ํ”„ ์˜ˆ์‹œ์ž…๋‹ˆ๋‹ค.
์งˆ๋ฌธ์„ ๋ฐ›๊ณ , ๋‹ต๋ณ€์„ ์ƒ์„ฑํ•˜๋Š” ๊ฐ„๋‹จํ•œ ํ๋ฆ„์ด์—์š”.

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์˜ ๋…ธ๋“œ๋ž€ ๋ฌด์—‡์ธ๊ฐ€?"})

๐Ÿง  3. ๋…ธ๋“œ์™€ ์—ฃ์ง€์˜ ํ•ต์‹ฌ ๊ทœ์น™

LangGraph๋Š” ๋‹จ์ˆœํžˆ ํ•จ์ˆ˜๋“ค์„ ์ด์–ด๋ถ™์ด๋Š” ๊ฒŒ ์•„๋‹™๋‹ˆ๋‹ค.
๊ฐ ๋…ธ๋“œ๋Š” ๋ฐ˜๋“œ์‹œ ์ƒํƒœ(state)๋ฅผ ๋ฐ›๊ณ , ์ƒํƒœ๋ฅผ ๋ฐ˜ํ™˜ํ•ด์•ผ ํ•ฉ๋‹ˆ๋‹ค.

์ฆ‰, ๋ชจ๋“  ๋…ธ๋“œ๋Š” ๋‹ค์Œ ํ˜•ํƒœ๋ฅผ ๊ธฐ๋ณธ์œผ๋กœ ํ•ฉ๋‹ˆ๋‹ค:

def node_name(state: GraphState) -> GraphState:
    # ์ƒํƒœ ์ˆ˜์ • ๋˜๋Š” ์—ฐ์‚ฐ ์ˆ˜ํ–‰
    return state

โœ… ์ด ๊ตฌ์กฐ๋ฅผ ์ง€ํ‚ค๋ฉด, ๊ทธ๋ž˜ํ”„ ๋‚ด ๋ฐ์ดํ„ฐ ํ๋ฆ„์ด ๋ช…์‹œ์ ์œผ๋กœ ์ถ”์  ๊ฐ€๋Šฅํ•ด์ง‘๋‹ˆ๋‹ค.
LLM์˜ reasoning ๊ฒฐ๊ณผ๋‚˜ API ํ˜ธ์ถœ ๊ฒฐ๊ณผ๋„ ๋ชจ๋‘ ์ƒํƒœ๋ฅผ ํ†ตํ•ด ์ „๋‹ฌ๋ฉ๋‹ˆ๋‹ค.


๐Ÿ”€ 4. ์กฐ๊ฑด๋ถ€ ๋ถ„๊ธฐ (Conditional Edge)

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์˜ ์—ฃ์ง€๋Š” โ€œ์กฐ๊ฑด ํ•จ์ˆ˜โ€๋ฅผ ํ†ตํ•ด ๋™์ ์œผ๋กœ ๊ฒฐ์ •๋ฉ๋‹ˆ๋‹ค.


๐Ÿงฑ 5. ๋ฐ์ดํ„ฐ ์ „๋‹ฌ๊ณผ ์ƒํƒœ ์ „์ด

LangGraph๋Š” state๋ฅผ ์ค‘์‹ฌ์œผ๋กœ ๋…ธ๋“œ ๊ฐ„ ๋ฐ์ดํ„ฐ๋ฅผ ์ „๋‹ฌํ•ฉ๋‹ˆ๋‹ค.
์˜ˆ๋ฅผ ๋“ค์–ด ํ•œ ๋…ธ๋“œ์—์„œ ์ƒ์„ฑํ•œ answer๊ฐ€ ๋‹ค์Œ ๋…ธ๋“œ์˜ ์ž…๋ ฅ์œผ๋กœ ์ „๋‹ฌ๋˜์ฃ .

์ด๊ฒƒ์ด โ€œํ”„๋กฌํ”„ํŠธ ๊ธฐ๋ฐ˜ ์‚ฌ๊ณ โ€์™€์˜ ํฐ ์ฐจ์ด์ ์ž…๋‹ˆ๋‹ค.

์ ‘๊ทผ ๋ฐฉ์‹๋ฐ์ดํ„ฐ ์ „๋‹ฌ ๋ฐฉ์‹์žฅ์ 
ํ”„๋กฌํ”„ํŠธ ๊ธฐ๋ฐ˜๋ฌธ์ž์—ด ๊ฒฐํ•ฉ (๋น„๊ตฌ์กฐ์ )๋‹จ์ˆœํ•˜์ง€๋งŒ ํ™•์žฅ์„ฑ ๋‚ฎ์Œ
๊ทธ๋ž˜ํ”„ ๊ธฐ๋ฐ˜state ๊ฐ์ฒด ์ „์ด (๊ตฌ์กฐ์ )์ถ”์ , ์กฐ๊ฑด ๋ถ„๊ธฐ, ๋กœ๊น… ๊ฐ€๋Šฅ

๐Ÿ” 6. ๋…ธ๋“œ ๊ฐ„ ๋ฐ์ดํ„ฐ ํ๋ฆ„ ์˜ˆ์‹œ

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)๋ฅผ ํ†ตํ•ด ์ „๋‹ฌ๋˜๊ณ  ์ง„ํ™”ํ•ฉ๋‹ˆ๋‹ค.


๐Ÿช„ 7. LangGraph๋ฅผ ์‹œ๊ฐํ™”ํ•˜๋Š” ๋ฐฉ๋ฒ•

LangGraph๋Š” ๊ธฐ๋ณธ์ ์œผ๋กœ graph.visualize() ๋ฉ”์„œ๋“œ๋ฅผ ์ œ๊ณตํ•ฉ๋‹ˆ๋‹ค.

graph.visualize("flow.png")

๋˜๋Š” Velog / Notion ๋“ฑ ๋ฌธ์„œ ๋‚ด์—์„œ๋Š” ์•„๋ž˜์ฒ˜๋Ÿผ Mermaid ๋‹ค์ด์–ด๊ทธ๋žจ์œผ๋กœ ์‹œ๊ฐํ™”ํ•  ์ˆ˜ ์žˆ์Šต๋‹ˆ๋‹ค.


๐Ÿงญ 8. ๋‹ค์Œ ํšŒ์ฐจ ์˜ˆ๊ณ 

๐Ÿ‘‰ 3ํŽธ: ์ƒํƒœ ๊ด€๋ฆฌ์™€ ํ”ผ๋“œ๋ฐฑ โ€“ Reducer & MessageGraph

์—ฌ๊ธฐ์„œ๋Š” LangGraph์˜ Reducer, Memory, MessageGraph ๊ตฌ์กฐ๋ฅผ ํ™œ์šฉํ•ด
๋Œ€ํ™”์˜ ๋งฅ๋ฝ์ด ๋ˆ„์ ๋˜๋Š” โ€œ์ง€์†ํ˜• ๊ทธ๋ž˜ํ”„โ€๋ฅผ ์„ค๊ณ„ํ•ด๋ณผ ์˜ˆ์ •์ž…๋‹ˆ๋‹ค.
(์ฆ‰, ๋‹จ์ˆœ ์งˆ์˜์‘๋‹ต โ†’ ์ง€์†์  ๋Œ€ํ™” ํ๋ฆ„์œผ๋กœ ๋ฐœ์ „)


๐ŸŽ“ 9. ๋” ๊นŠ์ด ๋ฐฐ์šฐ๊ธฐ ์œ„ํ•œ ๊ณ ๊ธ‰ ํ™•์žฅ ํ•™์Šต ๊ฐ€์ด๋“œ

์ฃผ์ œํ•™์Šต ์ด์œ ์ถ”์ฒœ ํ•™์Šต ๋ฐฉํ–ฅ
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)โ€์„ ์ฃผ์ž…ํ•˜๋Š” ๊ฒƒ์ด๋‹ค.

profile
okorion's Tech Study Blog.

0๊ฐœ์˜ ๋Œ“๊ธ€