โLLM์ ํ๊ณ๋ฅผ ๋์ด, ๊ฒ์์ผ๋ก ๊ฐํ๋ ์ง๋ฅํ ๊ทธ๋ํ ๋ง๋ค๊ธฐ.โ
LangGraph๋ก RAG(๊ฒ์-์์ฑ-ํ๊ฐ-์์ ) ๋ฃจํ๋ฅผ ๊ตฌํํ๋ค.
RAG(Retrieval-Augmented Generation)๋ ๊ฒ์(Retrieval) ๊ณผ ์์ฑ(Generation) ์ ๊ฒฐํฉํ ๊ตฌ์กฐ์ ๋๋ค.
๋จ์ LLM์ ๊ธฐ์ต์ด ์ ํ์ ์ด์ง๋ง, RAG๋ ์ธ๋ถ ์ง์(์: ๋ฌธ์, DB, ๋ฒกํฐDB)์ ๊ฒ์ํด ๊ทธ ๊ฒฐ๊ณผ๋ฅผ LLM ์ ๋ ฅ์ผ๋ก ๋ฃ์ด ์ ํ์ฑ์ ๋์ ๋๋ค.
LangGraph๋ ๊ธฐ์กด์ query โ retrieval โ generation ์์ฐจ ๊ตฌ์กฐ๋ฅผ
๊ทธ๋ํ ํํ๋ก ํํํ์ฌ ์กฐ๊ฑด ๋ถ๊ธฐ, ํผ๋๋ฐฑ, ๋ฃจํ ์ฌ์๋ ๋ฑ์ ๋ช
์์ ์ผ๋ก ์ ์ดํฉ๋๋ค.
๐ ์ผ๋ฐ์ ๊ตฌ์กฐ:

