En esta práctica vamos a analizar archivos de texto y logs de gran tamaño.
En lugar de enviar todo el archivo a Gemini, será Python quien procese el archivo localmente y entregue al modelo únicamente la información que necesita para realizar el análisis.
Esto permite reducir el volumen de información enviado al modelo y proporciona una arquitectura mucho más adecuada para trabajar con archivos grandes.
🎯 Objetivo
Queremos poder escribir algo como:
Analiza servidor.log y dime si observas
patrones relacionados con intentos de acceso
no autorizados.
Y que el agente sea capaz de:
- Identificar el archivo.
- Utilizar la herramienta
analizar_archivo(). - Leer el archivo localmente.
- Procesarlo línea por línea.
- Contar eventos.
- Detectar errores.
- Detectar advertencias.
- Detectar intentos fallidos.
- Extraer direcciones IP.
- Seleccionar ejemplos relevantes.
- Entregar únicamente el resumen a Gemini.
- Generar un informe de seguridad.
La arquitectura será:
USUARIO
│
▼
GEMINI
│
"analiza servidor.log"
│
▼
analizar_archivo()
│
▼
PYTHON
│
┌─────────────┼─────────────┐
▼ ▼ ▼
contar líneas extraer IPs eventos
│ │ │
└─────────────┼─────────────┘
▼
RESUMEN LOCAL
│
▼
GEMINI
│
▼
INFORME FINAL
⚠️ El problema de enviar el archivo completo
Una primera versión del agente podría hacer algo parecido a esto:
with open(ruta, "r", encoding="utf-8") as archivo:
contenido = archivo.read()
return {
"contenido": contenido
}
El problema aparece cuando el archivo es grande.
Podríamos tener:
10 KB
1 MB
10 MB
100 MB
1 GB
Si devolvemos todo el contenido:
ARCHIVO
↓
PYTHON
↓
TODO EL CONTENIDO
↓
GEMINI
estamos utilizando el modelo para recibir información que quizá ni siquiera necesita.
Además del consumo de tokens, aparecen otros problemas:
- límites de contexto;
- mayor latencia;
- mayor cantidad de datos procesados;
- mayor coste si el modelo se factura por tokens;
- repetición de información si mantenemos el contenido en el historial;
- posible exposición innecesaria de información sensible.
Por eso vamos a cambiar la arquitectura.
🧠 Nueva arquitectura
En lugar de:
100 MB
↓
Gemini
utilizaremos:
100 MB
↓
Python
↓
procesamiento local
↓
resumen
↓
Gemini
Por ejemplo:
servidor.log
│
│ 2.000.000 líneas
▼
PYTHON
│
├── total de líneas
├── errores
├── warnings
├── intentos fallidos
├── direcciones IP
└── ejemplos relevantes
│
▼
RESUMEN
│
▼
GEMINI
La idea es sencilla:
Python procesa los datos y Gemini interpreta los resultados.
🧪 1. Crear el proyecto
Creamos el archivo:
touch ciberagente_archivos.py
También podemos crear un archivo de prueba:
nano servidor.log
Introducimos algunos eventos:
2026-09-23 20:01:12 INFO Usuario admin inició sesión
2026-09-23 20:02:44 INFO Consulta realizada
2026-09-23 20:05:01 WARNING Intento de acceso fallido para usuario root
2026-09-23 20:05:04 WARNING Intento de acceso fallido para usuario root
2026-09-23 20:05:08 WARNING Intento de acceso fallido para usuario root
2026-09-23 20:06:12 ERROR Acceso bloqueado por exceso de intentos
2026-09-23 20:10:32 INFO Usuario admin cerró sesión
Tenemos así un pequeño log para probar nuestro agente.
🔧 2. Crear la función analizar_archivo()
La primera decisión importante es que no vamos a leer todo el archivo en memoria.
Utilizaremos:
for linea in archivo:
De esta manera podemos procesar el fichero línea por línea.
La función será:
def analizar_archivo(ruta: str) -> dict:
print(f"\n🔧 PYTHON → analizar_archivo('{ruta}')")
if not os.path.exists(ruta):
return {
"error": "El archivo no existe"
}
extension = os.path.splitext(ruta)[1].lower()
if extension not in [".log", ".txt"]:
return {
"error": f"Extensión no soportada: {extension}"
}
total_lineas = 0
errores = []
warnings = []
intentos_fallidos = []
ips = []
patron_ip = r"\b(?:\d{1,3}\.){3}\d{1,3}\b"
with open(
ruta,
"r",
encoding="utf-8",
errors="replace"
) as archivo:
for linea in archivo:
total_lineas += 1
texto = linea.strip()
mayusculas = texto.upper()
if "ERROR" in mayusculas:
errores.append(texto)
if "WARNING" in mayusculas:
warnings.append(texto)
if (
"FAILED" in mayusculas
or "DENIED" in mayusculas
or "LOGIN FAILED" in mayusculas
):
intentos_fallidos.append(texto)
direcciones = re.findall(
patron_ip,
texto
)
ips.extend(direcciones)
return {
"archivo": ruta,
"total_lineas": total_lineas,
"errores": len(errores),
"warnings": len(warnings),
"intentos_fallidos": len(intentos_fallidos),
"ips_detectadas": len(set(ips)),
"ips": list(set(ips))[:100],
"ejemplos_errores": errores[:20],
"ejemplos_warnings": warnings[:20],
"ejemplos_intentos_fallidos": intentos_fallidos[:50]
}
📊 3. ¿Qué hace Python con el archivo?
Supongamos que tenemos:
2.000.000 líneas
Python las recorre:
Línea 1
Línea 2
Línea 3
...
Línea 2.000.000
Pero no conserva todo el contenido.
En su lugar obtiene información como:
Total líneas: 2.000.000
Errores: 124
Warnings: 532
Intentos fallidos: 87
IPs diferentes: 23
Y conserva algunos ejemplos:
ERROR Authentication failure
WARNING Access denied
ERROR Login failed
Por tanto, Gemini recibe algo parecido a:
{
"total_lineas": 2000000,
"errores": 124,
"warnings": 532,
"intentos_fallidos": 87,
"ips_detectadas": 23
}
en lugar de recibir las dos millones de líneas.
🔍 4. Detección de direcciones IP
También podemos buscar direcciones IPv4.
Utilizamos una expresión regular:
patron_ip = r"\b(?:\d{1,3}\.){3}\d{1,3}\b"
Por ejemplo:
192.168.1.20
8.8.8.8
10.0.0.15
172.16.1.50
Se extraen mediante:
direcciones = re.findall(
patron_ip,
texto
)
Y después las acumulamos:
ips.extend(direcciones)
Finalmente eliminamos duplicados:
list(set(ips))
🤖 5. Conectar la herramienta con Gemini
Utilizamos el SDK:
from google import genai
from google.genai import types
import os
import re
Creamos el cliente:
client = genai.Client(
api_key=os.environ["GEMINI_API_KEY"]
)
Y utilizamos el modelo:
MODELO = "gemini-3.5-flash-lite"
🧰 6. Declarar la herramienta
Tenemos que explicar a Gemini que existe una función llamada:
analizar_archivo()
La declaración será:
herramienta_archivo = {
"name": "analizar_archivo",
"description": (
"Analiza localmente un archivo TXT o LOG "
"sin enviar su contenido completo al modelo. "
"Devuelve estadísticas, eventos relevantes, "
"direcciones IP y ejemplos."
),
"parameters": {
"type": "object",
"properties": {
"ruta": {
"type": "string",
"description": "Ruta local del archivo."
}
},
"required": ["ruta"]
}
}
La descripción es importante.
Estamos indicando que:
La herramienta procesa el archivo localmente.
⚙️ 7. Configurar la herramienta
tools = types.Tool(
function_declarations=[
herramienta_archivo
]
)
Ahora podemos crear la configuración:
config = types.GenerateContentConfig(
tools=[tools],
system_instruction="""
Eres un agente de ciberseguridad defensiva.
Puedes utilizar analizar_archivo()
para analizar archivos TXT y LOG.
La herramienta procesa los archivos localmente
y devuelve un resumen.
Utiliza la herramienta cuando el usuario
quiera analizar un archivo.
A partir de los resultados:
- identifica eventos relevantes
- busca patrones sospechosos
- analiza intentos fallidos
- interpreta errores
- analiza las IP detectadas
- explica las evidencias
No inventes información.
Diferencia entre datos observados,
interpretación e hipótesis.
No afirmes que existe un ataque si los datos
no permiten demostrarlo.
"""
)
🔄 8. Crear el historial
Pedimos al usuario una pregunta:
pregunta = input("\n🧑 Usuario: ")
Y creamos el historial:
historial = [
types.Content(
role="user",
parts=[
types.Part.from_text(
text=pregunta
)
]
)
]
🤖 9. Gemini decide utilizar la herramienta
Ahora ejecutamos:
respuesta = client.models.generate_content(
model=MODELO,
contents=historial,
config=config
)
Y comprobamos si Gemini ha solicitado una función:
llamadas = respuesta.function_calls
Si no necesita ninguna herramienta:
if not llamadas:
break
Pero si decide analizar el archivo podemos obtener:
🤖 GEMINI HA DECIDIDO:
Función: analizar_archivo
Argumentos:
{
"ruta": "servidor.log"
}
🖥️ 10. Python ejecuta la herramienta
Recuperamos el nombre:
nombre = llamada.name
Y los argumentos:
argumentos = dict(llamada.args)
Después ejecutamos:
resultado = analizar_archivo(
**argumentos
)
Por tanto:
GEMINI
│
│ decide
▼
analizar_archivo()
│
▼
PYTHON
│
▼
servidor.log
Gemini decide qué herramienta necesita.
Python ejecuta realmente la operación.
📦 11. Devolver el resultado
Después devolvemos el resultado a Gemini:
resultados.append(
types.Part.from_function_response(
name=nombre,
response=resultado
)
)
Y lo incorporamos al historial:
historial.append(
types.Content(
role="user",
parts=resultados
)
)
Gemini vuelve a recibir información.
Pero ahora recibe:
estadísticas
eventos relevantes
IPs
ejemplos
y no los millones de líneas originales.
🧩 12. Código completo
Este es el programa completo:
from google import genai
from google.genai import types
import os
import re
client = genai.Client(
api_key=os.environ["GEMINI_API_KEY"]
)
MODELO = "gemini-3.5-flash-lite"
def analizar_archivo(ruta: str) -> dict:
print(f"\n🔧 PYTHON → analizar_archivo('{ruta}')")
if not os.path.exists(ruta):
return {
"error": "El archivo no existe"
}
extension = os.path.splitext(ruta)[1].lower()
if extension not in [".log", ".txt"]:
return {
"error": f"Extensión no soportada: {extension}"
}
total_lineas = 0
errores = []
warnings = []
intentos_fallidos = []
ips = []
patron_ip = r"\b(?:\d{1,3}\.){3}\d{1,3}\b"
with open(
ruta,
"r",
encoding="utf-8",
errors="replace"
) as archivo:
for linea in archivo:
total_lineas += 1
texto = linea.strip()
mayusculas = texto.upper()
if "ERROR" in mayusculas:
errores.append(texto)
if "WARNING" in mayusculas:
warnings.append(texto)
if (
"FAILED" in mayusculas
or "DENIED" in mayusculas
or "LOGIN FAILED" in mayusculas
):
intentos_fallidos.append(texto)
direcciones = re.findall(
patron_ip,
texto
)
ips.extend(direcciones)
return {
"archivo": ruta,
"total_lineas": total_lineas,
"errores": len(errores),
"warnings": len(warnings),
"intentos_fallidos": len(intentos_fallidos),
"ips_detectadas": len(set(ips)),
"ips": list(set(ips))[:100],
"ejemplos_errores": errores[:20],
"ejemplos_warnings": warnings[:20],
"ejemplos_intentos_fallidos": intentos_fallidos[:50]
}
herramienta_archivo = {
"name": "analizar_archivo",
"description": (
"Analiza localmente un archivo TXT o LOG "
"sin enviar su contenido completo al modelo. "
"Devuelve estadísticas, eventos relevantes, "
"direcciones IP y ejemplos."
),
"parameters": {
"type": "object",
"properties": {
"ruta": {
"type": "string",
"description": "Ruta local del archivo."
}
},
"required": ["ruta"]
}
}
tools = types.Tool(
function_declarations=[
herramienta_archivo
]
)
config = types.GenerateContentConfig(
tools=[tools],
system_instruction="""
Eres un agente de ciberseguridad defensiva.
Puedes utilizar analizar_archivo()
para analizar archivos TXT y LOG.
La herramienta procesa los archivos localmente
y devuelve un resumen.
Utiliza la herramienta cuando el usuario
quiera analizar un archivo.
A partir de los resultados:
- identifica eventos relevantes
- busca patrones sospechosos
- analiza intentos fallidos
- interpreta errores
- analiza las IP detectadas
- explica las evidencias
No inventes información.
Diferencia entre datos observados,
interpretación e hipótesis.
No afirmes que existe un ataque si los datos
no permiten demostrarlo.
"""
)
pregunta = input("\n🧑 Usuario: ")
historial = [
types.Content(
role="user",
parts=[
types.Part.from_text(
text=pregunta
)
]
)
]
while True:
respuesta = client.models.generate_content(
model=MODELO,
contents=historial,
config=config
)
historial.append(
respuesta.candidates[0].content
)
llamadas = respuesta.function_calls
if not llamadas:
break
resultados = []
for llamada in llamadas:
nombre = llamada.name
argumentos = dict(llamada.args)
print("\n🤖 GEMINI HA DECIDIDO:")
print(f" Función: {nombre}")
print(f" Argumentos: {argumentos}")
if nombre == "analizar_archivo":
resultado = analizar_archivo(
**argumentos
)
else:
resultado = {
"error": "Herramienta desconocida"
}
print("\n📦 RESULTADO DEL ANÁLISIS LOCAL:")
print(resultado)
resultados.append(
types.Part.from_function_response(
name=nombre,
response=resultado
)
)
historial.append(
types.Content(
role="user",
parts=resultados
)
)
print("\n" + "=" * 60)
print("📋 INFORME DEL AGENTE")
print("=" * 60)
print(respuesta.text)
▶️ 13. Ejecutar el agente
Ejecutamos:
python3.12 ciberagente_archivos.py
Podemos escribir:
Analiza servidor.log y dime si observas
patrones relacionados con intentos de acceso
no autorizados.
El flujo será:
🧑 Usuario
│
▼
GEMINI
│
│ necesita analizar archivo
▼
analizar_archivo()
│
▼
PYTHON
│
├── lee línea por línea
├── cuenta eventos
├── busca errores
├── busca warnings
├── busca intentos fallidos
└── extrae IPs
│
▼
resumen reducido
│
▼
GEMINI
│
▼
informe de seguridad
🔐 14. ¿Qué puede detectar?
Con un log como:
2026-09-23 20:05:01 WARNING Intento de acceso fallido para usuario root
2026-09-23 20:05:04 WARNING Intento de acceso fallido para usuario root
2026-09-23 20:05:08 WARNING Intento de acceso fallido para usuario root
2026-09-23 20:06:12 ERROR Acceso bloqueado por exceso de intentos
Gemini puede recibir:
intentos_fallidos: 3
warnings: 3
errores: 1
y algunos ejemplos.
A partir de ellos puede generar una interpretación como:
Se observan tres intentos consecutivos de
autenticación fallida para el usuario root,
seguidos de un bloqueo por exceso de intentos.
Estos datos son compatibles con un patrón
de intentos repetidos de acceso.
El registro analizado no permite determinar
por sí solo el origen o la intención de la actividad.
La distinción es importante:
DATOS
↓
INTERPRETACIÓN
↓
HIPÓTESIS
El agente no debería convertir automáticamente una hipótesis en un hecho.
🚀 15. ¿Y si tenemos 10 millones de líneas?
Aquí aparece otro problema.
Imaginemos:
10.000.000 líneas
↓
Python
↓
50.000 eventos relevantes
Enviar esos 50.000 eventos tampoco sería necesariamente adecuado.
Entonces podemos utilizar chunking.
Dividimos los datos en bloques:
50.000 eventos
↓
┌────────────┐
│ CHUNK 1 │
│ 5.000 │
├────────────┤
│ CHUNK 2 │
│ 5.000 │
├────────────┤
│ CHUNK 3 │
│ 5.000 │
├────────────┤
│ ... │
├────────────┤
│ CHUNK 10 │
│ 5.000 │
└────────────┘
Cada bloque puede ser procesado individualmente.
CHUNK 1
↓
GEMINI
↓
resultado 1
CHUNK 2
↓
GEMINI
↓
resultado 2
...
CHUNK 10
↓
GEMINI
↓
resultado 10
Después podemos generar un informe global:
resultados parciales
│
▼
GEMINI
│
▼
informe global
🧠 16. Chunking y RAG
El concepto de dividir documentos en fragmentos es especialmente importante porque nos acerca a otra tecnología fundamental en el desarrollo de agentes:
RAG — Retrieval-Augmented Generation.
La arquitectura puede evolucionar hacia:
DOCUMENTOS
│
▼
PYTHON
│
▼
CHUNKING
│
▼
EMBEDDINGS
│
▼
BASE VECTORIAL
│
▼
BÚSQUEDA
│
▼
FRAGMENTOS RELEVANTES
│
▼
GEMINI
│
▼
RESPUESTA
En lugar de enviar todos los documentos al modelo, buscamos primero qué fragmentos son relevantes para la pregunta.
Por ejemplo:
Usuario:
¿Qué obligaciones del RGPD
están relacionadas con este incidente?
El sistema podría buscar primero los fragmentos pertinentes de la documentación normativa y después entregarlos a Gemini.
🛡️ 17. Hacia un agente de ciberseguridad
Nuestro proyecto puede seguir creciendo.
Podemos terminar teniendo:
🛡️ CIBERAGENTE
│
┌────────────────┼────────────────┐
▼ ▼ ▼
DNS/IP ARCHIVOS NORMATIVA
│ │ │
│ ┌─────┼─────┐ │
│ ▼ ▼ ▼ │
│ TXT LOG PDF │
│ │ │
└────────────────┼────────────────┘
▼
GEMINI
│
┌─────────┴─────────┐
▼ ▼
ANÁLISIS RIESGO
│ │
└─────────┬─────────┘
▼
INFORME FINAL
El agente podría utilizar diferentes herramientas dependiendo de la pregunta.
Por ejemplo:
analizar_ip()
dns_lookup()
analizar_archivo()
extraer_eventos()
extraer_ips()
consultar_normativa()
Gemini actuaría como el componente encargado de decidir qué herramienta necesita y en qué momento.
Python se ocuparía de ejecutar las operaciones.
🎓 Lo aprendido en esta práctica
Con este ejercicio hemos incorporado un concepto fundamental:
No todo debe enviarse al modelo.
Un agente bien diseñado puede dividir el trabajo:
PYTHON
──────
lectura
filtrado
conteo
extracción
procesamiento
↓
GEMINI
──────
interpretación
razonamiento
relación entre datos
generación del informe
Esto nos permite construir sistemas más eficientes y escalables.
La primera versión del agente era:
ARCHIVO
↓
PYTHON
↓
TODO EL CONTENIDO
↓
GEMINI
La versión mejorada es:
ARCHIVO
↓
PYTHON
↓
PROCESAMIENTO LOCAL
↓
DATOS RELEVANTES
↓
GEMINI
↓
INFORME
Y la siguiente evolución será:
ARCHIVO
↓
PYTHON
↓
CHUNKING
↓
EMBEDDINGS
↓
BASE VECTORIAL
↓
RECUPERACIÓN
↓
GEMINI
↓
RESPUESTA