포스트

2025년형 LLM RAG 에이전트 구현 튜토리얼: LangGraph + (OpenAI Responses/File Search)로 “검색→판단→생성→검증” 루프 만들기

2025년형 LLM RAG 에이전트 구현 튜토리얼: LangGraph + (OpenAI Responses/File Search)로 “검색→판단→생성→검증” 루프 만들기

들어가며

2023~2024년의 “단순 RAG”는 질문 → retrieve → LLM 답변의 직선 파이프라인이 주류였지만, 2025년 실무에서는 이 구조가 쉽게 한계에 부딪힙니다. 예를 들어 사용자의 질문이 (1) 검색이 필요 없는 상식인지, (2) 내부 문서 기반인지, (3) 웹 최신 정보가 필요한지, (4) 검색 결과 품질이 낮아 재질의/재랭킹이 필요한지 등을 매번 고정 파이프라인으로 처리하면 비용/지연/환각이 폭증합니다.

그래서 2025년에는 “Agentic RAG”가 표준에 가까워졌습니다. 핵심은 LLM이 도구(tool)를 ‘필요할 때만’ 호출하고, 결과를 보고 루프를 돌며 품질을 끌어올리는 제어 구조를 갖추는 것입니다. LangGraph는 이런 “상태(state)+분기(conditional edge)+루프(loop)”를 그래프 구조로 안정적으로 구현하도록 가이드합니다. (docs.langchain.com)
또한 OpenAI는 Responses API와 built-in tool(웹 검색, file search 등), 그리고 에이전트 오케스트레이션을 위한 Agents SDK를 공개해 “에이전트 + 검색”을 제품 레벨 구성요소로 제공하기 시작했습니다. (openai.com)


🔧 핵심 개념

1) Agentic RAG의 정의

  • RAG: LLM이 답변하기 전에 외부 지식(문서/DB/웹)에서 근거를 가져와 답변 품질을 올리는 방식
  • Agentic RAG: “항상 검색”이 아니라, LLM이 상황을 판단해 검색/재검색/요약/검증을 단계적으로 수행하는 방식
    • 예: 질문 난이도 분류 → 내부 문서 검색 → 근거 부족 시 질의 재작성(rewrite) → 재검색 → 답변 생성 → 근거 정합성 검사

LangGraph의 “retrieval agent” 튜토리얼도 이 포인트를 강조합니다. 즉, 에이전트가 retriever tool을 호출할지 말지 결정하는 것이 출발점입니다. (docs.langchain.com)

2) 왜 그래프(State Machine)가 필요한가

에이전트는 필연적으로 “여러 턴/여러 도구/여러 분기”를 갖습니다.

  • 상태 예시: messages, query, retrieved_docs, citations, retry_count
  • 노드 예시: route, retrieve, rewrite, generate, grounding_check
  • 엣지 예시: route -> retrieve 또는 route -> generate, grounding_fail -> rewrite

이걸 if-else로 덕지덕지 붙이면 관측(Tracing), 재실행, 루프 제한 같은 운영 요소가 망가지는데, LangGraph는 이를 그래프 프리미티브로 정리해줍니다. (docs.langchain.com)

3) “도구 표준화” 관점: MCP가 뜨는 이유

2025년에는 도구 연결이 프레임워크마다 제각각이어서 재사용이 어렵다는 문제가 커졌고, 이를 해결하려는 흐름 중 하나가 Model Context Protocol(MCP) 입니다. MCP는 JSON-RPC 기반으로 “tool/resource/prompt” 등을 서버가 표준 인터페이스로 제공하도록 정의합니다. (modelcontextprotocol.io)
즉, 장기적으로는 “RAG용 retriever”도 MCP tool로 노출해 에이전트/클라이언트가 바뀌어도 동일 도구를 재사용하는 방향이 유리합니다.


💻 실전 코드

아래 코드는 LangGraph로 Agentic RAG의 최소 실전형 뼈대를 만듭니다.

  • route: 검색 필요 여부 판단
  • retrieve: VectorStore 검색(예시는 로컬 Chroma)
  • rewrite: 검색이 부실하면 질의를 재작성
  • generate: 최종 답변 생성(근거 포함)
  • grounding_check: 답변이 근거에 “기댔는지” 간단 점검 후 루프

실행 전: pip install -U langgraph langchain langchain-community langchain-text-splitters chromadb langchain-openai
환경변수: OPENAI_API_KEY

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
from __future__ import annotations

from typing import TypedDict, List, Literal, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma


