Langgraph란?
LCEL은 다음과 같은 작업에 좋습니다
| 잘 맞는 task | 예시 |
|---|---|
| 단일 호출 작업 | 번역, 요약, 감정 분류 |
| 입력 → 출력이 명확한 변환 | 비정형 텍스트 → JSON |
| 단계가 정해진 파이프라인 | 검색 → 프롬프트 조립 → 호출 |
하지만 진짜 에이전트가 풀어야 하는 작업은 이런 모습입니다:
사용자가 “지난 분기 매출 보고서를 분석해줘” 라고 요청.
- 회사 DB에서 매출 데이터를 가져온다.
- 데이터에 이상치가 있는지 확인한다 — 이상치가 있으면 다시 가져와본다.
- 분석 결과를 작성한다.
- 보고서가 너무 길면 요약하고, 짧으면 그대로 둔다.
- 사용자에게 검토를 요청한다 — 사용자가 수정 요청하면 2단계로 돌아간다.
여기에 LCEL을 어떻게 끼워 맞출까요? 어색합니다. 분기·재시도·사람 개입·반복이 너무 많아요.
LCEL vs LangGraph 한 줄 비교
| 측면 | LCEL | LangGraph |
|---|---|---|
| 모델 | 파이프 (단방향 DAG) | 그래프 (분기·순환 가능) |
| 분기 | 어색 | 자연스러움 (조건부 엣지) |
| 순환 | 불가 | 자연스러움 (다시 같은 노드로) |
| 상태 | 매번 수동 전달 | State 객체로 자동 공유 |
| 사람 개입 | 어렵음 | interrupt() 한 줄 |
| 디버깅 | 단계 추적 | LangSmith로 그래프 시각화 |
LangGraph의 핵심 4요소 + Cycle (15분)
① State (상태) — “그래프의 중앙 저장소”
전체 워크플로의 컨텍스트를 관리하는 중앙 저장소입니다. 단순한 변수 전달을 넘어 그래프 실행 과정에서 지속적으로 유지되는 데이터입니다.
전체 워크플로의 컨텍스트를 관리하는 중앙 저장소입니다. 단순한 변수 전달을 넘어 그래프 실행 과정에서 지속적으로 유지되는 데이터입니다.복사
from typing import TypedDict
from pydantic import BaseModel # 타입 안정성을 위해 Pydantic도 사용 가능
class State(TypedDict):
query: str
documents: list
answer: str
# 또는 Pydantic으로 더 강력한 타입 체크
class GraphState(BaseModel):
query: str
documents: list = []
answer: str = ""② Node (노드) — “그래프의 기본 실행 단위”
특정 작업을 수행하는 함수나 에이전트입니다. State를 입력받아 작업을 수행하고, 업데이트된 State를 반환합니다.
def retrieve(state: State) -> dict:
# 1. 현재 State를 입력으로 받음
query = state[“query”]
# 2. 특정 작업 수행 (LLM 호출, DB 조회, API 호출 등)
docs = vectorstore.similarity_search(query)
# 3. 업데이트된 State 반환 (일부만 갱신 가능)
return {"documents": docs}add_node — 작업을 그래프에 등록하기
함수를 정의했다면, 그래프에 노드로 등록해야 합니다:
③ Edge (엣지) — “노드 간의 연결과 실행 흐름”
노드 간의 연결을 정의하고 전체적인 실행 흐름을 제어합니다. “다음엔 어디로 갈까?”를 결정합니다.
일반 에지 (Normal Edge)
항상 정해진 같은 경로로만 진행합니다:복사
graph.add_edge("retrieve", "generate")
# retrieve 노드가 끝나면 무조건 generate 노드로 간다START와 END — 그래프의 입구와 출구
LangGraph는 특별한 노드 두 개를 제공합니다:
- START: 그래프의 시작점. “이 그래프는 여기서부터 시작한다”
- END: 그래프의 종료점. “더 이상 갈 곳이 없으니 여기서 끝”
Agentic RAG (지금 빌드할 것)
위 패턴들이 한 그래프에 합쳐져 있음. 즉 우리가 빌드할 게 이 진화의 마지막 단계.
| 진화 | 추가된 노드 | 푸는 Naive 한계 |
|---|---|---|
| Groundedness | grade | 무관한 문서 차단 |
| Query Rewrite | rewrite + cycle | 모호한 질문 |
| Web Search | (선택) web_search | 문서에 답 없을 때 |
| Hallucination Check | (선택) check_hallucination | 부정확한 답변 |
| Agentic RAG | retrieve / grade / generate / rewrite / finalize 5노드 | 위 모두 + 안전장치 |
ReAct vs Agentic RAG — 트레이드오프
같은 기능, 두 가지 구현 방식. 어느 게 좋을까?
✅ ReAct 장점
- 코드가 짧고 단순
- 새 도구 추가만으로 기능 확장 (그래프 수정 X)
- LLM이 흐름을 결정해 유연성 ↑ — 예상 못 한 패턴도 처리
❌ ReAct 단점
- 흐름 제어가 LLM에 달려있어 예측이 어려움
- 시스템 프롬프트가 엉성하면 잘못된 도구 호출
- 디버깅 어려움 (“왜 모델이 이 도구를 호출했지?”)
✅ Agentic RAG (그래프) 장점
- 흐름이 코드에 명시 — 디버깅 쉬움, 재현성 높음
- 강제 분기 가능 (LLM 결정 ↔ 무관)
- 비용 예측 가능 (호출 수 거의 고정)
❌ Agentic RAG 단점
- 새 패턴 추가시 그래프 변경 필요
- 코드 길어짐
- LLM이 보지 못한 케이스에 취약
언제 어느 쪽?
| 상황 | 권장 |
|---|---|
| 도메인이 좁고 흐름이 정해짐 | 그래프 (예측가능성 ↑) |
| 도구 종류가 많고 다양한 시나리오 | ReAct (유연성 ↑) |
| 안전·재현성이 중요 (금융 등) | 그래프 + 강제 검증 |
| 빠른 프로토타이핑 | ReAct |
| 운영 안정성 + 기능 확장 | 하이브리드 (Day 5 멀티에이전트에서) |
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage
from pydantic import BaseModel, Field
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv())
# 데모 문서
SAMPLE_DOCS = [
"휴가는 연 15일 부여되며, 5일까지 다음 해로 이월 가능합니다.",
"출장비는 영수증 첨부 후 7일 내 정산됩니다.",
"신입사원은 4주 온보딩 과정을 거칩니다.",
"사내 자료는 외부 공유 금지, VPN 사용 필수.",
"프로젝트 예산 변경은 매니저 승인이 필요합니다.",
]
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
llm = ChatOpenAI(model="gpt-5.4-mini")
vectorstore = Chroma.from_texts(SAMPLE_DOCS, embeddings, collection_name="react_demo")
class State(TypedDict):
messages: Annotated[list, add_messages]
class Grade(BaseModel):
score: Literal["yes", "no"] = Field(description="관련 있으면 yes")
grader = llm.with_structured_output(Grade)
@tool
def retrieve_docs(query: str) -> str:
"""사내 문서를 벡터 검색합니다. 검색 후 grade_docs로 관련성을 평가하세요."""
docs = vectorstore.similarity_search(query, k=3)
if not docs:
return "검색 결과 없음"
result = "\n\n---\n\n".join(d.page_content for d in docs)
return f"검색된 문서 {len(docs)}개:\n\n{result}"
@tool
def grade_docs(question: str, document: str) -> str:
"""검색된 문서가 질문과 관련 있는지 평가합니다. yes/no 반환."""
result = grader.invoke(f"문서: {document}\n\n질문: {question}\n\n관련 있으면 yes.")
return f"관련성: {result.score}"
@tool
def rewrite_query(question: str) -> str:
"""검색 결과가 부족할 때 쿼리를 더 효과적으로 재작성합니다."""
result = llm.invoke(
f"다음 질문을 검색에 더 적합하게 재작성. 재작성된 질문만 출력:\n{question}"
)
return result.content.strip()
@tool
def calculator(expression: str) -> str:
"""간단한 수학 계산을 수행합니다."""
try:
return str(eval(expression, {"__builtins__": {}}, {}))
except Exception as e:
return f"계산 실패: {e}"
tools = [retrieve_docs, grade_docs, rewrite_query, calculator]
SYSTEM_PROMPT = """
당신은 TechCorp 사내 AI 어시스턴트입니다.
[도구 사용 전략]
1. 사내 규정/정책 질문 -> retrieve_docs로 검색
2. 검색 결과 -> grade_docs로 관련성 평가
3. 관련성이 no면 -> rewrite_query로 질문 재작성 -> retrieve_docs 재검색
4. 최대 3번 검색. 그래도 없으면 "관련 정보를 찾을 수 없습니다" 라고 답변해
5. 답변은 반드시 검색 문서에 기반
6. 계산이 필요하면 Calculator 도구 사용.
"""
llm_with_tools = llm.bind_tools(tools)
def agent(state: State) -> dict:
messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
return {"messages": [llm_with_tools.invoke(messages)]}
def should_continue(state: State) -> Literal["tools", "end"]:
last = state["messages"][-1]
return "tools" if last.tool_calls else "end"
builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode(tools)) #langGraph 내장 도구
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {
"tools": "tools",
"end": END,
})
builder.add_edge("tools", "agent") # 도구 사용 후 다시 agent로 돌아와서 답변 생성
graph = builder.compile()
questions = [
"휴가 며칠 받아?",
"예산 변경 절차 알려주고, 12 곱하기 5도 계산해줘",
]
for q in questions:
print(f"\n{'=' * 60}\n질문: {q}\n{'=' * 60}")
result = graph.invoke({"messages": [HumanMessage(content=q)]})
# 모든 step 출력 — tool_calls 가 있으면 그것까지 보여줌
for m in result["messages"]:
label = type(m).__name__
if hasattr(m, "tool_calls") and m.tool_calls:
for tc in m.tool_calls:
print(f" [{label}] tool_call: {tc['name']}({tc['args']})")
elif hasattr(m, "content"):
content = m.content[:200] if m.content else ""
print(f" [{label}] {content}")
