MCP RAG
Un servidor gestionado de Generación Aumentada por Recuperación (RAG) que utiliza MCP, integrado con bases de conocimiento y OpenSearch.
Documentación
MCP RAG
Puede utilizar RAG convenientemente mediante MCP. Aquí se explica cómo utilizar MCP con knowledge base, un servicio RAG totalmente administrado, y OpenSearch, un RAG administrado, así como el uso de advanced RAG, una técnica para mejorar el rendimiento de RAG. La arquitectura general es la siguiente y se describen 4 tipos de MCP que permiten utilizar RAG. Cuando el usuario sube documentos a Amazon S3, Knowledge Base los importa mediante sync y los carga en Amazon Opensearch Serverless. En este momento se utiliza un modelo de embedding predefinido y se puede aprovechar la información analizada mediante multimodal. Además, cuando se sube un documento a Amazon S3, el evento generado es recibido por AWS Lambda (s3-event-manager), que lo envía a SQS, y luego AWS Lambda (document-manager) carga secuencialmente el contexto obtenido mediante embedding y análisis multimodal en managed OpenSearch. La aplicación de IA en Amazon EC2 puede utilizar las herramientas del servidor MCP mediante la estructura de cliente/servidor MCP. Al utilizar RAG, se puede elegir uno de los 4 métodos siguientes, como se muestra en la figura: (1) AWS MCP (Knowledge Base), (2) MCP Lambda (Knowledge Base), (3) OpenSearch MCP y (4) MCP Lambda (OpenSearch). La infraestructura necesaria como OpenSearch, Lambda y SQS para utilizar RAG se puede implementar fácilmente con AWS CDK.
Si se capturan las páginas del documento y se analizan las imágenes, se puede obtener más información sobre tablas o imágenes dentro del documento. Para extraer y analizar las imágenes de las páginas del documento, se crea una canalización de procesamiento de datos en forma de eventos, como en Implementación de Agentic RAG. En este caso, el procesamiento de eventos se realiza con lambda-s3-event-manager y el procesamiento de documentos se implementa con lambda-document-manager.
Técnicas de Advanced RAG
Se explican las técnicas de advanced RAG para mejorar el rendimiento de RAG.
OCR
Después de convertir cada página del documento en una imagen, se analiza mediante multimodal. En este momento se utiliza contextual embedding para un análisis de imágenes acorde al contexto. Para más detalle, consulte lambda-document-manager. Para el contextual embedding se utiliza managed OpenSearch.
os_client se define de la siguiente manera y se utiliza al crear el índice de 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
)
Ahora se define el vectorstore y se utiliza al agregar o eliminar documentos.
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
)
Se extrae contextual_text para determinar qué significado tiene cada página dentro del documento completo y se utiliza.
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>')]
Para extraer texto de la imagen de la página, se ajusta el tamaño de la imagen y luego se registra en OpenSearch junto con el contextual text.
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: {width*height}")
isResized = False
while(width*height > 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)
La función que extrae texto de la imagen mediante multimodal es la siguiente.
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
Parent Child Chunking
Para aumentar la precisión de la búsqueda de documentos y utilizar suficiente contexto, se debe aplicar Parent Child Chunking. Para ello, se dividen parent y child con RecursiveCharacterTextSplitter de la siguiente manera.
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,
)
Primero se insertan los parent chunks en OpenSearch y se verifican sus id. Se agrega el id del parent chunk al meta del child chunk para buscar por child chunk, y el contexto real se obtiene consultando el texto del parent mediante el id del parent. Dado que el contextual text debe describir las características del parent chunk que se utilizará realmente, se agrega el contextual text obtenido del parent chunk al child chunk. Luego se registran también los child chunks en OpenSearch y se verifican sus id. Los id de parent/child se guardan en el meta del archivo y se utilizan al actualizar o eliminar documentos.
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
Uso de Knowledge Base
knowledge base, un servicio RAG totalmente administrado, ofrece una función de sync para almacenamientos como S3, por lo que se puede utilizar de forma conveniente. Sin embargo, al utilizar OCR o contextual embedding, se debe analizar directamente con Lambda u otras herramientas personalizadas e insertar los datos.
Uso de RAG con MCP
AWS MCP (Knowledge Base)
Se pueden consultar los documentos de Amazon Knowledge Base utilizando el MCP proporcionado por AWS, como Amazon Bedrock Knowledge Base Retrieval MCP Server. En este caso no se necesita infraestructura adicional para la consulta, por lo que se puede implementar fácilmente. La configuración de MCP en este caso es la siguiente.
{
"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 consulta knowledge base, busca la knowledge base con una etiqueta específica (por defecto mcp-tag) y luego ejecuta la consulta. Por lo tanto, al crear la knowledge base se debe configurar la etiqueta de la siguiente manera.
En Amazon Bedrock Knowledge Base Retrieval MCP Server, GetKnowledgeBases se invoca como resource, pero hay un problema de que no se puede referenciar en LangGraph. Por lo tanto, aquí se copia el código y se modifica como tool, como en mcp_server_knowledge_base.py. La configuración de MCP en este caso es la siguiente.
{
"mcpServers": {
"knowledge_base_custom": {
"command": "python",
"args": [
"application/mcp_server_knowledge_base.py"
],
"env": {
"KB_INCLUSION_TAG_KEY": "mcp-rag"
}
}
}
}
MCP Lambda (Knowledge Base)
Para consultar knowledge base mediante MCP, el uso de Lambda permite implementar un comportamiento de RAG adaptado a los objetivos del usuario. A continuación se explica cómo definir un servidor MCP personalizado con Lambda. El código relacionado con la consulta de knowledge base mediante Lambda se encuentra en lambda-knowledge-base. A continuación se define knowledge_base_search como 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 consulta los documentos de knowledge base invocando directamente Lambda, como en retrieve_knowledge_base definido en mcp_knowledge_base.py.
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']
Al implementar un servidor MCP con Lambda se necesita infraestructura adicional, pero se pueden realizar tareas personalizadas como excluir documentos de baja relevancia mediante grading, y dado que se realiza la consulta directamente sin listar las knowledge bases, se puede obtener una respuesta más rápida.
OpenSearch MCP
Con OpenSearch MCP se puede consultar OpenSearch directamente sin recursos adicionales. Sin embargo, a junio de 2025 solo se ofrece búsqueda de texto, por lo que hay limitaciones de rendimiento. Al utilizar OpenSearch MCP, se usa la siguiente configuración.
{
"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
}
}
}
}
MCP Lambda (OpenSearch)
Al implementar un RAG personalizado como OCR o contextual embedding, se debe insertar y consultar directamente en OpenSearch. Para la consulta, se puede definir un servidor MCP personalizado con Lambda o utilizar OpenSearch MCP. La consulta a OpenSearch con Lambda se puede implementar como en lambda-opensearch. Al implementarlo como servidor MCP personalizado, se implementa como en 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)
Aquí, como en mcp_opensearch.py, se puede obtener el documento de OpenSearch que Lambda recupera realizando una solicitud directa a lambda-opensearch. En Lambda no solo se consulta el documento en OpenSearch, sino que también se puede realizar grading basado en la relevancia del documento.
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
AgentCore Gateway puede proporcionar un servidor MCP con Lambda. Copie config.json.sample y cree el archivo config.json.
cd gateway/opensearch-retriever && cp config.json.sample config.json
En config.json, configure el nombre del proyecto, la URL de opensearch y sharing_url, que es el dominio de CloudFront, como se muestra a continuación. Esta información se puede verificar en los outputs generados durante la implementación con CDK.
{
"projectName": "mcp-rag",
"opensearch_url": "https://search-mcp-rag-mxtkul2z3qv5iiqprb7q3jx4wy.us-west-2.es.amazonaws.com",
"sharing_url": "https://d20lfnyi6fvd87.cloudfront.net"
}
Instale el rol para el gateway llamado "mcp-rag" en AgentCore de la siguiente manera.
python create_gateway_role.py
El resultado en este caso es el siguiente.
Instale AgentCore Gateway como se muestra a continuación e implemente "opensearch-retriever" como target. Si no existe una Lambda de RAG para consultar OpenSearch, instálela.
python create_gateway_tool.py
La instalación está completa, así que pruebe el funcionamiento de la siguiente manera.
test_mcp_remote.py
El resultado en este caso es el siguiente. Si el bearer token del secret expira, renueve el token e intente conectarse. En este momento, verifique las herramientas disponibles con list_tools y luego ejecute.
Instalación
Clone el repositorio.
git clone https://github.com/kyopark2014/mcp-rag/
Instale las librerías necesarias.
cd mcp-rag && pip install -r requirements.txt
Instale los secret de la infraestructura necesaria para CDK, como CloudFront, S3, OpenSearch, Knowledge base, tavily y weather. Si cdk bootstraping no está realizado, instálelo y luego ejecute.
cd cdk-mcp-rag/ && cdk deploy --all
Cuando la instalación esté completa, copie "CdkMcpRagStack.environmentformcprag" como se muestra a continuación y cree el archivo application/config.json.
config.json contiene la información necesaria para el funcionamiento del agente y no se comparte mediante git gracias a .gitignore. Un ejemplo del config.json generado es el siguiente.
{
"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"
}
Luego acceda a Secret Manager e ingrese las credenciales como se muestra a continuación.
Si streamlit no está instalado, consulte streamlit para instalarlo. Luego ejecute de la siguiente manera.
streamlit run application/app.py
Al ejecutar, se muestra la siguiente pantalla. Puede seleccionar un agente para ejecutarlo y verificar su funcionamiento.
Resultados de ejecución
Aquí se proporcionan los siguientes servidores MCP.
Si selecciona "AWS MCP (Knowledge Base)", se busca la knowledge base con la etiqueta mcp-rag mediante GetKnowledgeBases y luego se realiza la búsqueda con QueryKnowledgeBases, como se muestra a continuación.
Si selecciona "MCP Lambda (Knowledge Base)" y busca, puede obtener resultados de búsqueda de documentos utilizando knowledge_base_search como se muestra a continuación.
Si busca con "OpenSearch MCP", se realiza la consulta mediante OpenSearch MCP. Dado que es búsqueda de texto, es posible que no aparezcan resultados como se muestra a continuación.
Si busca "보일러 코드?" en "OpenSearch MCP", la búsqueda se realiza correctamente como se muestra a continuación.
Si busca con "MCP Lambda (OpenSearch)", puede obtener resultados de búsqueda en OpenSearch utilizando opensearch_search como se muestra a continuación.
Referencia
¿Por qué falla el RAG empresarial?…Cómo escalar RAG en entornos empresariales