# -----------------------------
# 1) 상태 정의: 그래프가 들고 다닐 데이터
# -----------------------------
class RAGState(TypedDict):
    messages: List[BaseMessage]
    query: str
    retrieved: List[Document]
    retry: int


llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
emb = OpenAIEmbeddings()

# -----------------------------
# 2) (예시) 아주 작은 문서 코퍼스 구축
#    - 실무에서는 PDF/HTML/DB 등 ingestion 파이프라인을 별도로 둡니다.
# -----------------------------
raw_docs = [
    Document(page_content="LangGraph는 상태 기반 그래프로 LLM 워크플로우를 구성한다."),
    Document(page_content="Agentic RAG는 검색 여부 판단, 재질의, 검증 루프를 포함한다."),
    Document(page_content="Retrieval 품질이 낮으면 query rewrite, rerank, hybrid search를 고려한다."),
]

splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=40)
chunks = splitter.split_documents(raw_docs)

vs = Chroma.from_documents(chunks, embedding=emb, collection_name="rag_demo")
retriever = vs.as_retriever(search_kwargs={"k": 4})


# -----------------------------
# 3) 노드 구현
# -----------------------------
def route(state: RAGState) -> Literal["retrieve", "generate"]:
    """
    LLM이 '검색이 필요한 질문인지' 판단.
    """
    q = state["query"]
    prompt = ChatPromptTemplate.from_messages([
        ("system",
         "너는 라우터다. 사용자의 질문이 내부 지식(문서 검색) 기반 답변이 필요한지 판단한다. "
         "검색이 필요하면 'retrieve', 아니면 'generate'만 출력한다."),
        ("human", "{q}")
    ])
    decision = llm.invoke(prompt.format_messages(q=q)).content.strip().lower()
    return "retrieve" if "retrieve" in decision else "generate"


def retrieve(state: RAGState) -> RAGState:
    docs = retriever.get_relevant_documents(state["query"])
    return {**state, "retrieved": docs}


def rewrite(state: RAGState) -> RAGState:
    """
    검색 품질이 낮을 때 질의를 재작성(의도 보존 + 키워드 강화).
    """
    prompt = ChatPromptTemplate.from_messages([
        ("system",
         "너는 검색 질의 최적화기다. 원 질문의 의도를 보존하면서 검색에 유리하게 질의를 1개로 재작성하라."),
        ("human", "원 질문: {q}\n현재 검색 결과가 부실함. 더 구체적인 검색 질의로 바꿔줘.")
    ])
    new_q = llm.invoke(prompt.format_messages(q=state["query"])).content.strip()
    return {**state, "query": new_q, "retry": state["retry"] + 1}


def generate(state: RAGState) -> RAGState:
    """
    retrieved 컨텍스트가 있으면 근거 기반 답변 생성.
    """
    context = "\n\n".join([f"- {d.page_content}" for d in state.get("retrieved", [])])
    prompt = ChatPromptTemplate.from_messages([
        ("system",
         "너는 시니어 개발자다. 주어진 CONTEXT를 우선 근거로 사용해 답변하라. "
         "근거가 부족하면 '추가 정보 필요'를 명시하라."),
        ("human", "QUESTION: {q}\n\nCONTEXT:\n{context}")
    ])
    answer = llm.invoke(prompt.format_messages(q=state["query"], context=context)).content
    msgs = state["messages"] + [AIMessage(content=answer)]
    return {**state, "messages": msgs}


def grounding_check(state: RAGState) -> Literal["end", "rewrite"]:
    """
    매우 단순한 groundedness 체크:
    - retrieved가 비었거나,
    - 답변에 '추가 정보 필요'가 뜨거나,
    - 재시도 여지가 있으면 rewrite로 루프
    """
    if state["retry"] >= 2:
        return "end"
    if not state.get("retrieved"):
        return "rewrite"
    last = state["messages"][-1].content if state["messages"] else ""
    if "추가 정보 필요" in last:
        return "rewrite"
    return "end"


# -----------------------------
# 4) 그래프 연결
# -----------------------------
g = StateGraph(RAGState)

g.add_node("retrieve", retrieve)
g.add_node("rewrite", rewrite)
g.add_node("generate", generate)

g.add_conditional_edges(START, route, {
    "retrieve": "retrieve",
    "generate": "generate",
})

g.add_edge("retrieve", "generate")
g.add_conditional_edges("generate", grounding_check, {
    "rewrite": "rewrite",
    "end": END,
})
g.add_edge("rewrite", "retrieve")