์ด ๊ตฌ์กฐ๋ฅผ ๊ทธ๋๋ก ์ฝ๋๋ก ๋ง๋ค ์ ์์ต๋๋ค.
| ๊ฐ๋ | ์ญํ | LangGraph ํํ |
|---|---|---|
| Retrieval | ์ธ๋ถ ์ง์ ๊ฒ์ | Node (VectorDB Query) |
| Generation | ๊ฒ์ ๊ฒฐ๊ณผ ๊ธฐ๋ฐ ์๋ต ์์ฑ | Node (LLM ํธ์ถ) |
| Evaluation | ์๋ต ํ์ง ํ๊ฐ | Conditional Edge |
| Correction | ์ ํ์ง ์ ์ฌ๊ฒ์/์ฌ์์ฑ | Feedback Loop |
| HITL | Human-in-the-Loop (์ฌ๋ ๊ฐ์ ) | SubGraph / Manual Node |
์๋๋ โ๊ฒ์ โ ์์ฑ โ ํ๊ฐโ 3๋จ๊ณ ๊ทธ๋ํ๋ฅผ ๊ตฌํํ ์์์ ๋๋ค.
from langgraph.graph import StateGraph
from langchain_openai import ChatOpenAI
# ์ํ ์ ์
class RAGState:
query: str
docs: list[str]
answer: str
score: float
# ๊ฒ์ ๋
ธ๋
def retrieve(state: RAGState):
# ์ค์ ๋ก๋ VectorDB (์: Pinecone, Chroma) ํธ์ถ
fake_docs = [
"LangGraph๋ LangChain ์ํ๊ณ์ ๊ทธ๋ํ ์ํฌํ๋ก์ฐ ์์ง์ด๋ค.",
"LangGraph๋ ๋
ธ๋์ ์ฃ์ง๋ก LLM ์คํ ํ๋ฆ์ ์ ์ดํ๋ค."
]
state.docs = fake_docs
print(f"[Retrieve] {len(fake_docs)}๊ฐ์ ๋ฌธ์ ๊ฒ์๋จ")
return state
# ์์ฑ ๋
ธ๋
def generate(state: RAGState):
llm = ChatOpenAI(model="gpt-4o-mini")
prompt = f"๋ฌธ์ ๋ด์ฉ:\n{state.docs}\n\n์ง๋ฌธ: {state.query}\n๋ต๋ณ:"
result = llm.invoke(prompt)
state.answer = result.content
print(f"[Generate] ๋ต๋ณ ์์ฑ ์๋ฃ")
return state
# ํ๊ฐ ๋
ธ๋
def evaluate(state: RAGState):
llm = ChatOpenAI(model="gpt-4o-mini")
eval_prompt = f"๋ค์ ๋ต๋ณ์ ์ ํ๋๋ฅผ 0~1 ์ฌ์ด ์ ์๋ก ํ๊ฐ:\n{state.answer}"
result = llm.invoke(eval_prompt)
try:
state.score = float(result.content.strip())
except:
state.score = 0.7 # ๊ธฐ๋ณธ๊ฐ
print(f"[Evaluate] ํ์ง ์ ์: {state.score}")
return "finish" if state.score >= 0.8 else "retry"
# ๊ทธ๋ํ ๊ตฌ์ฑ
graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.add_node("evaluate", evaluate)
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", "evaluate")
graph.add_conditional_edges("evaluate", lambda s: "finish" if s.score >= 0.8 else "retrieve")
graph.set_entry_point("retrieve")
graph.set_finish_point("evaluate")
app = graph.compile()
app.invoke({"query": "LangGraph๋ ๋ฌด์์ธ๊ฐ?"})
๐งฉ ์คํ ๊ฒฐ๊ณผ ์์:
[Retrieve] 2๊ฐ์ ๋ฌธ์ ๊ฒ์๋จ
[Generate] ๋ต๋ณ ์์ฑ ์๋ฃ
[Evaluate] ํ์ง ์ ์: 0.85
LangGraph๋ RAG๋ฅผ ๋จ์ ์คํํ์์ ์ง๋ฅํ ๋ฃจํํ์ผ๋ก ๋ฐ์ ์ํฌ ์ ์์ต๋๋ค.
| ์ ํ | ์ค๋ช | ๊ทธ๋ํ ํน์ง |
|---|---|---|
| Adaptive RAG | ์ง๋ฌธ ๋์ด๋์ ๋ฐ๋ผ ๊ฒ์ ๊น์ด๋ฅผ ์กฐ์ | โ์ง๋ฌธ ๊ธธ์ด or ๋์ด๋ ๊ธฐ๋ฐ ์กฐ๊ฑด ๋ถ๊ธฐโ |
| Self-RAG | LLM์ด ์ค์ค๋ก ๋ต๋ณ ํ๊ฐ ๋ฐ ์์ | โ์์ฒด ํ๊ฐ ๋ฃจํโ |
| Corrective RAG | LLM + ์ฌ์ฉ์ ํผ๋๋ฐฑ์ผ๋ก ์์ | โHITL + SubGraph ๊ตฌ์กฐโ |
def choose_strategy(state):
if len(state.query) < 20:
state.depth = "shallow"
else:
state.depth = "deep"
return state
def retrieve_shallow(state): ...
def retrieve_deep(state): ...
graph.add_node("strategy", choose_strategy)
graph.add_node("shallow", retrieve_shallow)
graph.add_node("deep", retrieve_deep)
graph.add_conditional_edges("strategy", lambda s: s.depth)
๐ ์๊ฐํ:

