MCP RAG

MCP를 이용하여 RAG를 편리하게 이용할 수 있습니다. 여기에서는 완전 관리형 RAG 서버스인 knowledge base와 관리형 RAG인 OpenSearch에서 MCP를 활용하는 방법과 RAG의 성능향상 기법인 advanced RAG를 활용하는 방법을 설명합니다. 전체적인 architecture는 아래와 같고 RAG를 활용할 수 있는 4가지 형태의 MCP를 설명하고 있습니다. 사용자가 문서를 Amazon S3로 업로드하면 Knowledge Base에서는 sync를 통해 문서를 가져와서 Amazon Opensearch Serverless로 문서를 적재합니다. 이때 미리 지정한 embedding model을 이용하고 multi modal을 통해 분석된 정보를 활용할 수 있습니다. 또한 Amazon S3에 문서가 업로드 될 때 발생하는 event를 AWS Lambda (s3-event-manager)가 받아서 SQS에 전달한 다음에 순차적으로 AWS Lambda (document-manager)가 embedding 및 multi modal 분석을 통해 얻어진 context를 managed OpenSearch에 적재할 수 있습니다. Amazon EC2에 있는 AI application은 MCP client / server 구조를 이용하여 MCP 서버의 tool들을 활용할 수 있습니다. 이때 RAG를 활용할 때에 아래 그림과 같이 (1) AWS MCP (Knowledge Base) (2) MCP Lambda (Knowledge Base) (3) OpenSearch MCP (4) MCP Lambda (OpenSearch)의 4가지 방법중에 한가지를 선택하여 활용할 수 있습니다. RAG를 이용해 필요한 OpenSearch, lambda, SQS 등의 인프라는 AWS CDK를 이용하여 쉽게 배포할 수 있습니다.

문서의 각 페이지들을 캡춰하여 이미지를 분석하면 문서 안의 표나 이미지에 대한 좀더 많은 정보를 가져올 수 있습니다. 문서로 부터 페이지 이미지를 추출하여 분석하기 위하여Agentic RAG 구현하기와 같이 event 형태로 데이터 처리 파이프라인을 만들어서 활용합니다. 이때, 이벤트 처리는lambda-s3-event-manager로 수행하고, 문서의 처리는lambda-document-manager로 구현합니다.

문서의 각 페이지를 이미지로 변환한 후에 multimodal을 통해 분석합니다. 이때 맥락에 맞는 이미지 분석을 위해 contextual embedding을 활용합니다. 상세한 코드는lambda-document-manager을 참조합니다. Contextual embedding을 위해 managed OpenSearch를 활용합니다.

session = boto3.Session(region_name=region) credentials = session.get_credentials() awsauth = AWS4Auth( credentials.access_key, credentials.secret_key, region, 'es', session_token=credentials.token ) os_client = OpenSearch( hosts=[{ 'host': opensearch_url.replace("https://", ""), 'port': 443 }], http_compress=True, http_auth=awsauth, use_ssl=True, verify_certs=True, ssl_assert_hostname=False, ssl_show_warn=False, connection_class=RequestsHttpConnection )

이제 vectorstore 정의해서 문서를 추가하거나 삭제할때에 활용합니다.

from langchain_community.vectorstores.opensearch_vector_search import OpenSearchVectorSearch vectorstore = OpenSearchVectorSearch( index_name=index_name, is_aoss = False, embedding_function=bedrock_embeddings, opensearch_url=opensearch_url, http_auth=awsauth, connection_class=RequestsHttpConnection )

각 페이지가 전체 문서에서 어떤 의미를 가지는지 contextual_text를 추출하여 활용합니다.

def get_contextual_text(whole_text, splitted_text, llm): contextual_template = ( "<document>" "{WHOLE_DOCUMENT}" "</document>" "Here is the chunk we want to situate within the whole document." "<chunk>" "{CHUNK_CONTENT}" "</chunk>" "Please give a short succinct context to situate this chunk within the overall document for the purposes of improving search retrieval of the chunk." "Answer only with the succinct context and nothing else in English." "Put it in <result> tags." ) contextual_prompt = ChatPromptTemplate([ ('human', contextual_template) ]) contextual_text = "" contexual_chain = contextual_prompt | llm response = contexual_chain.invoke( { "WHOLE_DOCUMENT": whole_text, "CHUNK_CONTENT": splitted_text } ) output = response.content return output[output.find('<result>')+8:output.find('</result>')]

