RAG와 MCP를 사용한 AI 에이전트: 외부 도구 통합

RAG와 외부 도구를 MCP로 활용한 AI 에이전트 구축

현대적인 AI 에이전트는 텍스트 생성을 넘어 외부 API와 상호작용하고, 상태를 관리하며, 맥락에 기반한 결정을 내립니다. 이 글에서는 내부 문서 작업을 위한 RAG와 외부 서비스(날씨와 GitHub Issues)에 접근하기 위한 Model Context Protocol (MCP)을 결합한 에이전트를 구축하겠습니다. 이 아키텍처는 LangGraph와 FastMCP을 기반으로 하며, 커스텀 어댑터를 작성하지 않고도 시스템을 확장할 수 있습니다.

아키텍처 구성 요소: ReAct, RAG 및 MCP

우리의 AI 에이전트는 ReAct 사이클로 작동합니다: 모델이 쿼리를 분석 → 도구 선택 → 결과 획득 → 최종 답변까지 사이클 반복. 이는 단계 순서가 엄격히 미리 정의된 선형 체인과 다릅니다. LangGraph는 의사결정을 지원하기 위해 사용되며, 명시적인 상태 관리와 전환 그래프를 제공합니다.

여기서 RAG는 별도의 모듈이 아니라 에이전트의 도구 중 하나로 작동합니다. 사용자가 내부 문서에 대해 물어보면, 에이전트는 모델의 매개변수 메모리에 의존하는 대신 search_documentation을 호출합니다. 이는 환각(hallucinations) 위험을 줄이고 응답 정확성을 높입니다.

Model Context Protocol (MCP)은 외부 도구 연결을 표준화합니다. 각 API마다 래퍼를 수동으로 작성하는 대신, JSON-RPC를 통해 도구를 제공하는 MCP 서버를 실행합니다. 에이전트는 클라이언트를 통해 서버에 연결하며, 모든 도구가 자동으로 사용 가능해집니다. 이는 프로덕션 환경에서 통합 수가 빠르게 증가할 때 특히 유용합니다.

기술 스택 및 선택 이유

기술 선택은 확장성, 비동기 지원, MCP 호환성 요구사항에 따라 결정되었습니다:

• Python 3.11+: 현대적인 async/await 지원, 안정적인 생태계.
• LangGraph + LangChain: LangGraph는 상태와 ReAct 사이클 제어를 제공하며, LangChain은 모델과 도구 추상화를 담당합니다.
• FastMCP: 데코레이터를 통해 MCP 서버를 빠르게 생성하는 Python 라이브러리. 2.0 버전은 활발히 개발 중이며 클라이언트 기능도 포함합니다.
• ChromaDB: 프로토타입에 이상적인 경량 임베디드 벡터 DB.
• OpenAI text-embedding-3-small: 임베딩에 최적의 가격/성능 비율.
• Claude 또는 GPT-4: 함수 호출을 지원하는 모든 모델.

외부 도구를 위한 MCP 서버 구현

첫 번째 단계는 날씨 조회와 GitHub Issue 생성이라는 두 도구를 제공하는 MCP 서버 생성입니다. 서버는 FastMCP을 기반으로 하며 stdio 전송을 통해 서브프로세스로 실행됩니다.

# tools_server.py
import os
import httpx
from fastmcp import FastMCP

mcp = FastMCP("ExternalTools")

@mcp.tool()
async def get_weather(city: str) -> str:
    async with httpx.AsyncClient() as client:
        geo_response = await client.get(
            "https://geocoding-api.open-meteo.com/v1/search",
            params={"name": city, "count": 1, "language": "en", "format": "json"},
            timeout=10.0
        )
        geo_data = geo_response.json()
        if not geo_data.get("results"):
            return f"Gorod '{city}' not found"
        
        location = geo_data["results"][0]
        lat, lon = location["latitude"], location["longitude"]
        
        weather_response = await client.get(
            "https://api.open-meteo.com/v1/forecast",
            params={"latitude": lat, "longitude": lon, "current_weather": True},
            timeout=10.0
        )
        current = weather_response.json()["current_weather"]
        return f"Pogoda in {location['name']}: {current['temperature']}°C, veter {current['windspeed']} km/ch"

@mcp.tool()
async def create_github_issue(repo: str, title: str, body: str, labels: list[str] | None = None) -> str:
    github_token = os.environ.get("GITHUB_TOKEN")
    if not github_token:
        return "Error: GITHUB_TOKEN not ustanovlen"
    
    url = f"https://api.github.com/repos/{repo}/issues"
    headers = {"Authorization": f"Bearer {github_token}", "Accept": "application/vnd.github+json"}
    payload = {"title": title, "body": body}
    if labels: payload["labels"] = labels
    
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=payload, headers=headers)
        if response.status_code == 201:
            return f"Issue sozdan: {response.json()['html_url']}"
        else:
            return f"Error: {response.status_code}"