def human_feedback(state):
feedback = input("๋ต๋ณ์ ๋ํ ํผ๋๋ฐฑ์ ์
๋ ฅํ์ธ์: ")
state.feedback = feedback
return state
human_feedback ๋
ธ๋๋ฅผ ๊ทธ๋ํ์ ํฌํจํ๋ฉด ์ฌ์ฉ์๊ฐ ๋ฃจํ ์ค๊ฐ์ ์ง์ ๊ฐ์
(HITL, Human-in-the-loop)ํ ์ ์์ต๋๋ค.
LangGraph๋ SubGraph ๊ธฐ๋ฅ์ ํตํด RAG ์ ์ฒด๋ฅผ ํ๋์ ๋ชจ๋๋ก ๋ฌถ์ด ๋ค๋ฅธ ๊ทธ๋ํ์์ ์ฌ์ฌ์ฉํ ์ ์์ต๋๋ค.
from langgraph.graph import SubGraph
rag_module = SubGraph(graph)
main_graph.add_node("RAGPipeline", rag_module)
โ ์ด๋ ๊ฒ ํ๋ฉด โ๊ฒ์-์์ฑ-ํ๊ฐโ ๋จ์์ RAG ํ๋ฆ์ ๋ค๋ฅธ ์์ด์ ํธ๋ ์์คํ ์ ๊ทธ๋๋ก ์ฌํ์ฉํ ์ ์์ต๋๋ค.
๋๊ท๋ชจ RAG๋ฅผ ์ด์ํ ๋๋ ๋ค์์ ๋ฐ๋์ ๊ณ ๋ คํ์ธ์:
| ํฌ์ธํธ | ์ค๋ช |
|---|---|
| ์บ์ฑ (Cache) | ๋ฐ๋ณต ์ง๋ฌธ์ ๋ํ ์๋ต ์ ์ฅ |
| ๋น๋๊ธฐ ์ฒ๋ฆฌ | ๊ฒ์/์์ฑ ๋์ ์คํ (asyncio) |
| ๋ก๊ทธ ์ถ์ | ๊ฐ ๋ ธ๋๋ณ latency ๊ธฐ๋ก |
| ์ค๋ฅ ๋ณต๊ตฌ | VectorDB / LLM ํธ์ถ ์คํจ ์ fallback |
| ๋ณ๋ ฌ ๊ฒ์ | ์ฌ๋ฌ ์์ค ๋์ query |
LangGraph๋ ๊ตฌ์กฐ์ ์ผ๋ก ์ด ๋ชจ๋ ์ ์ด๋ฅผ ๊ทธ๋ํ ๋ ๋ฒจ์์ ์ํํ ์ ์์ต๋๋ค.
๐ 7ํธ: LangGraph + UI ํตํฉ (Gradio, Streamlit)
์ฌ๊ธฐ์๋ ์ง๊ธ ๋ง๋ RAG ๊ทธ๋ํ๋ฅผ ์๊ฐํํ๊ณ ์กฐ์ ๊ฐ๋ฅํ ์ธํฐํ์ด์ค๋ก ์ฐ๊ฒฐํฉ๋๋ค.
์ฌ์ฉ์๊ฐ ์ง์ ์
๋ ฅํ๊ณ ๊ฒฐ๊ณผ๋ฅผ ์ค์๊ฐ ํ์ธํ ์ ์๋ LangGraph ๋์๋ณด๋ํ ํ๊ฒฝ์ ๋ง๋ค์ด๋ด
๋๋ค.
| ์ฃผ์ | ํ์ต ์ด์ | ์ถ์ฒ ํ์ต ๋ฐฉํฅ |
|---|---|---|
| VectorDB ์ต์ ํ | ๊ฒ์ ์๋์ ์ ํ๋ ํฅ์ | Pinecone, Chroma, FAISS ์ธ๋ฑ์ฑ ์ค์ต |
| Evaluation Loop ์ค๊ณ | Self-RAG ๊ตฌํ ํต์ฌ | ๋ชจ๋ธ ํ๊ฐ/์ค์ฝ์ด๋ง ๋ฃจํ |
| SubGraph ๋ชจ๋ํ | ๋ํ ๊ทธ๋ํ ์ ์ง๋ณด์ ํจ์จ์ฑ | Graph Composition, Nested Graph |
| LangGraph HITL ์ค๊ณ | ์ธ๊ฐ ๊ฐ์ ํ ํ๊ฐ ์์คํ | Human-in-the-Loop workflow |
| LangGraph + LangServe ๋ฐฐํฌ | RAG ์์คํ APIํ | LangServe or FastAPI endpoint |
| RAG ์ฑ๋ฅ ์งํ ์ค๊ณ | ํ์ง ๊ด๋ฆฌ | Recall@k, F1, Faithfulness metric |
- LangGraph๋ RAG์ ๊ฒ์-์์ฑ-ํ๊ฐ ๋ฃจํ๋ฅผ ๊ทธ๋ํ ๋จ์๋ก ์ ์ดํ ์ ์๋ค.
- Adaptive / Self / Corrective RAG๋ก ํ์ฅํ๋ฉด LLM์ด ์ค์ค๋ก ํ๊ฐยท์์ ํ ์ ์๋ค.
- SubGraph, Feedback, HITL๋ก ์ค์ ์์ค์ RAG ํ์ดํ๋ผ์ธ ์ค๊ณ๊ฐ ๊ฐ๋ฅํ๋ค.
๐ก RAG๋ ๋จ์ํ ๋ณด์กฐ ๊ธฐ๋ฅ์ด ์๋๋ผ, LangGraph์์ โ์ง๋ฅํ ๋ฃจํ์ ํต์ฌโ์ด๋ค.