개요
본 글에서는 Retrieval-Augmented Generation(RAG)과 AI 에이전트(AI Agent) 개념을 결합한 에이전틱 RAG(Agentic RAG)에 대해 상세히 다룹니다. 먼저 RAG의 검색 증강 생성 원리와 AI 에이전트의 지각-의사 결정 메커니즘을 개별적으로 설명하고, LangChain과 LangGraph 프레임워크를 활용하여 문서 처리, 벡터 저장소 구축, 지능형 질의응답 시스템의 전체 구현 과정을 제시합니다. 마지막으로 에이전틱 RAG 도입 시 고려해야 할 트리거 최적화, 환각 문제 등 실제 과제를 논의하며 개발자에게 기술적 방향성을 제시합니다.
1. RAG(Retrieval-Augmented Generation)란?
RAG는 대규모 언어 모델(LLM)이 더 정확하게 질문에 답변할 수 있도록 돕는 방식입니다. 대기업이나 개인이 보유한 문서(사내 규정, 학습 자료 등)에서 관련 정보를 찾아 LLM의 답변 생성을 보강합니다. RAG는 크게 두 단계로 구성됩니다: 첫째, 검색(Retrieval) 단계에서는 사용자 질문과 유사한 문서 조각을 벡터 데이터베이스에서 찾고, 둘째, 생성(Generation) 단계에서는 검색된 내용을 프롬프트에 포함시켜 LLM이 정확한 답변을 생성하도록 합니다.
RAG가 필요한 이유는 다음과 같습니다:
- 정보 필터링: 전체 문서를 프롬프트에 넣으면 불필요한 정보가 많아 LLM의 추론을 방해합니다. RAG는 정확히 관련된 부분만 추출합니다.
- 토큰 제한 해결: LLM에는 최대 입력 토큰 제한이 있습니다. RAG를 통해 필요한 정보만 선택적으로 전달할 수 있습니다.
RAG 구현 방식은 완전 수동, 프레임워크 활용, 수동+프레임워크 혼합으로 나뉩니다. 대부분의 기업은 유연성과 효율성을 갖춘 혼합 방식을 선호합니다.
2. LangChain을 이용한 RAG 질문-답변 시스템 구현
다음은 LangChain과 일부 수동 코드를 결합하여 RAG 파이프라인을 구축하는 예시입니다.
2.1 문서 분할
# ! pip install PyPDF2==3.0.1 langchain==0.3.9 langchain-ollama==0.2.1 langchain_community==0.3.9 langchain_milvus==0.1.7 langgraph==0.2.56
from PyPDF2 import PdfReader
def load_pdf_text(pdf_paths):
"""PDF 파일에서 텍스트를 추출합니다."""
full_text = ""
for path in pdf_paths:
reader = PdfReader(path)
for page in reader.pages:
full_text += page.extract_text()
return full_text
# 예시 PDF 파일 경로
content = load_pdf_text(['./01_대모델_응용_에이전트_기술_트렌드.pdf'])
print(content[:200]) # 상위 200자 출력 확인텍스트를 LangChain의 Document 객체로 감싼 후, RecursiveCharacterTextSplitter를 이용해 적절한 크기의 청크로 분할합니다.
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
documents = [Document(page_content=content)]
chunk_size = 1000
chunk_overlap = 300
text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
splits = text_splitter.split_documents(documents)
print(f"총 {len(splits)}개의 청크로 분할되었습니다.")2.2 임베딩 모델 사용
Ollama를 통해 오픈소스 임베딩 모델(bge-m3)을 로드합니다.
from langchain_ollama import OllamaEmbeddings
embeddings = OllamaEmbeddings(
base_url="http://192.168.110.131:11434", # 실제 엔드포인트로 교체 필요
model="bge-m3",
)
# 임베딩 테스트
test_vector = embeddings.embed_query("AI 에이전트 개념")
print(f"벡터 차원: {len(test_vector)}") # bge-m3는 1024차원 출력2.3 벡터 데이터베이스 구축
Milvus (Zilliz Cloud)를 사용하여 무료 온라인 인스턴스에 연결합니다.
from langchain_milvus import Milvus
vector_store = Milvus.from_documents(
documents=splits,
collection_name="agentic_rag_demo",
embedding=embeddings,
connection_args={
"uri": "https://in03-xxxxxx.serverless.gcp-us-west1.cloud.zilliz.com", # 실제 URI
"user": "your_user",
"password": "your_password",
},
)
print("벡터 저장소에 데이터가 저장되었습니다.")2.4 LLM(생성 모델) 연결
Ollama를 통해 QWQ:32B 모델을 연결합니다.
from langchain_ollama import ChatOllama
llm = ChatOllama(
base_url="http://192.168.110.131:11434",
model="qwq:latest",
)
# LLM 테스트
response = llm.invoke("안녕하세요, 자기소개를 해주세요.")
print(response.content)2.5 질문-답변 체인 구성
프롬프트 템플릿을 정의하고, LLM 및 출력 파서와 연결합니다.
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
prompt = PromptTemplate(
template="""You are an assistant for question-answering tasks.
Use the following pieces of retrieved context to answer the question.
If you don't know the answer, just say that you don't know.
Use three sentences maximum and keep the answer concise:
Question: {question}
Context: {context}
Answer:
""",
input_variables=["question", "context"],
)
rag_chain = prompt | llm | StrOutputParser()
# 검색기 생성 및 테스트
retriever = vector_store.as_retriever(search_kwargs={"k": 3})
query = "AI 에이전트란 무엇인가요?"
retrieved_docs = retriever.invoke(query)
# RAG 체인 실행
final_answer = rag_chain.invoke({"context": retrieved_docs, "question": query})
print(f"질문: {query}")
print(f"답변: {final_answer}")위 과정은 수동 + 프레임워크 방식의 간단한 RAG 예시이며, 단일 지식 소스 및 단일 검색 호출이라는 한계가 있습니다.
3. AI 에이전트 개요
AI 에이전트는 환경을 지각하고, 정보를 처리하며, 목표 달성을 위해 행동을 취하는 소프트웨어 시스템입니다. 에이전트는 단순한 LLM 호출을 넘어, 동적으로 계획을 수립하고 외부 도구(API, 웹 검색 등)를 호출하여 복잡한 작업을 수행합니다.
대표적인 패러다임은 ReAct(Reason + Act) 입니다. ReAct는 추론(Thought)과 행동(Action)을 교차하며 환경 관찰(Observation)을 기반으로 다음 단계를 결정합니다. 이는 인간의 문제 해결 방식을 모방한 것입니다.
# 간단한 CoT 예시
print(llm.invoke("로저는 테니스 공 5개를 가지고 있습니다. 2캔의 테니스 공을 더 샀고, 각 캔에는 3개의 공이 들어 있습니다. 로저는 지금 몇 개의 공을 가지고 있습니까?").content)4. LangGraph를 이용한 ReAct 에이전트 구현
LangGraph는 LangChain 기반의 그래프 구조를 통해 에이전트 워크플로우를 정의합니다.
4.1 상태 및 노드 정의
from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
# 실시간 웹 검색 도구 정의
from langchain_core.tools import tool
import requests, json
@tool
def fetch_web_info(query: str) -> str:
"""실시간 인터넷 정보를 검색합니다."""
url = "https://google.serper.dev/search"
payload = json.dumps({"q": query, "num": 3})
headers = {
'X-API-KEY': 'your_serper_api_key', # 실제 키로 교체
'Content-Type': 'application/json'
}
response = requests.post(url, headers=headers, data=payload)
data = response.json()
if 'organic' in data:
return json.dumps(data['organic'], ensure_ascii=False)
return json.dumps({"error": "검색 결과 없음"}, ensure_ascii=False)4.2 그래프 구축
from langgraph.graph import StateGraph, END
tools = [fetch_web_info]
model = llm.bind_tools(tools)
def agent_node(state: AgentState):
system_prompt = SystemMessage("You are a helpful AI assistant.")
response = model.invoke([system_prompt] + state["messages"])
return {"messages": [response]}
def tool_node(state: AgentState):
outputs = []
for tool_call in state["messages"][-1].tool_calls:
tool_result = tools_by_name[tool_call["name"]].invoke(tool_call["args"])
outputs.append(ToolMessage(content=json.dumps(tool_result), name=tool_call["name"], tool_call_id=tool_call["id"]))
return {"messages": outputs}
def should_continue(state: AgentState) -> str:
if state["messages"][-1].tool_calls:
return "continue"
return "end"
# 그래프 구성
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"continue": "tools", "end": END})
workflow.add_edge("tools", "agent")
graph = workflow.compile()
# 실행
inputs = {"messages": [("user", "최근 OpenAI 관련 뉴스가 있나요?")]}
for event in graph.stream(inputs, stream_mode="values"):
event["messages"][-1].pretty_print()5. 에이전틱 RAG 구현
에이전틱 RAG는 전통적인 RAG에 AI 에이전트를 통합하여, 검색 과정을 더 지능적으로 제어하고 다양한 도구(벡터 저장소, 웹 검색, 외부 API)를 조합합니다.
5.1 사용자 정의 RAG 도구 정의
@tool
def query_knowledge_base(question: str) -> str:
"""개인 지식 베이스에서 질문에 대한 답변을 검색합니다."""
# 재사용: 미리 구축된 vector_store 사용
retriever = vector_store.as_retriever(search_kwargs={"k": 2})
docs = retriever.invoke(question)
context = "\n".join([doc.page_content for doc in docs])
prompt = PromptTemplate(
template="Use the following context to answer the question concisely:\nContext: {context}\nQuestion: {question}\nAnswer:",
input_variables=["context", "question"],
)
chain = prompt | llm | StrOutputParser()
return chain.invoke({"context": context, "question": question})5.2 통합 에이전트 그래프
tools = [fetch_web_info, query_knowledge_base]
model = llm.bind_tools(tools)
# 동일한 그래프 구조 재사용 (위의 graph 코드와 동일)
# tools_by_name = {tool.name: tool for tool in tools}
# ... (그래프 정의 코드 반복)
# 실행 예시: 지식 베이스 + 웹 검색 통합 질문
inputs = {"messages": [("user", "내 지식 베이스와 인터넷을 활용하여 'AI 에이전트'를 설명해주세요. 한국어로 답변해주세요.")]}
for event in graph.stream(inputs, stream_mode="values"):
event["messages"][-1].pretty_print()위 실습에서 확인할 수 있듯이, 에이전틱 RAG는 단일 검색을 넘어 다중 소스 정보를 통합하지만, 여전히 모델이 도구 호출을 생략하거나 환각을 생성하는 문제가 발생할 수 있습니다. 이러한 문제들은 실제 애플리케이션 개발 시 최적화가 필요한 부분입니다.
참고 자료
- LangChain 공식 문서: https://python.langchain.com/
- LangGraph 공식 문서: https://langchain-ai.github.io/langgraph/
- ReAct 논문: ReAct: Synergizing Reasoning and Acting in Language Models