Introducción
Tuve la oportunidad de unirme al equipo de Kiu, un proyecto innovador que permite conocer charlas, meetups y eventos de la comunidad AWS. Descubrí Kiu en el AWS Community Day Chile 2024 y me fascinó desde el primer momento.
En febrero, Hazel Saenz me invitó a participar en el equipo. Mi primera tarea fue integrar contenido de podcasts dentro de Kiu, creando un subagente virtual capaz de consultar y procesar esta información para responder preguntas de forma precisa y eficiente.
En este blog compartiré cómo diseñé y desplegué una infraestructura serverless robusta usando AWS CDK en Python, integrando Pinecone para búsquedas vectoriales y Amazon Bedrock para inteligencia artificial. Además, te contaré los desafíos técnicos que enfrenté y cómo los superé para entregar una solución segura, escalable y automatizada.
El problema y el objetivo
Kiu es un agente virtual que responde consultas utilizando inteligencia artificial apoyada en bases de conocimiento actualizadas. Nuestro objetivo era incorporar el mundo de los podcasts para enriquecer la base y mejorar la experiencia del usuario.
Para esto, necesitábamos:
- Extraer datos de podcasts desde una API externa y limpiarlos.
- Almacenar esta información estructurada en un bucket S3, que funciona como base de conocimiento.
- Usar Pinecone para búsquedas semánticas rápidas y eficientes.
- Que el agente Bedrock acceda a esta base para responder consultas.
- Automatizar la actualización periódica de los datos.
La solución técnica
Arquitectura Serverless con AWS CDK y Python
Me recomendaron implementar toda la infraestructura como código usando AWS CDK en Python, lo que facilitó mantener la solución versionada, segura y reproducible.
Los componentes clave fueron:
- Amazon S3: Bucket cifrado y versionado para almacenar archivos JSON con la información procesada de los podcasts.
- AWS Lambda: Función que consulta la API externa, procesa los datos limpiando y normalizando el texto, y actualiza el bucket S3.
-
Lambda Layers: Capas que contienen las librerías comunes (
requests,unidecode) para optimizar despliegues. - AWS Secrets Manager: Almacena de forma segura las credenciales de la API externa, como URL y cookies.
- AWS Scheduler: Programa la ejecución semanal automática de la función Lambda para mantener la base actualizada.
- Amazon Bedrock: Configura el agente virtual Kiu para consultar la base de conocimiento.
- Pinecone: Base de datos vectorial para búsquedas semánticas con un índice de dimensión 1024.
Flujo de datos
- El Scheduler activa semanalmente la función Lambda.
- Lambda obtiene las credenciales desde Secrets Manager.
- Lambda consulta y procesa los datos de la API externa.
- La información limpia y estructurada se almacena en S3.
- El agente Bedrock utiliza esta base para responder consultas.
- Pinecone facilita búsquedas semánticas rápidas y precisas.
Desafíos y aprendizajes
El proyecto implicó una curva de aprendizaje significativa y varios retos técnicos, que me ayudaron a crecer profesionalmente.
1. Aprender AWS CDK en Python
Aunque conocía CDK, tuve que aprender a usarlo correctamente en Python para alinearme con el equipo. Aquí un fragmento donde se define el bucket S3 y el secreto para las credenciales:
class PodcastKiuStack(Stack):
def __init__(self, scope: Construct, construct_id: str,
pinecone_api_key: str, pinecone_connection_url: str,
api_url: str, cookie: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
kb_bucket = s3.Bucket(
self, "PodcastKnowledgeBase",
versioned=True,
encryption=s3.BucketEncryption.S3_MANAGED,
removal_policy=RemovalPolicy.RETAIN
)
api_secret = secretsmanager.Secret(
self, "PodcastApiSecret",
secret_name="podcast-api-credentials",
description="Credenciales para acceder a la API de podcasts",
generate_secret_string=secretsmanager.SecretStringGenerator(
secret_string_template=json.dumps({
"API_URL": api_url,
"COOKIE": cookie
}),
generate_string_key="dummy"
)
)
2. Integrar Pinecone con Amazon Bedrock
Fue un reto configurar la knowledge base para usar Pinecone como base vectorial para búsquedas semánticas y configurar los roles necesarios para Bedrock:
bedrock_agent_role = iam.Role(
self, "BedrockAgentRole",
assumed_by=iam.ServicePrincipal("bedrock.amazonaws.com")
)
kb_bucket.grant_read(bedrock_agent_role)
knowledge_base = bedrock.CfnKnowledgeBase(
self, "MyKnowledgeBaseKiuPodcast",
name="KnowledgeBasePineconeKiuPodcast",
description="Knowledge base for the podcast content with pinecone",
role_arn=bedrock_role.role_arn,
knowledge_base_configuration=bedrock.CfnKnowledgeBase.KnowledgeBaseConfigurationProperty(
type="VECTOR",
vector_knowledge_base_configuration=bedrock.CfnKnowledgeBase.VectorKnowledgeBaseConfigurationProperty(
embedding_model_arn="arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2:0"
)
),
storage_configuration=bedrock.CfnKnowledgeBase.StorageConfigurationProperty(
type="PINECONE",
pinecone_configuration=bedrock.CfnKnowledgeBase.PineconeConfigurationProperty(
connection_string=pinecone_connection_url,
credentials_secret_arn=pinecone_url_secret.secret_arn,
field_mapping=bedrock.CfnKnowledgeBase.PineconeFieldMappingProperty(
text_field="text",
metadata_field="metafield"
),
)
)
)
3. Función Lambda para Procesamiento de Datos
Diseñé una función Lambda que consulta la API, procesa los datos limpiando texto con expresiones regulares y unidecode, y almacena la información en S3:
def lambda_handler(event, context):
secrets_client = boto3.client('secretsmanager')
secret_response = secrets_client.get_secret_value(SecretId=os.environ['API_SECRET_ARN'])
secret = json.loads(secret_response['SecretString'])
api_url = secret['API_URL']
cookie = secret['COOKIE']
response = requests.get(api_url, headers={'Cookie': cookie})
podcasts = response.json()
processed_data = []
for podcast in podcasts:
title = clean_text(podcast.get('title', ''))
description = clean_text(podcast.get('description', ''))
processed_data.append({
'id': podcast.get('id'),
'title': title,
'description': description,
'url': podcast.get('url'),
'published_date': podcast.get('published_date'),
'processed_timestamp': context.invoked_function_arn
})
s3_client = boto3.client('s3')
s3_client.put_object(
Bucket=os.environ['KB_BUCKET_NAME'],
Key='processed_podcasts.json',
Body=json.dumps(processed_data),
ContentType='application/json'
)
4. Seguridad: Roles IAM y Políticas granulares
Configuré roles con permisos mínimos necesarios para Lambda y Bedrock, garantizando la seguridad del sistema:
lambda_role = iam.Role(
self, "PodcastProcessorRole",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
)
lambda_role.add_managed_policy(
iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AWSLambdaBasicExecutionRole")
)
lambda_role.add_to_policy(
iam.PolicyStatement(
actions=["s3:PutObject", "s3:GetObject", "s3:ListBucket"],
resources=[kb_bucket.bucket_arn, f"{kb_bucket.bucket_arn}/*"]
)
)
lambda_role.add_to_policy(
iam.PolicyStatement(
actions=["secretsmanager:GetSecretValue"],
resources=[api_secret.secret_arn]
)
)
5. Pruebas Unitarias para asegurar calidad
Implementé pruebas unitarias que simulan la interacción con Secrets Manager, S3 y la API externa para validar el procesamiento de datos:
class TestLambdaFunction(unittest.TestCase):
@patch('lambda_function.boto3.client')
@patch('lambda_function.requests.get')
def test_successful_processing(self, mock_get, mock_boto3_client):
mock_context = MagicMock()
mock_context.invoked_function_arn = 'test-arn'
# mocks configurados...
result = lambda_function.lambda_handler({}, mock_context)
self.assertEqual(result['statusCode'], 200)
self.assertIn('Procesados 1 podcasts', result['body'])
mock_s3.put_object.assert_called_once()
6. Emojis en los títulos de los episodios
Durante las pruebas unitarias descubrí un problema curioso: si el título de un pódcast incluía un emoji, los metadatos no se almacenaban correctamente y fallaba el envío a S3. Este detalle, aunque pequeño, tenía un impacto directo en la integridad del flujo de datos. Con la ayuda de Chris Gonzalez, logramos identificar el origen del error y validamos el manejo de emojis usando la tabla oficial de Unicode. Esta corrección fue clave para asegurar que todos los títulos —emoji incluidos— se procesaran de forma segura y consistente.
Conclusión
Este proyecto ha sido para mí una gran curva de aprendizaje, no solo en términos técnicos, sino también en el trabajo colaborativo. Integrar tecnologías como AWS CDK en Python, Pinecone y Amazon Bedrock me llevó a salir de mi zona de confort y a descubrir nuevas formas de construir soluciones serverless escalables y seguras.
Además, trabajar en equipo ha sido fundamental. Cada compañero aporta desde distintos frentes, y esa sinergia nos está permitiendo avanzar con velocidad y calidad. Un agradecimiento muy especial a Hazel Saenz, quien ha sido una guía constante, enseñándome mucho, apoyándome en cada error y alentándome a seguir mejorando. Su paciencia y conocimiento han sido claves para mi crecimiento dentro del proyecto y para la entrega exitosa de esta funcionalidad.
Estoy emocionado por lo que viene y orgulloso de ser parte de este equipo que está construyendo el futuro de Kiu con tecnologías punteras.
Recursos
- AWS CDK Python
- Amazon Bedrock CDK
- Pinecone
- Repositorio GitHub: Próximamente disponible
Top comments (0)