LangGraph Node & Edge ์„ค๊ณ„ ์‹ค์ „ ํŒ

์ข‹์€ ์›Œํฌํ”Œ๋กœ์šฐ๋Š” ์ข‹์€ ๋…ธ๋“œ์—์„œ ์‹œ์ž‘๋œ๋‹ค


LangGraph์˜ ๋ฌธ๋ฒ• ์ž์ฒด๋Š” ์–ด๋ ต์ง€ ์•Š๋‹ค. ํ•ต์‹ฌ์€ โ€œ์ข‹์€ ํ’ˆ์งˆ์˜ ๋…ธ๋“œโ€ ๋ฅผ ์–ด๋–ป๊ฒŒ ์„ค๊ณ„ํ•˜๋А๋ƒ์— ์žˆ๋‹ค. Node๋Š” ์ตœ๋Œ€ํ•œ ์„ธ๋ถ„ํ™”ํ•ด์„œ ์„ค๊ณ„ํ•˜๋Š” ๊ฒƒ์ด ์œ ๋ฆฌํ•˜๋‹ค.

์˜ˆ๋ฅผ ๋“ค์–ด, ์งˆ๋ฌธ ์ดํ•ด, ๋ฌธ์„œ ๊ฒ€์ƒ‰, ๋‹ต๋ณ€ ์ƒ์„ฑ, ๋‹ต๋ณ€ ํ‰๊ฐ€ ๋ฅผ ํ•˜๋‚˜์˜ ๋…ธ๋“œ์—์„œ ์ฒ˜๋ฆฌํ•˜๋Š” ๊ฒƒ๋ณด๋‹ค, ๊ฐ task๋ฅผ ๋ณ„๋„ ๋…ธ๋“œ๋กœ ๋ถ„๋ฆฌํ•˜๋ฉด ๋””๋ฒ„๊น…๋„ ์‰ฌ์›Œ์ง€๊ณ , ํ๋ฆ„ ์ œ์–ด ๋ฐ ํŠœ๋‹๋„ ๋” ์ •๋ฐ€ํ•ด์ง„๋‹ค.

**LangGraph์—์„œ๋Š” โ€œ์ž‘๊ฒŒ ์ชผ๊ฐ  Nodeโ€๊ฐ€ ์ตœ๊ณ ์˜ ๋ฌด๊ธฐ๋‹ค.


โœ… ๊ธฐ๋ณธ ์˜ˆ์ œ: ๋ฌธ์„œ ๊ฒ€์ƒ‰ ๋…ธ๋“œ ๋งŒ๋“ค๊ธฐ

def retrieve_document(state: GraphState) -> GraphState:
	retrieved_docs = pdf_retriever.invoke(state["question"])
	return GraphState(context=format_docs(retrieved_docs))
  • ์ž…๋ ฅ๋„ GraphState, ์ถœ๋ ฅ๋„ GraphState
  • context์— ํ•„์š”ํ•œ ์ •๋ณด๋ฅผ ์ €์žฅํ•˜์—ฌ ๋‹ค์Œ ๋…ธ๋“œ์— ๋„˜๊ธด๋‹ค

๐Ÿ” State ํƒ€์ž… ๊ตฌ์„ฑ ์‹œ ์ฃผ์˜ํ•  ์ 

from typing import TypedDict, Annotated, List
from langchain_core.documents import Document
from operator import add
 
class GraphState(TypedDict):
	context: Annotated[List[Document], add]
	answer: Annotated[List[Document], add]
	# question: Annotated[str, add] โ† โŒ ๋ฌธ์ž์—ด์—” add Reducer ์‚ฌ์šฉ ๋ถˆ๊ฐ€
	question: Annotated[str, "user_question"]

โš ๏ธ add Reducer๋Š” ๋ฌธ์ž์—ด์—๋Š” ์ ์šฉํ•  ์ˆ˜ ์—†๋‹ค.

๐Ÿ“Œ ์ถ”์ฒœ

from langgraph.graph.message import add_messages
  • operator.add ๋Œ€์‹  add_messages ๋ฅผ ์“ฐ๋Š” ๊ฒƒ์ด ๋ช…์‹œ์ ์ด๊ณ  ์•ˆ์ „ํ•˜๋‹ค.

๐Ÿ”ง ๋…ธ๋“œ ๋ฐ˜ํ™˜ ๋ฐฉ์‹: ์–ด๋–ค ์Šคํƒ€์ผ์„ ์“ธ๊นŒ?

LangGraph ํŠœํ† ๋ฆฌ์–ผ์—์„œ ๋งŽ์ด ๋ณด์ด๋Š” ๋‘ ๊ฐ€์ง€ ๋ฐฉ์‹์ด ์žˆ๋‹ค.

โœ… ๋ฐฉ์‹ 1: ๋”•์…”๋„ˆ๋ฆฌ ๋ฐ˜ํ™˜

def retrieve(state: GraphState) -> GraphState:
	documents = "๊ฒ€์ƒ‰๋œ ๋ฌธ์„œ"
	return {"context": documents}

โœ… ๋ฐฉ์‹ 2: GraphState ํƒ€์ž… ๋ฐ˜ํ™˜

def retrieve(state: GraphState) -> GraphState:
	documents = "๊ฒ€์ƒ‰๋œ ๋ฌธ์„œ"
	return GraphState(context=documents)