페이지 이미지에서 텍스트를 추출하기 위해 이미지 사이즈를 조정한 후에 contextual text와 함께 OpenSearch에 등록하여 활용합니다.

def store_image_for_opensearch(key): image_obj = s3_client.get_object(Bucket=s3_bucket, Key=key) image_content = image_obj['Body'].read() img = Image.open(BytesIO(image_content)) width, height = img.size print(f"width: {width}, height: {height}, size: {widthheight}") isResized = False while(widthheight > 5242880): width = int(width/2) height = int(height/2) isResized = True print(f"width: {width}, height: {height}, size: {width*height}") buffer = BytesIO() img.save(buffer, format="PNG") img_base64 = base64.b64encode(buffer.getvalue()).decode("utf-8") llm = get_model() text = extract_text(llm, img_base64) extracted_text = text[text.find('<result>')+8:text.find('</result>')] contextual_text = object_meta["contextual_text"] summary = summary_image(llm, img_base64, contextual_text) image_summary = summary[summary.find('<result>')+8:summary.find('</result>')] contents = f"[이미지 요약]\n{image_summary}\n\n[추출된 텍스트]\n{extracted_text}" page = object_meta["page"] docs = [] docs.append( Document( page_content=contents, metadata={ 'name': key, 'page': page, 'url': path+parse.quote(key) } ) ) return add_to_opensearch(docs)

여기서 multimodal을 이용해 이미지에서 텍스트를 추출하는 함수는 아래와 같습니다.

def summary_image(llm, img_base64, contextual_text): query = "이미지가 의미하는 내용을 풀어서 자세히 알려주세요. markdown 포맷으로 답변을 작성합니다." if contextual_text: query += "\n아래 <reference>는 이미지와 관련된 내용입니다. 이미지 분석시 참고하세요. \n<reference>\n"+contextual_text+"\n</reference>" messages = [ HumanMessage( content=[ { "type": "image_url", "image_url": { "url": f"data:image/png;base64,{img_base64}", }, }, { "type": "text", "text": query }, ] ) ] result = llm.invoke(messages) extracted_text = result.content return extracted_text

문서 검색의 정확도를 높이면서 Context를 충분히 사용하기 위해서는 Parent Child Chunking을 적용하여야 합니다. 이를 위해 아래와 같이 RecursiveCharacterTextSplitter로 parent와 child를 각각 나눕니다.

parent_splitter = RecursiveCharacterTextSplitter( chunk_size=2000, chunk_overlap=100, separators=["\n\n", "\n", ".", " ", ""], length_function = len, ) child_splitter = RecursiveCharacterTextSplitter( chunk_size=400, chunk_overlap=50, length_function = len, )

먼저 parent chunk들을 OpenSearch에 넣고 id들을 확인합니다. child chunk의 meta에 parent chunk의 id를 추가해서 검색시 child chunk를 하고, 실제 context는 parent의 id로 조회한 parent의 text를 활용합니다. Contextual text는 실제 사용하게 될 parent chunk의 특징을 설명하여야 하므로, parent chunk로 얻은 contextual text를 child chunk에 추가하여 활용합니다. 이후 child chunk들도 OpenSearch에 등록하여 id들을 확인합니다. parent/child의 id들을 파일 meta에 저장하였다가 문서 업데이트/삭제시에 활용합니다.

splitted_docs = parent_splitter.split_documents(docs) parent_docs, contexualized_chunks = get_contextual_docs_using_parallel_processing(docs[-1], splitted_docs) for i, doc in enumerate(parent_docs): doc.metadata["doc_level"] = "parent" parent_doc_ids = vectorstore.add_documents(parent_docs, bulk_size = 10000) ids = parent_doc_ids for i, doc in enumerate(splitted_docs): _id = parent_doc_ids[i] child_docs = child_splitter.split_documents([doc]) for _doc in child_docs: _doc.metadata["parent_doc_id"] = _id _doc.metadata["doc_level"] = "child" contexualized_child_docs = [] for _doc in child_docs: contexualized_child_docs.append( Document( page_content=contexualized_chunks[i]+"\n\n"+_doc.page_content, metadata=_doc.metadata ) ) child_docs = contexualized_child_docs child_doc_ids = vectorstore.add_documents(child_docs, bulk_size = 10000) ids += child_doc_ids