if __name__ == "__main__":
    mcp.run(transport="stdio")

구현의 주요 특징:

• @mcp.tool() 데코레이터는 타입 어노테이션과 docstring을 기반으로 매개변수 JSON 스키마를 자동 생성합니다.
• 비동기 HTTP 클라이언트를 사용하여 블로킹되지 않는 요청을 처리합니다.
• 각 도구 로직에 오류 처리가 내장되어 있어 에이전트는 예외 대신 사람이 읽을 수 있는 메시지를 받습니다.
• Stdio 전송은 클라이언트가 서버를 자식 프로세스로 실행하고 stdin/stdout으로 통신할 수 있게 합니다.

에이전트 도구로서 RAG 모듈 설정

RAG 시스템은 PDF 문서를 인덱싱하고 LangChain 호환 도구를 통해 검색 기능을 제공합니다. 중요한 점은 이 도구가 MCP 도구와 동일한 풀에 등록되어 에이전트가 외부 API와 함께 선택할 수 있다는 것입니다.

# rag_tool.py
import os
from pathlib import Path
from langchain_openai import OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain.tools import tool

CHROMA_PATH = "./chroma_db"
DATA_PATH = "./data"

@tool
def search_documentation(query: str) -> str:
    """Ischet relevantnye fragmenty in documentation by request."""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma(persist_directory=CHROMA_PATH, embedding_function=embeddings)
    docs = vectorstore.similarity_search(query, k=3)
    return "\n\n".join([f"Withtranitsa {d.metadata.get('page', 'N/A')}: {d.page_content[:500]}..." for d in docs])

# Function initsializatsii database (vyzyvaetsya when starte agenta)
def initialize_vector_store(force_reload=False):
    if os.path.exists(CHROMA_PATH) and not force_reload:
        return
    
    documents = []
    for pdf_file in Path(DATA_PATH).glob("*.pdf"):
        loader = PyPDFLoader(str(pdf_file))
        documents.extend(loader.load())
    
    if not documents:
        return
    
    splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
    chunks = splitter.split_documents(documents)
    
    Chroma.from_documents(chunks, OpenAIEmbeddings(model="text-embedding-3-small"), persist_directory=CHROMA_PATH)

MCP 클라이언트 통합 및 LangGraph에서의 에이전트 조립

마지막 단계는 MCP 도구와 RAG를 결합한 에이전트 조립입니다. langchain-mcp-adapters를 사용하여 MCP 도구를 LangChain 형식으로 변환한 후, RAG 도구와 함께 그래프에 등록합니다.

# agent.py
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage
from langchain_mcp_adapters import McpToolNode
from rag_tool import search_documentation
import subprocess
import asyncio

# Run MCP-server how podprotsessa
mcp_process = subprocess.Popen(
    ["python", "tools_server.py"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)

# Withbudynek MCP-client and adaptera narzędzieov
mcp_node = McpToolNode(
    server_command=["python", "tools_server.py"],
    transport="stdio"
)

# Receiving list narzędzieov from MCP-server
mcp_tools = asyncio.run(mcp_node.get_tools())
all_tools = [*mcp_tools, search_documentation]

# Initialization modeli and graph
model = ChatAnthropic(model="claude-3-5-sonnet-20240620").bind_tools(all_tools)

# Withstoyanie graph
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]

# Uzly graph
def call_model(state):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state):
    last_message = state["messages"][-1]
    return "continue" if last_message.tool_calls else "end"

# Postroenie graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", mcp_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {"continue": "tools", "end": END})
workflow.add_edge("tools", "agent")

app = workflow.compile()

# Example run
result = app.invoke({"messages": [HumanMessage(content="Uznay pogodu in Londone and sozday issue in repo/my-project with etim rezultatom")]})
print(result["messages"][-1].content)

중요한 점:

• MCP는 새로운 도구마다 커스텀 어댑터를 작성할 필요를 없애줍니다. 서버에서 @mcp.tool() 데코레이터만 추가하면 됩니다.
• RAG는 일반 도구로 통합되어 에이전트가 문서 검색과 외부 API 호출 사이에서 선택할 수 있습니다.
• LangGraph는 복잡한 다단계 작업에 필수적인 투명한 상태 관리와 ReAct 루프 제어를 제공합니다.
• 비동기 도구 구현으로 에이전트 메인 스레드를 블로킹하지 않습니다.
• Stdio 전송은 네트워크 종속성 없이 동일 컨테이너나 머신에서 서버와 에이전트를 실행할 수 있게 해 배포를 단순화합니다.

— Editorial Team