๋‘ ๋ฐฉ์‹ ๋ชจ๋‘ ๊ธฐ๋Šฅ์ ์œผ๋กœ ๋™์ผํ•˜์ง€๋งŒ,

ํƒ€์ž… ํžŒํŒ… + ์ผ๊ด€์„ฑ ์žˆ๋Š” ์ฝ”๋“œ ์œ ์ง€๋ฅผ ์œ„ํ•ด 2๋ฒˆ ๋ฐฉ์‹ ์‚ฌ์šฉ์„ ๊ถŒ์žฅํ•œ๋‹ค.


โš™๏ธ ๊ทธ๋ž˜ํ”„ ์ƒ์„ฑ ๋ฐฉ์‹: ์–ด๋–ค ๊ฒŒ ๋” ์ข‹์„๊นŒ?

LangGraph์—์„œ๋Š” ๋‘ ๊ฐ€์ง€ ๋ฐฉ์‹์œผ๋กœ ๊ทธ๋ž˜ํ”„๋ฅผ ์ •์˜ํ•  ์ˆ˜ ์žˆ๋‹ค.

๐Ÿ“Œ ๋ฐฉ์‹ 1: START ๋ช…์‹œ

from langgraph.graph import START, StateGraph
 
workflow = StateGraph(state_schema=GraphState)
workflow.add_node("retrieve", retrieve)
workflow.add_edge(START, "retrieve")

๐Ÿ“Œ ๋ฐฉ์‹ 2: set_entry_point ์‚ฌ์šฉ

from langgraph.graph import StateGraph
 
workflow = StateGraph(state_schema=GraphState)
workflow.add("retrieve", retrieve)
workflow.set_entry_point("retrieve")

โœ… ๊ธฐ๋Šฅ์ ์œผ๋กœ๋Š” ๋™์ผํ•˜๋ฉฐ, ๋‘˜ ์ค‘ ์–ด๋–ค ๊ฑธ ์‚ฌ์šฉํ•ด๋„ ๊ฒฐ๊ณผ๋Š” ๊ฐ™๋‹ค.
โ†’ ์ทจํ–ฅ ๋˜๋Š” ์ฝ”๋“œ ์Šคํƒ€์ผ์— ๋งž์ถฐ ์„ ํƒํ•˜๋ฉด ๋œ๋‹ค.


๐Ÿ“ ์„ค๊ณ„ ํŒ: ๊ทธ๋ž˜ํ”„ ๊ตฌ์กฐ๋Š” ๋จผ์ € ์ข…์ด์— ๊ทธ๋ ค๋ผ

์ง์ ‘ ๊ตฌํ˜„์— ๋“ค์–ด๊ฐ€๊ธฐ ์ „์—,

์งˆ๋ฌธ ํ๋ฆ„, ๋ถ„๊ธฐ ์กฐ๊ฑด, ๋…ธ๋“œ ์ˆœ์„œ ๋“ฑ์„ ๋ฉ”๋ชจ์žฅ์ด๋‚˜ ๋‹ค์ด์–ด๊ทธ๋žจ ํˆด์— ๊ทธ๋ ค๋ณด๋Š” ๊ฒƒ์„ ๊ฐ•๋ ฅ ์ถ”์ฒœํ•˜๋‹ค.

  • ๋ถˆํ•„์š”ํ•œ ๋ถ„๊ธฐ ์ œ๊ฑฐ
  • ๋ณ‘๋ ฌ ์ฒ˜๋ฆฌ ๊ฐ€๋Šฅํ•œ ์ง€์  ํ™•์ธ
  • ๋‚˜์ค‘์— LangGraph ์‹œ๊ฐํ™” ๊ธฐ๋Šฅ๊ณผ 1:1 ๋งคํ•‘ํ•˜๊ธฐ ์‰ฌ์›€

๐Ÿ“š RAG ์‚ฌ์šฉ ์‹œ ์ถ”๊ฐ€ ํŒ: ๋ฌธ์„œ ํฌ๋งท ์ฒ˜๋ฆฌ

LLM์— List[Document] ๊ฐ์ฒด๋ฅผ ๋„˜๊ธฐ๊ธฐ๋ณด๋‹ค๋Š”,

๋ฌธ์„œ๋“ค์„ ํฌ๋งทํŒ…ํ•ด์„œ ํ•˜๋‚˜์˜ ๋ฌธ์ž์—ด๋กœ ๋ณ€ํ™˜ํ•˜๋Š” ๊ฒƒ์ด ์„ฑ๋Šฅ์ƒ ํ›จ์”ฌ ์œ ๋ฆฌํ•˜๋‹ค.

formatted_context = format_docs(retrieved_docs)

๐ŸŽฏ ์ด์œ 

  • LLM์ด ์ดํ•ดํ•˜๊ธฐ ์‰ฌ์šด ํ˜•ํƒœ๋กœ ์ •์ œ
  • ๋ถˆํ•„์š”ํ•œ metadata ์ œ๊ฑฐ
  • ๋ฌธ์„œ ๊ฒฝ๊ณ„์— ํƒœ๊ทธ ์ถ”๊ฐ€ (์˜ˆ: ### ๋ฌธ์„œ 1 ###, <document></document> ๋“ฑ)