완전 관리형 RAG 서비스인 knowledge base는 S3와 같은 storage에 대해서 sync 기능을 제공하므로써 편리하게 사용할 수 있습니다. 하지만 OCR이나 contextual embedding을 이용할 경우에는 custom으로 lambda등을 활용하여 직접 파싱후 넣어주여야 합니다.

Amazon Bedrock Knowledge Base Retrieval MCP Server와 같이 AWS에서 제공하는 MCP를 이용하여 Amazon Knowledge Base의 문서를 조회할 수 있습니다. 이때 별도로 조회하는 인프라가 필요하지 않으므로 쉽게 구현이 가능합니다. 이때의 MCP 설정은 아래와 같습니다.

{ "mcpServers": { "awslabs.bedrock-kb-retrieval-mcp-server": { "command": "uvx", "args": ["awslabs.bedrock-kb-retrieval-mcp-server@latest"], "env": { "AWS_PROFILE": "your-profile-name", "AWS_REGION": "us-east-1", "FASTMCP_LOG_LEVEL": "ERROR", "KB_INCLUSION_TAG_KEY": "optional-tag-key-to-filter-kbs", "BEDROCK_KB_RERANKING_ENABLED": "false" }, "disabled": false, "autoApprove": [] } } }

AWS의 knowledge base MCP는 knowledge base를 조회하여 특정 tag(기본은 mcp-tag)를 knowledge base를 찾은 다음에 query를 수행합니다. 따라서 knowledge base를 생성할 때에 아래와 같이 tag를 설정하여야 합니다.

Amazon Bedrock Knowledge Base Retrieval MCP Server에서는 GetKnowledgeBases를 resource로 호출하는데, LangGraph에서 참조가 안되는 이슈가 있습니다. 따라서, 여기에서는 소스를 복사하여mcp_server_knowledge_base.py와 같이 tool로 수정하여서 활용하고 있습니다. 이때의 MCP 설정은 아래와 같습니다.

{ "mcpServers": { "knowledge_base_custom": { "command": "python", "args": [ "application/mcp_server_knowledge_base.py" ], "env": { "KB_INCLUSION_TAG_KEY": "mcp-rag" } } } }

MCP로 knowledge base를 조회하기 위해서, lambda를 이용하면 사용자의 목적에 맞는 RAG 동작을 구현할 수 있습니다. 아래에서는 Lambda를 이용한 custom MCP 서버를 정의하는것을 설명합니다. Lambda를 이용해 knowledge base를 조회하는 것은lambda-knowledge-base에 관련된 코드가 있습니다. 아래와 같이 knowledge_base_search를 tool로 정의합니다.

@mcp.tool() def knowledge_base_search(keyword: str) -> list: """ Search the knowledge base with the given keyword. keyword: the keyword to search return: the result of search """ return rag.retrieve_knowledge_base(keyword)

knowledge_base_search는mcp_knowledge_base.py에 정의된 retrieve_knowledge_base와 같이 lambda를 직접 호출하는 방식으로 knowledge base의 문서들을 조회합니다.

def retrieve_knowledge_base(query): lambda_client = boto3.client( service_name='lambda', region_name=bedrock_region ) functionName = f"knowledge-base-for-{projectName}" mcp_env = utils.load_mcp_env() grading_mode = mcp_env['grading_mode'] multi_region = mcp_env['multi_region'] payload = { 'function': 'search_rag', 'knowledge_base_name': knowledge_base_name, 'keyword': query, 'top_k': numberOfDocs, 'grading': grading_mode, 'model_name': model_name, 'multi_region': multi_region } output = lambda_client.invoke( FunctionName=functionName, Payload=json.dumps(payload), ) payload = json.load(output['Payload']) return payload['response']

Lambda로 MCP 서버를 구현하면 추가적인 인프라가 필요하지만, grading을 통해 관련도가 낮은 문서를 제외하는 것과 같은 custom 작업을 수행할 수 있고, knowledge base 리스트를 조회하지 않고 바로 query를 하므로 더 빠른 응답을 얻을 수 있습니다.

OpenSearch MCP를 이용하면 추가적인 리소스 없이 바로 OpenSearch를 조회할 수 있습니다. 하지만 2025년 6월 현재는 텍스트 검색만을 제공하여 성능상 제한이 있습니다. OpenSearch MCP를 이용할 때에는 아래 config를 활용합니다.