app = g.compile(checkpointer=MemorySaver())


# -----------------------------
# 5) 실행
# -----------------------------
if __name__ == "__main__":
    question = "Agentic RAG에서 LangGraph를 쓰는 이유를 실무 관점으로 설명해줘"
    init_state: RAGState = {
        "messages": [HumanMessage(content=question)],
        "query": question,
        "retrieved": [],
        "retry": 0,
    }

    out = app.invoke(init_state)
    print(out["messages"][-1].content)

이 형태가 “2025년형 튜토리얼”로 중요한 이유는, 단순 RAG 샘플이 아니라 (1) 라우팅, (2) 검색, (3) 재질의 루프, (4) 종료 조건이 들어가서 실제 운영 시나리오에 가까워지기 때문입니다. 또한 LangGraph 문서가 말하는 핵심(상태/노드/조건부 엣지)을 그대로 코드로 옮겼습니다. (docs.langchain.com)


⚡ 실전 팁

1) 검색은 “항상”이 아니라 “조건부”로

  • 질문의 타입(FAQ/정의/정책/최신 뉴스/사내 문서)을 분류해 retrieve를 최소화해야 비용과 latency가 줄고, 불필요한 컨텍스트 주입으로 인한 성능 저하도 줄어듭니다. (LangGraph의 retrieval agent 방향과 일치) (docs.langchain.com)

2) retrieval 품질이 낮을 때의 “플랜 B”를 그래프에 박아라

  • query rewrite(키워드 강화), top-k 확대, rerank, hybrid search(BM25+vector) 같은 보정 전략은 “예외 처리”가 아니라 핵심 플로우입니다.
  • 중요한 건 “무한 루프 방지”: retry 카운터와 종료 조건을 상태에 반드시 두세요.

3) 관측(Observability)을 처음부터 설계

  • 에이전트는 디버깅이 곧 제품 품질입니다. LangGraph 계열에서는 LangSmith 같은 트레이싱 도구를 붙여 병목/환각 패턴을 찾는 접근이 문서에서도 권장됩니다. (docs.langchain.com)
  • OpenAI 쪽도 에이전트 빌딩 블록과 함께 tracing/inspect를 강조합니다. (openai.com)

4) 도구 연결의 미래: MCP 고려

  • 사내에서 retriever, 정책 조회, 권한 체크 같은 도구가 늘어나면 “프레임워크 종속”이 바로 비용이 됩니다. MCP처럼 JSON-RPC 기반 표준으로 tool을 노출하면 장기 유지보수성이 좋아집니다. (modelcontextprotocol.io)

5) OpenAI built-in File Search vs 직접 VectorStore

  • 빠르게 MVP를 만들고 운영 복잡도를 줄이려면 OpenAI의 file search 같은 managed retrieval을 고려할 수 있습니다(쿼리 최적화/metadata filtering/reranking 지원을 강조). (openai.com)
  • 반대로 규제/비용/커스터마이징(자체 reranker, hybrid, 도메인 특화 chunking)이 중요하면 직접 VectorStore+pipeline이 유리합니다. 실무에선 두 접근을 혼합하기도 합니다(핵심 문서는 managed, 대규모 로그/데이터는 사내 검색).

🚀 마무리

2025년의 RAG 구현은 “Vector DB 붙이고 끝”이 아니라, 에이전트 제어 구조(라우팅/루프/검증) + 도구 생태계(Responses API, LangGraph, MCP 등)를 어떻게 조합하느냐가 승부처입니다.
추천 다음 학습 루트는:

  • LangGraph의 “custom RAG agent” 패턴을 확장해 rerank/grounding evaluator/웹 검색 분기를 그래프에 추가하기 (docs.langchain.com)
  • OpenAI Responses API의 built-in tool(웹 검색, file search) 기반으로 도구 호출+관측을 제품 수준으로 끌어올리기 (openai.com)
  • 조직 단위로는 MCP를 염두에 두고 tool 인터페이스를 정리해 에이전트 프레임워크 교체 비용을 낮추는 구조를 설계하기 (modelcontextprotocol.io)

원하시면, 위 코드에 (1) reranker 추가, (2) “답변-근거 정합성” LLM grader, (3) 웹 검색 노드, (4) 캐시/메모리/세션 컨텍스트 유지까지 포함한 “프로덕션형 그래프”로 확장 버전도 이어서 작성해드릴게요.

이 기사는 저작권자의 CC BY 4.0 라이센스를 따릅니다.