{ "mcpServers": { "opensearch-mcp-server": { "command": "uvx", "args": [ "opensearch-mcp-server-py" ], "env": { "OPENSEARCH_URL": managed_opensearch_url, "AWS_REGION": aws_region, "AWS_ACCESS_KEY_ID": credentials.access_key, "AWS_SECRET_ACCESS_KEY": credentials.secret_key } } } }

OCR, contextual embedding과 같은 custom한 RAG를 구현할 때에는 직접 OpenSearch에 넣고 조회하여야 합니다. 이를 조회할 때에는 lambda로 custom MCP 서버를 정의하거나OpenSearch MCP을 사용할 수 있습니다. Lambda로 OpenSearch를 조회하는 것은lambda-opensearch와 같이 구현할 수 있습니다. 이를 custom MCP 서버로 구현할 때에는mcp_server_lambda_opensearch.py와 같이 구현합니다.

@mcp.tool() def opensearch_search(keyword: str) -> list: """ Search the knowledge base with the given keyword. keyword: the keyword to search return: the result of search """ return rag.retrieve_opensearch(keyword)

여기서mcp_opensearch.py와 같이lambda-opensearch로 직접 요청하여 Lambda가 가져온 OpenSearch 문서를 얻을 수 있습니다. Lambda에서는 OpenSearch로 문서 조회뿐 아니라, 문서의 관련도를 기준으로 grading을 수행할 수 있습니다.

def retrieve_opensearch(query): lambda_client = boto3.client( service_name='lambda', region_name=bedrock_region ) functionName = f"opensearch-for-{projectName}" mcp_env = utils.load_mcp_env() grading_mode = mcp_env['grading_mode'] multi_region = mcp_env['multi_region'] payload = { 'function': 'search_rag', 'keyword': query, 'top_k': numberOfDocs, 'grading': grading_mode, 'model_name': model_name, 'multi_region': multi_region } output = lambda_client.invoke( FunctionName=functionName, Payload=json.dumps(payload), ) payload = json.load(output['Payload']) return payload['response']

AgentCore Gateway는 Lambda로 MCP서버를 제공할 수 있습니다.config.json.sample을 복사하여 config.json 파일을 생성합니다.

cd gateway/opensearch-retriever && cp config.json.sample config.json

{ "projectName": "mcp-rag", "opensearch_url": "https://search-mcp-rag-mxtkul2z3qv5iiqprb7q3jx4wy.us-west-2.es.amazonaws.com", "sharing_url": "https://d20lfnyi6fvd87.cloudfront.net" }

git clone https://github.com/kyopark2014/mcp-rag/

cd mcp-rag && pip install -r requirements.txt

cd cdk-mcp-rag/ && cdk deploy --all

config.json은 agent의 동작에 필요한 정보를 가지고 있고,.gitignore에 의해 git으로 공유 되지 않습니다. 생성된 config.json의 셈플은 아래와 같습니다.

{ "projectName":"mcp-rag", "accountId":"862926741992", "region":"us-west-2", "roleKnowledgeBase":"arn:aws:iam::862926741992:role/role-knowledge-base-for-mcp-rag-us-west-2", "collectionArn":"arn:aws:aoss:us-west-2:862926741992:collection/8krsnuq4it9gpl70i3u6", "serverless_opensearch_url":"https://8krsnuq4it9gpl70i3u6.us-west-2.aoss.amazonaws.com", "managed_opensearch_url":"https://search-mcp-rag-mxtkul3z3qv5iiqprb7q3jx4wy.us-west-2.es.amazonaws.com", "knowledge_base_role":"arn:aws:iam::862926741992:role/role-knowledge-base-for-mcp-rag-us-west-2", "s3_bucket":"storage-for-mcp-rag-862926741992-us-west-2", "s3_arn":"arn:aws:s3:::storage-for-mcp-rag-862926741992-us-west-2", "sharing_url":"https://d3mo4kqj5cjiuy.cloudfront.net" }

이후Secret Manager에 접속하여 아래와 같은 credential을 입력합니다.

만약 streamlit이 설치되어 있지 않다면streamlit을 참조하여 설치합니다. 이후 아래와 같이 실행합니다.

기업용 RAG는 왜 실패하는가…엔터프라이즈 환경에서 RAG를 확장하는 법

RAG-focused MCP subregistry and server to discover and route to retrieval-capable MCP servers with semantic search, filters, and explainable ranking.