Curso Completo: Python para IA con FastAPI y LangChain | Blog

Curso Completo: Python para IA con FastAPI y LangChain

Curso

April 2, 2026

3 hours

Módulo 1 — Fundamentos de Python

1.1 ¿Qué es Python y por qué usarlo para IA?

Python es un lenguaje de programación creado en 1991. Es uno de los más populares del mundo, especialmente en el campo de la Inteligencia Artificial, por tres razones clave:

Sintaxis clara: El código se parece al inglés, lo que lo hace muy legible.
Ecosistema enorme: Existen miles de librerías gratuitas para IA, ciencia de datos, APIs, etc.
Comunidad: Millones de desarrolladores lo usan, así que siempre hay ayuda disponible.

Python es de tipado dinámico — igual que JavaScript

Python no te obliga a declarar el tipo de una variable. El intérprete lo deduce solo en tiempo de ejecución:

Sin embargo, desde Python 3.5+ existe el sistema de type hints: anotaciones opcionales que no impone el intérprete pero sí usan herramientas como FastAPI y Pydantic para validar datos automáticamente:

Regla práctica: En scripts personales puedes ignorar los type hints. En APIs y proyectos de equipo, úsalos siempre — te ahorrarán horas de depuración.

1.2 Instalación de Python

En Windows

Ve a https://www.python.org/downloads/
Descarga la versión más reciente (Python 3.11 o superior).
Ejecuta el instalador y marca la casilla "Add Python to PATH" antes de hacer clic en Install.
Abre el símbolo del sistema (busca "cmd" en el menú inicio) y escribe:

Deberías ver algo como Python 3.11.5. Si es así, ¡está instalado!

En macOS

macOS moderno ya incluye Python, pero es mejor instalar una versión actualizada:

En Linux (Ubuntu/Debian)

1.3 Tu primer programa

Abre el bloc de notas (o cualquier editor de texto), escribe lo siguiente y guárdalo como hola.py:


python
print("¡Hola, mundo!")


bash
python hola.py

¡Hola, mundo!


python
# Texto (se llama "string" o "str")
nombre = "Ana"
apellido = 'García'  # También puedes usar comillas simples

# Números enteros (int)
edad = 28
año = 2024

# Números decimales (float)
altura = 1.72
precio = 9.99

# Verdadero o Falso (bool)
es_estudiante = True
tiene_trabajo = False

# Mostrar variables
print(nombre)        # Ana
print(edad)          # 28
print(es_estudiante) # True


python
print(type(nombre))    # <class 'str'>
print(type(edad))      # <class 'int'>
print(type(altura))    # <class 'float'>
print(type(es_estudiante))  # <class 'bool'>


python
# Matemáticas
suma = 10 + 5         # 15
resta = 10 - 3        # 7
multiplicacion = 4 * 6 # 24
division = 15 / 4     # 3.75 (siempre devuelve float)
division_entera = 15 // 4  # 3 (descarta los decimales)
modulo = 15 % 4       # 3 (el resto de la división)
potencia = 2 ** 8     # 256

# Con texto (concatenación)
nombre = "Carlos"
saludo = "Hola, " + nombre + "!"
print(saludo)  # Hola, Carlos!

# Forma moderna (f-strings) — mucho más cómoda
saludo = f"Hola, {nombre}! Tienes {edad} años."
print(saludo)  # Hola, Carlos! Tienes 28 años.


python
precio = 1234567.891
pi = 3.14159265

# Formatear números
print(f"{precio:,.2f}")     # 1,234,567.89  — separador de miles y 2 decimales
print(f"{pi:.4f}")          # 3.1416        — 4 decimales
print(f"{0.75:.0%}")        # 75%           — porcentaje

# Alinear texto (útil para tablas en consola)
print(f"{'Producto':<15} {'Precio':>10}")   # alineado izquierda / derecha
print(f"{'Portátil':<15} {999.99:>10.2f}")

# Modo debug (Python 3.8+): muestra el nombre y el valor
x = 42
print(f"{x=}")          # x=42  — muy útil para depurar

nombre = "Ana"
edad = 28
print(f"{nombre=}, {edad=}")  # nombre='Ana', edad=28

# Expresiones directamente dentro de las llaves
numeros = [3, 1, 8, 2]
print(f"Máximo: {max(numeros)}, Suma: {sum(numeros)}")
print(f"{'par' if x % 2 == 0 else 'impar'}")  # par


python
edad = 20

if edad >= 18:
    print("Eres mayor de edad")
elif edad >= 13:
    print("Eres adolescente")
else:
    print("Eres menor de edad")


python
# Ejemplo con varias condiciones
temperatura = 25

if temperatura > 30:
    print("Hace mucho calor")
elif temperatura > 20:
    print("Temperatura agradable")
elif temperatura > 10:
    print("Algo fresco")
else:
    print("Hace frío")


python
# Contar del 0 al 4
for i in range(5):
    print(f"Iteración número {i}")

# Contar del 1 al 10
for numero in range(1, 11):
    print(numero)

# Contar de 2 en 2
for par in range(0, 20, 2):
    print(par)  # 0, 2, 4, 6, ..., 18


python
contador = 0

while contador < 5:
    print(f"El contador vale: {contador}")
    contador = contador + 1  # Equivalente: contador += 1

print("El bucle ha terminado")


python
# Crear una lista
frutas = ["manzana", "banana", "naranja", "uva"]

# Acceder a elementos (empieza en 0)
print(frutas[0])   # manzana
print(frutas[1])   # banana
print(frutas[-1])  # uva (el último)

# Modificar un elemento
frutas[1] = "mango"
print(frutas)  # ['manzana', 'mango', 'naranja', 'uva']

# Añadir elementos
frutas.append("kiwi")       # Al final
frutas.insert(1, "pera")    # En una posición específica

# Eliminar elementos
frutas.remove("naranja")    # Por valor
frutas.pop()                # El último
frutas.pop(0)               # Por posición

# Longitud de la lista
print(len(frutas))

# Recorrer una lista con for
for fruta in frutas:
    print(f"- {fruta}")


python
# Slicing — extraer porciones de una lista
numeros = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

print(numeros[2:5])    # [2, 3, 4]       — del índice 2 al 4
print(numeros[:3])     # [0, 1, 2]       — los primeros 3
print(numeros[-3:])    # [7, 8, 9]       — los últimos 3
print(numeros[::2])    # [0, 2, 4, 6, 8] — de 2 en 2
print(numeros[::-1])   # [9,8,7,...,0]   — invertir la lista ← truco clásico

# Desempaquetar listas (unpacking)
primero, *resto = numeros
print(primero)  # 0
print(resto)    # [1, 2, 3, 4, 5, 6, 7, 8, 9]

a, b, *medio, ultimo = [1, 2, 3, 4, 5]
print(a, b, medio, ultimo)  # 1 2 [3, 4] 5

# Combinar listas
lista_a = [1, 2, 3]
lista_b = [4, 5, 6]
combinada = lista_a + lista_b       # [1,2,3,4,5,6]
combinada = [*lista_a, *lista_b]    # Igual, con spread operator

# Ordenar
nums = [3, 1, 8, 2, 9]
nums.sort()                          # Modifica la lista original
ordenada = sorted(nums)              # Devuelve una nueva lista
ordenada_desc = sorted(nums, reverse=True)

# Comprobar si un elemento existe
if "manzana" in frutas:
    print("¡Hay manzanas!")

# enumerate — iterar con índice sin usar range(len(...))
for i, fruta in enumerate(frutas, start=1):
    print(f"{i}. {fruta}")

# zip — combinar dos listas en paralelo
nombres = ["Ana", "Luis", "Sara"]
edades  = [25, 30, 22]
for nombre, edad in zip(nombres, edades):
    print(f"{nombre} tiene {edad} años")


python
# Crear un diccionario
persona = {
    "nombre": "Lucía",
    "edad": 30,
    "ciudad": "Madrid",
    "es_programadora": True
}

# Acceder a valores
print(persona["nombre"])   # Lucía
print(persona["edad"])     # 30

# Forma segura (no falla si la clave no existe)
print(persona.get("telefono", "Sin teléfono"))  # Sin teléfono

# Añadir o modificar
persona["profesion"] = "Desarrolladora"
persona["edad"] = 31

# Eliminar
del persona["ciudad"]

# Recorrer un diccionario
for clave, valor in persona.items():
    print(f"{clave}: {valor}")

# Diccionarios anidados (muy comunes en respuestas de APIs)
usuario = {
    "id": 1,
    "datos": {
        "nombre": "Pedro",
        "correo": "pedro@ejemplo.com"
    },
    "roles": ["admin", "usuario"]
}

print(usuario["datos"]["nombre"])  # Pedro
print(usuario["roles"][0])         # admin


python
# Función básica
def saludar():
    print("¡Hola!")

saludar()  # Llamas a la función así

# Función con parámetros
def saludar_a(nombre):
    print(f"¡Hola, {nombre}!")

saludar_a("María")

# Función que devuelve un valor
def sumar(a, b):
    resultado = a + b
    return resultado

total = sumar(5, 3)
print(total)  # 8

# Función con valores por defecto
def presentar(nombre, pais="España"):
    return f"Me llamo {nombre} y soy de {pais}"

print(presentar("Carlos"))          # Me llamo Carlos y soy de España
print(presentar("Ana", "México"))   # Me llamo Ana y soy de México

# Función con múltiples valores de retorno
def minmax(lista):
    return min(lista), max(lista)

minimo, maximo = minmax([3, 1, 8, 2, 9, 4])
print(f"Mínimo: {minimo}, Máximo: {maximo}")  # Mínimo: 1, Máximo: 9


python
# Sin manejo de errores (el programa se rompe)
numero = int("abc")  # ValueError!

# Con manejo de errores
try:
    numero = int("abc")
    print(f"El número es: {numero}")
except ValueError:
    print("Error: eso no es un número válido")

# Múltiples tipos de error
def dividir(a, b):
    try:
        resultado = a / b
        return resultado
    except ZeroDivisionError:
        print("Error: no se puede dividir entre cero")
        return None
    except TypeError:
        print("Error: los valores deben ser números")
        return None
    finally:
        # Este bloque SIEMPRE se ejecuta, haya error o no
        print("Operación de división finalizada")

print(dividir(10, 2))   # 5.0
print(dividir(10, 0))   # Error: no se puede dividir entre cero
print(dividir(10, "a")) # Error: los valores deben ser números


python
# Definir una clase
class Persona:
    
    # El método __init__ se llama al crear un objeto
    def __init__(self, nombre, edad):
        self.nombre = nombre  # self.xxx son atributos del objeto
        self.edad = edad
    
    # Métodos (funciones dentro de la clase)
    def saludar(self):
        return f"Hola, me llamo {self.nombre} y tengo {self.edad} años"
    
    def cumpleaños(self):
        self.edad += 1
        return f"¡Feliz cumpleaños {self.nombre}! Ahora tienes {self.edad} años"
    
    # Representación en texto del objeto
    def __str__(self):
        return f"Persona({self.nombre}, {self.edad})"


# Crear objetos (instancias de la clase)
persona1 = Persona("Laura", 25)
persona2 = Persona("Marcos", 32)

print(persona1.saludar())
print(persona2.cumpleaños())
print(persona1)  # Usa __str__


# Herencia — una clase que extiende otra
class Programador(Persona):
    
    def __init__(self, nombre, edad, lenguaje_favorito):
        super().__init__(nombre, edad)  # Llama al __init__ de Persona
        self.lenguaje_favorito = lenguaje_favorito
    
    def programar(self):
        return f"{self.nombre} está programando en {self.lenguaje_favorito}"


dev = Programador("Sara", 28, "Python")
print(dev.saludar())    # Hereda el método de Persona
print(dev.programar())  # Método propio de Programador


python
# archivo: matematicas.py
def sumar(a, b):
    return a + b

def restar(a, b):
    return a - b

PI = 3.14159


python
# archivo: main.py
import matematicas

resultado = matematicas.sumar(5, 3)
print(resultado)  # 8
print(matematicas.PI)  # 3.14159

# También puedes importar solo lo que necesitas
from matematicas import sumar, PI

resultado = sumar(10, 5)  # Sin necesidad de poner "matematicas."


python
# Escribir en un archivo
with open("datos.txt", "w", encoding="utf-8") as archivo:
    archivo.write("Primera línea\n")
    archivo.write("Segunda línea\n")

# Leer un archivo completo
with open("datos.txt", "r", encoding="utf-8") as archivo:
    contenido = archivo.read()
    print(contenido)

# Leer línea por línea
with open("datos.txt", "r", encoding="utf-8") as archivo:
    for linea in archivo:
        print(linea.strip())  # strip() elimina espacios y saltos de línea

# Trabajar con JSON (muy común en APIs e IA)
import json

# Guardar un diccionario como JSON
datos = {
    "nombre": "Pepe",
    "edad": 40,
    "habilidades": ["Python", "IA", "FastAPI"]
}

with open("usuario.json", "w", encoding="utf-8") as f:
    json.dump(datos, f, ensure_ascii=False, indent=2)

# Leer JSON
with open("usuario.json", "r", encoding="utf-8") as f:
    usuario = json.load(f)
    print(usuario["nombre"])  # Pepe


python
# Forma tradicional
cuadrados = []
for i in range(10):
    cuadrados.append(i ** 2)

# Con list comprehension (equivalente, pero más limpio)
cuadrados = [i ** 2 for i in range(10)]
print(cuadrados)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

# Con condición
pares = [i for i in range(20) if i % 2 == 0]
print(pares)  # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

# Transformar listas
nombres = ["ana", "CARLOS", "María"]
normalizados = [n.lower().capitalize() for n in nombres]
print(normalizados)  # ['Ana', 'Carlos', 'María']

# Dict comprehension
precios = {"manzana": 1.2, "banana": 0.5, "naranja": 0.8}
precios_con_iva = {fruta: round(precio * 1.21, 2) for fruta, precio in precios.items()}
print(precios_con_iva)


python
# *args: acepta un número variable de argumentos
def sumar_todo(*numeros):
    return sum(numeros)

print(sumar_todo(1, 2, 3))        # 6
print(sumar_todo(1, 2, 3, 4, 5))  # 15

# **kwargs: acepta argumentos con nombre variable
def crear_perfil(**datos):
    for clave, valor in datos.items():
        print(f"{clave}: {valor}")

crear_perfil(nombre="Julia", edad=25, ciudad="Barcelona")

# Lambda: función pequeña de una sola línea
doblar = lambda x: x * 2
print(doblar(5))  # 10

# Muy útil con sorted, map, filter
numeros = [3, 1, 8, 2, 9, 4]
ordenados = sorted(numeros, key=lambda x: -x)  # Orden descendente
print(ordenados)  # [9, 8, 4, 3, 2, 1]


python
import time
import functools

# Decorador básico: medir el tiempo de ejecución
def medir_tiempo(funcion):
    @functools.wraps(funcion)  # Preserva el nombre y docstring original
    def wrapper(*args, **kwargs):
        inicio = time.time()
        resultado = funcion(*args, **kwargs)
        fin = time.time()
        print(f"⏱ {funcion.__name__} tardó {fin - inicio:.4f}s")
        return resultado
    return wrapper

@medir_tiempo
def tarea_pesada():
    time.sleep(1)
    return "hecho"

tarea_pesada()  # ⏱ tarea_pesada tardó 1.0012s


# Decorador con parámetros: reintentar N veces si falla
def reintentar(veces=3, espera=1.0):
    def decorador(funcion):
        @functools.wraps(funcion)
        def wrapper(*args, **kwargs):
            for intento in range(1, veces + 1):
                try:
                    return funcion(*args, **kwargs)
                except Exception as e:
                    print(f"Intento {intento}/{veces} falló: {e}")
                    if intento < veces:
                        time.sleep(espera)
            raise RuntimeError(f"Falló tras {veces} intentos")
        return wrapper
    return decorador

@reintentar(veces=3, espera=0.5)
def llamar_api_externa():
    import random
    if random.random() < 0.7:
        raise ConnectionError("Timeout")
    return {"datos": "ok"}

# Muy útil cuando llamas a APIs de IA que a veces fallan por rate limits


python
# Sin generador — carga TODO en memoria
def primeros_n_cuadrados_lista(n):
    return [i ** 2 for i in range(n)]

# Con generador — produce uno cada vez
def primeros_n_cuadrados_gen(n):
    for i in range(n):
        yield i ** 2  # yield en lugar de return

# Usar el generador
gen = primeros_n_cuadrados_gen(1_000_000)
print(next(gen))  # 0
print(next(gen))  # 1
print(next(gen))  # 4

# O iterar con for (consume el generador)
for cuadrado in primeros_n_cuadrados_gen(5):
    print(cuadrado)

# Caso de uso real: leer un fichero enorme línea a línea
def leer_lineas(ruta_archivo):
    with open(ruta_archivo, "r", encoding="utf-8") as f:
        for linea in f:
            yield linea.strip()

# Esto NO carga el archivo entero en RAM — perfecto para logs de gigabytes
for linea in leer_lineas("log_enorme.txt"):
    if "ERROR" in linea:
        print(linea)

# Generator expression (como list comprehension pero perezoso)
cuadrados = (i ** 2 for i in range(10))  # Paréntesis, no corchetes
print(sum(cuadrados))  # 285 — calcula sobre la marcha


python
from dataclasses import dataclass, field
from typing import List

# Sin dataclass — mucho código repetitivo
class UsuarioNormal:
    def __init__(self, nombre, edad, tags):
        self.nombre = nombre
        self.edad = edad
        self.tags = tags
    def __repr__(self):
        return f"Usuario({self.nombre}, {self.edad})"
    def __eq__(self, otro):
        return self.nombre == otro.nombre and self.edad == otro.edad

# Con dataclass — todo automático
@dataclass
class Usuario:
    nombre: str
    edad: int
    tags: List[str] = field(default_factory=list)  # Lista vacía por defecto
    activo: bool = True

    def es_adulto(self) -> bool:
        return self.edad >= 18

# Python genera __init__, __repr__, __eq__ automáticamente
u1 = Usuario("Ana", 25, ["python", "ia"])
u2 = Usuario("Ana", 25, ["python", "ia"])

print(u1)           # Usuario(nombre='Ana', edad=25, tags=['python', 'ia'], activo=True)
print(u1 == u2)     # True  — compara por valor, no por referencia
print(u1.es_adulto())  # True

# Dataclass frozen — inmutable (como una tupla con nombres)
@dataclass(frozen=True)
class Coordenada:
    lat: float
    lon: float

madrid = Coordenada(40.4168, -3.7038)
# madrid.lat = 0  # Error — no se puede modificar


bash
# Crear un entorno virtual (dentro de la carpeta de tu proyecto)
python -m venv .venv

# Activarlo en Windows
.venv\Scripts\activate

# Activarlo en macOS/Linux
source .venv/bin/activate

# Verás que el prompt cambia, algo así:
# (.venv) usuario@ordenador:~/proyecto$

# Instalar paquetes dentro del entorno
pip install fastapi langchain

# Ver qué tienes instalado
pip list

# Guardar las dependencias en un archivo
pip freeze > requirements.txt

# En otro ordenador, instalar todo de golpe
pip install -r requirements.txt

# Desactivar el entorno virtual
deactivate


bash
# Instalar un paquete
pip install nombre_paquete

# Instalar una versión específica
pip install fastapi==0.110.0

# Actualizar un paquete
pip install --upgrade nombre_paquete

# Desinstalar
pip uninstall nombre_paquete


bash
# Instalar uv
pip install uv

# Crear proyecto con entorno virtual
uv init mi-proyecto
cd mi-proyecto

# Añadir dependencias (actualiza pyproject.toml automáticamente)
uv add fastapi langchain-openai

# Añadir dependencias solo para desarrollo
uv add --dev pytest ruff mypy

# Instalar todo desde pyproject.toml
uv sync

# Ejecutar scripts dentro del entorno sin activarlo manualmente
uv run python main.py
uv run uvicorn main:app --reload


toml
# pyproject.toml
[project]
name = "api-ia"
version = "1.0.0"
requires-python = ">=3.11"
dependencies = [
    "fastapi>=0.110.0",
    "langchain>=0.3.0",
    "langchain-openai>=0.2.0",
    "uvicorn[standard]>=0.27.0",
    "pydantic-settings>=2.0.0",
    "python-dotenv>=1.0.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0.0",
    "pytest-asyncio>=0.23.0",
    "httpx>=0.27.0",
    "ruff>=0.3.0",
    "mypy>=1.9.0",
]


bash
# Instalar python-dotenv
pip install python-dotenv

# .env
OPENAI_API_KEY=sk-tu-clave-aqui
DATABASE_URL=postgresql://usuario:contraseña@localhost/midb
SECRET_KEY=mi-clave-secreta
DEBUG=True

# .gitignore
.env
.venv/
__pycache__/
*.pyc


python
from dotenv import load_dotenv
import os

load_dotenv()  # Carga el archivo .env

api_key = os.getenv("OPENAI_API_KEY")
debug = os.getenv("DEBUG", "False")  # Valor por defecto

print(f"API Key cargada: {'Sí' if api_key else 'No'}")


bash
pip install fastapi uvicorn


python
from fastapi import FastAPI

# Crear la aplicación
app = FastAPI(
    title="Mi primera API",
    description="Una API de ejemplo para aprender FastAPI",
    version="1.0.0"
)

# Definir una ruta (endpoint)
@app.get("/")
def inicio():
    return {"mensaje": "¡Hola desde FastAPI!"}

@app.get("/saludo/{nombre}")
def saludar(nombre: str):
    return {"saludo": f"¡Hola, {nombre}!"}


bash
uvicorn main:app --reload

Método	Uso	Ejemplo
GET	Obtener datos	Leer un usuario
POST	Crear datos	Crear un usuario
PUT	Actualizar datos (completo)	Actualizar todos los datos de un usuario
PATCH	Actualizar datos (parcial)	Solo cambiar el email
DELETE	Eliminar datos	Borrar un usuario


python
from fastapi import FastAPI
from typing import Optional

app = FastAPI()

# Base de datos simulada en memoria
usuarios = {}

# GET — Obtener todos los usuarios
@app.get("/usuarios")
def obtener_usuarios():
    return {"usuarios": list(usuarios.values())}

# GET — Obtener un usuario por ID
@app.get("/usuarios/{id}")
def obtener_usuario(id: int):
    if id not in usuarios:
        return {"error": "Usuario no encontrado"}
    return usuarios[id]

# POST — Crear usuario (el cuerpo se envía en el body de la petición)
@app.post("/usuarios/{id}")
def crear_usuario(id: int, nombre: str, email: str):
    usuarios[id] = {"id": id, "nombre": nombre, "email": email}
    return {"mensaje": "Usuario creado", "usuario": usuarios[id]}

# DELETE — Eliminar un usuario
@app.delete("/usuarios/{id}")
def eliminar_usuario(id: int):
    if id in usuarios:
        del usuarios[id]
        return {"mensaje": f"Usuario {id} eliminado"}
    return {"error": "Usuario no encontrado"}


bash
uv add "pydantic[email]"
 o 
 pip install email-validator


python
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime

app = FastAPI()

# Modelo de datos
class Usuario(BaseModel):
    nombre: str
    email: EmailStr
    edad: int = Field(ge=0, le=150, description="Edad entre 0 y 150")
    activo: bool = True
    bio: Optional[str] = None  # Campo opcional

class UsuarioRespuesta(BaseModel):
    id: int
    nombre: str
    email: EmailStr
    creado_en: datetime

# Base de datos simulada
db_usuarios: dict[int, dict] = {}
contador_id = 1

@app.post("/usuarios", response_model=UsuarioRespuesta)
def crear_usuario(usuario: Usuario):
    global contador_id
    nuevo_usuario = {
        "id": contador_id,
        "nombre": usuario.nombre,
        "email": usuario.email,
        "creado_en": datetime.now()
    }
    db_usuarios[contador_id] = nuevo_usuario
    contador_id += 1
    return nuevo_usuario


python
from fastapi import FastAPI
from typing import Optional

app = FastAPI()

productos = [
    {"id": 1, "nombre": "Portátil", "categoria": "electronica", "precio": 999},
    {"id": 2, "nombre": "Ratón", "categoria": "electronica", "precio": 25},
    {"id": 3, "nombre": "Silla", "categoria": "muebles", "precio": 150},
    {"id": 4, "nombre": "Mesa", "categoria": "muebles", "precio": 200},
]

@app.get("/productos")
def buscar_productos(
    categoria: Optional[str] = None,
    precio_max: Optional[float] = None,
    pagina: int = 1,
    por_pagina: int = 10
):
    resultado = productos
    
    if categoria:
        resultado = [p for p in resultado if p["categoria"] == categoria]
    
    if precio_max:
        resultado = [p for p in resultado if p["precio"] <= precio_max]
    
    # Paginación simple
    inicio = (pagina - 1) * por_pagina
    fin = inicio + por_pagina
    
    return {
        "total": len(resultado),
        "pagina": pagina,
        "productos": resultado[inicio:fin]
    }


python
from fastapi import FastAPI, HTTPException, status

app = FastAPI()

usuarios = {1: {"nombre": "Ana", "email": "ana@email.com"}}

@app.get("/usuarios/{id}")
def obtener_usuario(id: int):
    if id not in usuarios:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Usuario con id {id} no encontrado"
        )
    return usuarios[id]

@app.post("/usuarios")
def crear_usuario(nombre: str, email: str):
    # Validación personalizada
    if "@" not in email:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="El email no tiene un formato válido"
        )
    
    nuevo_id = max(usuarios.keys()) + 1
    usuarios[nuevo_id] = {"nombre": nombre, "email": email}
    return {"mensaje": "Creado", "id": nuevo_id}


python
import asyncio
import time
from fastapi import FastAPI

app = FastAPI()

# Función SÍNCRONA — bloquea el servidor mientras espera
@app.get("/sincrono")
def endpoint_sincrono():
    time.sleep(3)  # Simula una espera (p.ej., llamar a una IA)
    return {"mensaje": "Respuesta síncrona"}

# Función ASÍNCRONA — no bloquea el servidor
@app.get("/asincrono")
async def endpoint_asincrono():
    await asyncio.sleep(3)  # Espera sin bloquear
    return {"mensaje": "Respuesta asíncrona"}


python
from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Optional

app = FastAPI()

# Dependencia: verificar que hay un token en la cabecera
async def verificar_token(x_token: str = Header(...)):
    tokens_validos = ["mi-token-secreto", "otro-token"]
    if x_token not in tokens_validos:
        raise HTTPException(status_code=401, detail="Token inválido")
    return x_token

# Esta ruta requiere que la dependencia pase sin errores
@app.get("/datos-privados")
async def datos_privados(token: str = Depends(verificar_token)):
    return {"datos": "Información confidencial", "token_usado": token}

# Dependencia para paginación reutilizable
async def parametros_paginacion(pagina: int = 1, tamaño: int = 10):
    return {"offset": (pagina - 1) * tamaño, "limit": tamaño}

@app.get("/articulos")
async def listar_articulos(paginacion: dict = Depends(parametros_paginacion)):
    # Usamos los parámetros calculados por la dependencia
    return {"paginacion": paginacion, "articulos": []}


python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import time

app = FastAPI()

# Configurar CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://miweb.com"],
    allow_credentials=True,
    allow_methods=["*"],  # GET, POST, PUT, DELETE, etc.
    allow_headers=["*"],
)

# Middleware personalizado: registrar el tiempo de cada petición
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class TiempoMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        inicio = time.time()
        respuesta = await call_next(request)
        duracion = time.time() - inicio
        respuesta.headers["X-Tiempo-Proceso"] = str(duracion)
        return respuesta

app.add_middleware(TiempoMiddleware)


python
from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

def procesar_en_segundo_plano(texto: str):
    """Esta función se ejecuta después de responder al usuario"""
    time.sleep(5)  # Simula procesamiento pesado
    print(f"Procesado: {texto}")

@app.post("/analizar")
async def analizar_texto(texto: str, tareas: BackgroundTasks):
    # Programa la tarea para ejecutarse después
    tareas.add_task(procesar_en_segundo_plano, texto)
    
    # El usuario recibe respuesta INMEDIATAMENTE
    return {"mensaje": "Texto recibido, procesando en segundo plano..."}


python
from fastapi import FastAPI
from contextlib import asynccontextmanager
from langchain_openai import ChatOpenAI
import logging

logger = logging.getLogger(__name__)

# Estado global de la aplicación
class AppState:
    llm: ChatOpenAI = None
    vector_store = None

state = AppState()

@asynccontextmanager
async def lifespan(app: FastAPI):
    # ── STARTUP ── código que se ejecuta al iniciar el servidor
    logger.info("Iniciando servidor...")
    state.llm = ChatOpenAI(model="gpt-4o-mini")
    logger.info("Modelo de IA cargado ✓")
    
    yield  # El servidor está corriendo aquí
    
    # ── SHUTDOWN ── código que se ejecuta al apagar el servidor
    logger.info("Apagando servidor, liberando recursos...")

app = FastAPI(lifespan=lifespan)

@app.post("/chat")
async def chat(mensaje: str):
    # state.llm ya está inicializado y listo
    respuesta = await state.llm.ainvoke(mensaje)
    return {"respuesta": respuesta.content}


python
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
import logging

logger = logging.getLogger(__name__)
app = FastAPI()

# Handler para errores de validación de Pydantic
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errores = []
    for error in exc.errors():
        errores.append({
            "campo": " → ".join(str(x) for x in error["loc"]),
            "mensaje": error["msg"],
            "valor": error.get("input")
        })
    return JSONResponse(
        status_code=422,
        content={"detalle": "Error de validación", "errores": errores}
    )

# Handler para errores HTTP
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.detail,
            "ruta": str(request.url),
            "metodo": request.method
        }
    )

# Handler para errores inesperados (500)
@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    logger.error(f"Error inesperado en {request.url}: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"error": "Error interno del servidor"}
    )


python
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class Producto(BaseModel):
    id: int
    nombre: str
    descripcion: Optional[str] = None
    stock: Optional[int] = None
    precio_oferta: Optional[float] = None

@app.get("/producto/{id}", response_model=Producto)
def obtener_producto(id: int):
    # Solo devuelve los campos que tienen valor
    return Producto(id=id, nombre="Portátil")
    # Sin exclude_unset: {"id":1,"nombre":"Portátil","descripcion":null,"stock":null,"precio_oferta":null}
    # Con exclude_unset:  {"id":1,"nombre":"Portátil"}  ← mucho más limpio


python
# Activa exclude_unset en el endpoint
@app.get(
    "/producto/{id}",
    response_model=Producto,
    response_model_exclude_unset=True  # ← aquí
)


python
from fastapi.middleware.gzip import GZipMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)
# Comprime respuestas > 1KB automáticamente
# Reduce el tamaño de respuestas JSON grandes hasta un 70%


python
from functools import lru_cache
from fastapi import FastAPI, Depends

app = FastAPI()

@lru_cache(maxsize=128)
def cargar_configuracion():
    """Se ejecuta solo una vez, el resultado queda en caché"""
    print("Cargando configuración...")  # Solo verás esto una vez
    return {"modelo": "gpt-4o-mini", "temperatura": 0.7}

@app.get("/config")
def obtener_config(config: dict = Depends(cargar_configuracion)):
    return config

mi-api-ia/
├── .env                    # Variables de entorno (no subir a Git)
├── .gitignore
├── requirements.txt
├── main.py                 # Punto de entrada
├── app/
│   ├── __init__.py
│   ├── config.py           # Configuración de la app
│   ├── models/
│   │   ├── __init__.py
│   │   └── usuario.py      # Modelos Pydantic
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── usuarios.py     # Rutas de usuarios
│   │   └── ia.py           # Rutas de IA
│   ├── services/
│   │   ├── __init__.py
│   │   └── ia_service.py   # Lógica de IA
│   └── dependencies.py     # Dependencias compartidas


python
# app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "Mi API de IA"
    openai_api_key: str
    debug: bool = False
    
    class Config:
        env_file = ".env"

settings = Settings()


python
# app/routers/usuarios.py
from fastapi import APIRouter, HTTPException

router = APIRouter(prefix="/usuarios", tags=["usuarios"])

@router.get("/")
def listar_usuarios():
    return []

@router.get("/{id}")
def obtener_usuario(id: int):
    raise HTTPException(status_code=404, detail="No encontrado")


python
# main.py
from fastapi import FastAPI
from app.routers import usuarios, ia
from app.config import settings

app = FastAPI(title=settings.app_name)

app.include_router(usuarios.router)
app.include_router(ia.router)

@app.get("/")
def raiz():
    return {"app": settings.app_name, "version": "1.0"}


bash
pip install langchain langchain-openai langchain-community openai python-dotenv

OPENAI_API_KEY=sk-tu-clave-aqui


python
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from dotenv import load_dotenv

load_dotenv()

# Crear el modelo
llm = ChatOpenAI(
    model="gpt-4o-mini",  # Modelo más económico para aprender
    temperature=0.7,       # 0 = determinista, 1 = muy creativo
    max_tokens=500
)

# Enviar un mensaje simple
respuesta = llm.invoke("Explica qué es la inteligencia artificial en 3 frases")
print(respuesta.content)

# Con mensajes del sistema (instrucciones de comportamiento)
mensajes = [
    SystemMessage(content="Eres un experto en Python que explica conceptos de forma sencilla"),
    HumanMessage(content="¿Qué es una lista en Python?")
]

respuesta = llm.invoke(mensajes)
print(respuesta.content)


python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini")

# Template básico
template = ChatPromptTemplate.from_messages([
    ("system", "Eres un asistente experto en {materia}. Responde siempre en español."),
    ("human", "{pregunta}")
])

# Formatear el template con valores concretos
mensajes = template.format_messages(
    materia="programación Python",
    pregunta="¿Cuándo debo usar una lista y cuándo un diccionario?"
)

respuesta = llm.invoke(mensajes)
print(respuesta.content)

# Template para generar contenido estructurado
template_resumen = PromptTemplate(
    input_variables=["texto", "longitud"],
    template="""
    Analiza el siguiente texto y proporciona:
    1. Un resumen en {longitud} palabras máximo
    2. Los 3 puntos clave
    3. El tono del texto (formal/informal/técnico)
    
    Texto: {texto}
    
    Responde en formato JSON.
    """
)

prompt_formateado = template_resumen.format(
    texto="Python es un lenguaje de programación de alto nivel...",
    longitud="50"
)
print(prompt_formateado)


python
from langchain_core.prompts import FewShotChatMessagePromptTemplate, ChatPromptTemplate
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Ejemplos de clasificación de sentimientos
ejemplos = [
    {
        "input": "¡Este producto es increíble, lo recomiendo a todos!",
        "output": '{"sentimiento": "positivo", "confianza": 0.98}'
    },
    {
        "input": "Llegó roto y el servicio al cliente no respondió.",
        "output": '{"sentimiento": "negativo", "confianza": 0.95}'
    },
    {
        "input": "El producto está bien, cumple su función.",
        "output": '{"sentimiento": "neutro", "confianza": 0.82}'
    },
]

# Template para cada ejemplo
ejemplo_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}"),
])

# Template few-shot
few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=ejemplo_prompt,
    examples=ejemplos,
)

# Template completo
prompt_final = ChatPromptTemplate.from_messages([
    ("system", "Clasifica el sentimiento de reseñas. Responde SOLO en JSON válido."),
    few_shot_prompt,
    ("human", "{input}"),
])

cadena = prompt_final | llm

reseñas = [
    "Tardó 3 semanas en llegar, decepcionante.",
    "¡Exactamente lo que buscaba! Perfecto.",
    "Ni bueno ni malo, es lo que es."
]

for reseña in reseñas:
    resultado = cadena.invoke({"input": reseña})
    print(f"Reseña: {reseña[:40]}...")
    print(f"Resultado: {resultado.content}\n")


python
# ❌ MAL prompt — vago, sin estructura
mal_prompt = "Resume esto: {texto}"

# ✅ BUEN prompt — rol + tarea + formato + restricciones
buen_prompt = """Eres un editor técnico senior con 10 años de experiencia.

Tu tarea es resumir el siguiente artículo técnico.

INSTRUCCIONES:
- Longitud máxima: 3 frases
- Incluye el concepto principal y las 2 conclusiones más importantes
- Usa lenguaje técnico pero accesible
- NO incluyas opiniones, solo hechos del texto

FORMATO DE RESPUESTA (JSON):
{{
  "resumen": "...",
  "concepto_principal": "...",
  "conclusiones": ["...", "..."]
}}

ARTÍCULO:
{texto}"""

# Truco: usa {{ }} para escapar llaves literales en f-strings y PromptTemplates


python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from pydantic import BaseModel
from typing import List
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini")

# Parser de texto simple
parser_texto = StrOutputParser()

# Parser de JSON
parser_json = JsonOutputParser()

# Modelo Pydantic para la respuesta
class RecetaAnalisis(BaseModel):
    ingredientes: List[str]
    tiempo_coccion: int  # minutos
    dificultad: str
    calorias_aprox: int

# Crear cadena: prompt → llm → parser
template = ChatPromptTemplate.from_messages([
    ("system", "Eres un chef experto. Responde SIEMPRE en JSON válido."),
    ("human", """
    Analiza esta receta: {receta}
    
    Devuelve un JSON con:
    - ingredientes: lista de ingredientes principales
    - tiempo_coccion: minutos estimados
    - dificultad: 'fácil', 'media' o 'difícil'
    - calorias_aprox: calorías aproximadas por ración
    """)
])

cadena = template | llm | parser_json

resultado = cadena.invoke({
    "receta": "Tortilla de patatas con cebolla pochada y aceite de oliva"
})

print(resultado)
print(type(resultado))  # <class 'dict'>


python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

# Cadena simple
cadena = (
    ChatPromptTemplate.from_template("Traduce al inglés: {texto}")
    | llm
    | parser
)

resultado = cadena.invoke({"texto": "El cielo es azul y el sol brilla"})
print(resultado)  # The sky is blue and the sun shines

# Cadena en secuencia: traducir y luego resumir
template_traducir = ChatPromptTemplate.from_template(
    "Traduce el siguiente texto al inglés: {texto}"
)

template_resumir = ChatPromptTemplate.from_template(
    "Resume el siguiente texto en una frase: {texto_traducido}"
)

# Encadenar los dos pasos
from langchain_core.runnables import RunnablePassthrough

cadena_compleja = (
    {"texto_traducido": template_traducir | llm | parser}
    | template_resumir
    | llm
    | parser
)

texto_espanol = "La inteligencia artificial está transformando la sociedad de maneras que no podíamos imaginar hace apenas una década, desde la medicina hasta el entretenimiento."

resultado = cadena_compleja.invoke({"texto": texto_espanol})
print(resultado)


python
from langchain_openai import ChatOpenAI
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini")

# Almacén de conversaciones (en memoria, por session_id)
store = {}

def obtener_historial(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# Template que incluye el historial
prompt = ChatPromptTemplate.from_messages([
    ("system", "Eres un asistente amigable. Recuerda todo lo que el usuario te cuenta."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{mensaje}")
])

# Cadena con memoria
cadena_base = prompt | llm
cadena_con_memoria = RunnableWithMessageHistory(
    cadena_base,
    obtener_historial,
    input_messages_key="mensaje",
    history_messages_key="history"
)

# Configuración de sesión
config = {"configurable": {"session_id": "usuario-123"}}

# Conversación
respuesta1 = cadena_con_memoria.invoke(
    {"mensaje": "Hola, me llamo Carlos y me gusta el fútbol"},
    config=config
)
print(f"Bot: {respuesta1.content}")

respuesta2 = cadena_con_memoria.invoke(
    {"mensaje": "¿Cuál es mi nombre y qué me gusta?"},
    config=config
)
print(f"Bot: {respuesta2.content}")
# El bot recordará que se llama Carlos y le gusta el fútbol


bash
pip install langchain-community faiss-cpu tiktoken


python


python
from langchain.text_splitter import RecursiveCharacterTextSplitter

# ❌ Chunking básico — puede partir frases por la mitad
splitter_basico = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0)

# ✅ Chunking con overlap — preserva contexto entre fragmentos
splitter_bueno = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=100,        # Los últimos 100 caracteres se repiten en el siguiente chunk
    length_function=len,
    separators=["\n\n", "\n", ". ", " ", ""]  # Prioriza cortar en párrafos
)

# ✅✅ Chunking semántico — agrupa por significado, no por tamaño
# pip install langchain-experimental
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

splitter_semantico = SemanticChunker(
    OpenAIEmbeddings(model="text-embedding-3-small"),
    breakpoint_threshold_type="percentile"  # Corta donde cambia el tema
)


python
# Reranking — mejora los resultados del retriever con un segundo modelo
# pip install langchain-cohere
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

# 1. Retriever base: recupera los 10 más similares (rápido pero impreciso)
retriever_base = vector_store.as_retriever(search_kwargs={"k": 10})

# 2. Reranker: de esos 10, selecciona los 3 más relevantes (lento pero preciso)
reranker_model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-base")
compressor = CrossEncoderReranker(model=reranker_model, top_n=3)

# 3. Retriever con reranking
retriever_mejorado = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever_base
)

# El retriever mejorado devuelve resultados mucho más relevantes
# sin cambiar nada más en tu cadena RAG
cadena_rag_mejorada = (
    {
        "contexto": retriever_mejorado | formatear_documentos,
        "pregunta": RunnablePassthrough()
    }
    | template | llm | parser
)


python
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

# Retriever por palabras clave (BM25, como un motor de búsqueda clásico)
bm25_retriever = BM25Retriever.from_documents(fragmentos)
bm25_retriever.k = 5

# Retriever vectorial (semántico)
faiss_retriever = vector_store.as_retriever(search_kwargs={"k": 5})

# Retriever híbrido — combina ambos con pesos
retriever_hibrido = EnsembleRetriever(
    retrievers=[bm25_retriever, faiss_retriever],
    weights=[0.4, 0.6]  # 40% palabras clave, 60% semántica
)


python
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import tool
from langchain import hub
from dotenv import load_dotenv
import math
import requests

load_dotenv()

# Definir herramientas (tools)
@tool
def calcular(expresion: str) -> str:
    """Evalúa expresiones matemáticas. Usa para cálculos."""
    try:
        resultado = eval(expresion, {"__builtins__": {}}, {"math": math})
        return str(resultado)
    except Exception as e:
        return f"Error: {e}"

@tool
def buscar_web(query: str) -> str:
    """Busca información en la web. Usa cuando necesites datos actuales."""
    # Simulación — en producción usarías la API de Tavily o similar
    return f"Resultados simulados para: {query}. Python 3.12 fue lanzado en octubre 2023."

@tool
def convertir_moneda(cantidad_origen: str) -> str:
    """Convierte EUR a USD. El formato es: '100 EUR'"""
    try:
        cantidad = float(cantidad_origen.split()[0])
        return f"{cantidad} EUR = {cantidad * 1.09:.2f} USD (tasa aproximada)"
    except:
        return "Error en el formato. Usa: '100 EUR'"

# Configurar el agente
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
herramientas = [calcular, buscar_web, convertir_moneda]

# Prompt para el agente (descarga de LangChain Hub)
prompt = hub.pull("hwchase17/react")

agente = create_react_agent(llm, herramientas, prompt)
ejecutor = AgentExecutor(agente, herramientas, verbose=True, max_iterations=5)

# El agente decidirá qué herramientas usar
resultado = ejecutor.invoke({
    "input": "Calcula la raíz cuadrada de 144 y conviértela en USD siendo EUR"
})

print(f"\nRespuesta final: {resultado['output']}")


python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)

template = ChatPromptTemplate.from_template(
    "Escribe un poema corto sobre {tema}"
)

cadena = template | llm

# Streaming en consola
print("Generando poema:")
for fragmento in cadena.stream({"tema": "la programación"}):
    print(fragmento.content, end="", flush=True)
print("\n---")


bash
pip install fastapi uvicorn langchain langchain-openai langchain-community \
            faiss-cpu python-dotenv pydantic-settings tiktoken sse-starlette

api-ia-produccion/
├── .env
├── requirements.txt
├── main.py
├── app/
│   ├── __init__.py
│   ├── config.py
│   ├── dependencies.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── chat.py
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── chat.py
│   │   └── documentos.py
│   └── services/
│       ├── __init__.py
│       ├── chat_service.py
│       └── rag_service.py
└── documentos/       # Carpeta para archivos a indexar


python
# app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "API de IA"
    openai_api_key: str
    openai_model: str = "gpt-4o-mini"
    temperatura: float = 0.7
    max_tokens: int = 1000
    
    class Config:
        env_file = ".env"

settings = Settings()


python
# app/models/chat.py
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
from enum import Enum

class RolMensaje(str, Enum):
    usuario = "usuario"
    asistente = "asistente"
    sistema = "sistema"

class Mensaje(BaseModel):
    rol: RolMensaje
    contenido: str
    timestamp: datetime = Field(default_factory=datetime.now)

class PeticionChat(BaseModel):
    session_id: str = Field(..., description="ID único de la sesión del usuario")
    mensaje: str = Field(..., min_length=1, max_length=2000)
    usar_rag: bool = Field(default=False, description="Si True, busca en documentos cargados")
    streaming: bool = Field(default=False)

class RespuestaChat(BaseModel):
    session_id: str
    respuesta: str
    tokens_usados: Optional[int] = None
    tiempo_respuesta_ms: Optional[int] = None

class PeticionDocumento(BaseModel):
    contenido: str
    nombre: str = "documento"
    metadatos: Optional[dict] = None


python


python


python
# app/routers/chat.py
from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from app.models.chat import PeticionChat, RespuestaChat
from app.services.chat_service import chat_service
from app.services.rag_service import rag_service
import time

router = APIRouter(prefix="/chat", tags=["Chat con IA"])

@router.post("/", response_model=RespuestaChat)
async def chat(peticion: PeticionChat):
    inicio = time.time()
    
    try:
        if peticion.usar_rag:
            respuesta = await rag_service.consultar(peticion.mensaje)
        else:
            respuesta = await chat_service.chat(
                peticion.session_id,
                peticion.mensaje
            )
        
        tiempo_ms = int((time.time() - inicio) * 1000)
        
        return RespuestaChat(
            session_id=peticion.session_id,
            respuesta=respuesta,
            tiempo_respuesta_ms=tiempo_ms
        )
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@router.get("/stream")
async def chat_stream(session_id: str, mensaje: str):
    """Endpoint de streaming — el texto llega en tiempo real"""
    async def generador():
        async for fragmento in chat_service.chat_stream(session_id, mensaje):
            yield f"data: {fragmento}\n\n"
        yield "data: [FIN]\n\n"
    
    return StreamingResponse(
        generador(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no"
        }
    )

@router.get("/historial/{session_id}")
async def obtener_historial(session_id: str):
    historial = chat_service.obtener_historial(session_id)
    return {"session_id": session_id, "mensajes": historial}

@router.delete("/sesion/{session_id}")
async def limpiar_sesion(session_id: str):
    chat_service.limpiar_sesion(session_id)
    return {"mensaje": f"Sesión {session_id} eliminada"}


python
# app/routers/documentos.py
from fastapi import APIRouter, UploadFile, File, HTTPException
from app.models.chat import PeticionDocumento
from app.services.rag_service import rag_service

router = APIRouter(prefix="/documentos", tags=["Gestión de Documentos"])

@router.post("/texto")
async def agregar_texto(peticion: PeticionDocumento):
    """Agrega texto directamente al índice"""
    fragmentos = rag_service.agregar_documentos(
        [peticion.contenido],
        [{"nombre": peticion.nombre, **(peticion.metadatos or {})}]
    )
    return {
        "mensaje": f"Documento '{peticion.nombre}' indexado",
        "fragmentos_creados": fragmentos
    }

@router.post("/archivo")
async def subir_archivo(archivo: UploadFile = File(...)):
    """Sube un archivo .txt y lo indexa"""
    if not archivo.filename.endswith(".txt"):
        raise HTTPException(status_code=400, detail="Solo se aceptan archivos .txt")
    
    contenido = await archivo.read()
    texto = contenido.decode("utf-8")
    
    fragmentos = rag_service.agregar_documentos(
        [texto],
        [{"nombre": archivo.filename}]
    )
    
    return {
        "mensaje": f"Archivo '{archivo.filename}' indexado",
        "fragmentos_creados": fragmentos,
        "tamaño_bytes": len(contenido)
    }

@router.get("/estado")
async def estado_rag():
    return {
        "tiene_documentos": rag_service.tiene_documentos(),
        "mensaje": "Listo para consultas" if rag_service.tiene_documentos() else "Sin documentos cargados"
    }


python
# main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import chat, documentos
from app.config import settings

app = FastAPI(
    title=settings.app_name,
    description="API de Inteligencia Artificial con FastAPI y LangChain",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Routers
app.include_router(chat.router)
app.include_router(documentos.router)

@app.get("/", tags=["Estado"])
def raiz():
    return {
        "app": settings.app_name,
        "estado": "operativo",
        "documentacion": "/docs"
    }

@app.get("/health", tags=["Estado"])
def health():
    return {"status": "ok"}


bash
# 1. Activar entorno virtual
source .venv/bin/activate  # macOS/Linux
# .venv\Scripts\activate   # Windows

# 2. Instalar dependencias
pip install -r requirements.txt

# 3. Crear .env con tu API key
echo "OPENAI_API_KEY=sk-tu-clave" > .env

# 4. Ejecutar el servidor
uvicorn main:app --reload --port 8000


python
from langchain_community.cache import InMemoryCache
from langchain.globals import set_llm_cache
from langchain_openai import ChatOpenAI
import time

# Habilitar caché en memoria
set_llm_cache(InMemoryCache())

llm = ChatOpenAI(model="gpt-4o-mini")

# Primera llamada (llama a la API de verdad)
inicio = time.time()
respuesta1 = llm.invoke("¿Cuánto es 2+2?")
print(f"Primera llamada: {time.time() - inicio:.2f}s")

# Segunda llamada idéntica (sale del caché, casi instantánea)
inicio = time.time()
respuesta2 = llm.invoke("¿Cuánto es 2+2?")
print(f"Segunda llamada (caché): {time.time() - inicio:.4f}s")

# Para persistir el caché entre reinicios, usa SQLite
from langchain_community.cache import SQLiteCache
set_llm_cache(SQLiteCache(database_path=".cache_llm.db"))


bash
pip install slowapi


python
from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/chat")
@limiter.limit("10/minute")  # Máximo 10 peticiones por minuto por IP
async def chat(request: Request, mensaje: str):
    # ... lógica del chat
    return {"respuesta": "..."}


python
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List, Literal
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Definir el esquema de respuesta exacto
class Ingrediente(BaseModel):
    nombre: str
    cantidad: str
    unidad: str

class RecetaEstructurada(BaseModel):
    titulo: str = Field(description="Nombre de la receta")
    tiempo_preparacion: int = Field(description="Minutos de preparación")
    tiempo_coccion: int = Field(description="Minutos de cocción")
    porciones: int
    dificultad: Literal["fácil", "media", "difícil"]
    ingredientes: List[Ingrediente]
    pasos: List[str]
    calorias_por_porcion: int

# Vincular el esquema al LLM — garantiza que la respuesta es válida
llm_estructurado = llm.with_structured_output(RecetaEstructurada)

receta = llm_estructurado.invoke(
    "Dame la receta de una tortilla española de 4 personas"
)

# receta es directamente un objeto RecetaEstructurada, sin parsear nada
print(receta.titulo)
print(receta.dificultad)
for ingrediente in receta.ingredientes:
    print(f"  - {ingrediente.cantidad} {ingrediente.unidad} de {ingrediente.nombre}")


python
import cProfile
import pstats
import io
from functools import wraps
import time
import asyncio

# Profiler simple para funciones async
def profile_async(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        inicio = time.perf_counter()
        resultado = await func(*args, **kwargs)
        duracion = time.perf_counter() - inicio
        print(f"[PERF] {func.__name__}: {duracion*1000:.2f}ms")
        return resultado
    return wrapper

# Profiler completo con cProfile
def profile_detallado(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        profiler = cProfile.Profile()
        profiler.enable()
        resultado = func(*args, **kwargs)
        profiler.disable()
        
        buffer = io.StringIO()
        stats = pstats.Stats(profiler, stream=buffer).sort_stats("cumulative")
        stats.print_stats(10)  # Top 10 funciones más lentas
        print(buffer.getvalue())
        return resultado
    return wrapper

# Uso en FastAPI
from fastapi import FastAPI
app = FastAPI()

@app.post("/chat-con-perfil")
@profile_async
async def chat_perfilado(mensaje: str):
    # Tu lógica aquí
    await asyncio.sleep(0.1)  # Simula procesamiento
    return {"respuesta": "ok"}


bash
# Herramienta externa: py-spy (profiling sin modificar código)
pip install py-spy

# Ver qué hace tu API en tiempo real (requiere proceso corriendo)
py-spy top --pid $(pgrep -f uvicorn)

# Generar flamegraph (visualización de rendimiento)
py-spy record -o perfil.svg --pid $(pgrep -f uvicorn)


python
# .env — añade estas variables
# LANGCHAIN_TRACING_V2=true
# LANGCHAIN_API_KEY=tu-clave-de-langsmith
# LANGCHAIN_PROJECT=mi-api-ia


python
import os
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini")
respuesta = llm.invoke("¿Qué es LangSmith?")

# Cada llamada quedará registrada en app.smith.langchain.com
# con el prompt exacto, la respuesta, latencia, tokens usados, etc.


python
# .env — añade estas variables
# LANGCHAIN_TRACING_V2=true
# LANGCHAIN_API_KEY=tu-clave-de-langsmith
# LANGCHAIN_PROJECT=mi-api-ia


python
import os
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

# Con las variables de entorno configuradas, LangChain
# enviará trazas automáticamente a LangSmith

llm = ChatOpenAI(model="gpt-4o-mini")
respuesta = llm.invoke("¿Qué es LangSmith?")

# Cada llamada quedará registrada en app.smith.langchain.com
# con el prompt exacto, la respuesta, latencia, tokens usados, etc.


bash
pip install pytest pytest-asyncio httpx


python
# tests/test_api.py
import pytest
from httpx import AsyncClient, ASGITransport
from main import app

@pytest.mark.asyncio
async def test_raiz():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as cliente:
        respuesta = await cliente.get("/")
    assert respuesta.status_code == 200
    assert "estado" in respuesta.json()

@pytest.mark.asyncio
async def test_health():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as cliente:
        respuesta = await cliente.get("/health")
    assert respuesta.status_code == 200
    assert respuesta.json() == {"status": "ok"}

@pytest.mark.asyncio
async def test_chat_requiere_session_id():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as cliente:
        respuesta = await cliente.post("/chat/", json={
            "mensaje": "Hola"
            # Falta session_id — debe dar error 422
        })
    assert respuesta.status_code == 422

@pytest.mark.asyncio
async def test_agregar_documento():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as cliente:
        respuesta = await cliente.post("/documentos/texto", json={
            "contenido": "Python es un lenguaje de programación",
            "nombre": "test_doc"
        })
    assert respuesta.status_code == 200
    assert "fragmentos_creados" in respuesta.json()


bash
pytest tests/ -v


dockerfile
# Dockerfile
FROM python:3.11-slim

# Directorio de trabajo
WORKDIR /app

# Instalar dependencias del sistema
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# Copiar e instalar dependencias Python
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copiar el código
COPY . .

# Puerto expuesto
EXPOSE 8000

# Comando de inicio
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]


yaml
# docker-compose.yml
version: "3.9"

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - LANGCHAIN_TRACING_V2=${LANGCHAIN_TRACING_V2}
      - LANGCHAIN_API_KEY=${LANGCHAIN_API_KEY}
    volumes:
      - ./documentos:/app/documentos
    restart: unless-stopped


bash
# Construir y ejecutar con Docker
docker build -t api-ia .
docker run -p 8000:8000 --env-file .env api-ia

# O con docker-compose
docker-compose up --build


nginx
# /etc/nginx/sites-available/api-ia
server {
    listen 80;
    server_name tudominio.com;

    location / {
        proxy_pass http://localhost:8000;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection 'upgrade';
        proxy_set_header Host $host;
        proxy_cache_bypass $http_upgrade;
        
        # Para streaming
        proxy_buffering off;
        proxy_read_timeout 300s;
    }
}


python
from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

# Analizar texto desde múltiples perspectivas a la vez
cadena_sentimiento = (
    ChatPromptTemplate.from_template("Analiza el sentimiento de: {texto}")
    | llm | parser
)

cadena_entidades = (
    ChatPromptTemplate.from_template("Extrae las entidades (personas, lugares) de: {texto}")
    | llm | parser
)

cadena_resumen = (
    ChatPromptTemplate.from_template("Resume en una frase: {texto}")
    | llm | parser
)

# Ejecutar las 3 cadenas en paralelo
analisis_paralelo = RunnableParallel(
    sentimiento=cadena_sentimiento,
    entidades=cadena_entidades,
    resumen=cadena_resumen
)

resultado = analisis_paralelo.invoke({
    "texto": "Ayer María García visitó el Museo del Prado en Madrid y quedó encantada con la obra de Velázquez."
})

print("Sentimiento:", resultado["sentimiento"])
print("Entidades:", resultado["entidades"])
print("Resumen:", resultado["resumen"])


python
from langchain_core.runnables import RunnableBranch

# Decidir qué cadena usar según el input
cadena_tecnica = (
    ChatPromptTemplate.from_template("Responde técnicamente: {pregunta}")
    | llm | parser
)

cadena_sencilla = (
    ChatPromptTemplate.from_template("Explica de forma muy simple, para niños: {pregunta}")
    | llm | parser
)

# Rama condicional
cadena_condicional = RunnableBranch(
    (lambda x: x.get("modo") == "tecnico", cadena_tecnica),
    cadena_sencilla  # caso por defecto
)

r1 = cadena_condicional.invoke({"pregunta": "¿Qué es un kernel?", "modo": "tecnico"})
r2 = cadena_condicional.invoke({"pregunta": "¿Qué es un kernel?", "modo": "simple"})

print("Técnico:", r1)
print("Simple:", r2)


python
from langchain.evaluation import load_evaluator
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Evaluador de relevancia
evaluador = load_evaluator("criteria", llm=llm, criteria="relevance")

evaluacion = evaluador.evaluate_strings(
    prediction="Python fue creado por Guido van Rossum en 1991.",
    input="¿Quién creó Python?",
    reference="Python fue creado por Guido van Rossum"
)

print(f"Resultado: {evaluacion['value']}")
print(f"Puntuación: {evaluacion['score']}")
print(f"Razonamiento: {evaluacion['reasoning']}")


bash
pip install langgraph


python


bash
# 1. Instalar Ollama desde https://ollama.ai
# 2. Descargar un modelo (Llama 3, Mistral, etc.)
ollama pull llama3.2        # Modelo de Meta, 3B parámetros (~2GB)
ollama pull mistral         # Excelente para tareas en inglés
ollama pull qwen2.5:7b     # Muy bueno en español y código

# 3. Ejecutar el servidor de Ollama (en segundo plano)
ollama serve


python
# Ollama con LangChain — misma interfaz que OpenAI
from langchain_ollama import ChatOllama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Solo cambia el LLM, todo lo demás es idéntico
llm_local = ChatOllama(
    model="llama3.2",
    temperature=0.7,
    # base_url="http://localhost:11434"  # Por defecto
)

template = ChatPromptTemplate.from_messages([
    ("system", "Eres un asistente experto en Python. Responde en español."),
    ("human", "{pregunta}")
])

cadena = template | llm_local | StrOutputParser()

respuesta = cadena.invoke({"pregunta": "¿Qué ventajas tiene usar async en Python?"})
print(respuesta)

# RAG con modelo local — 100% privado, sin datos a servidores externos
from langchain_ollama import OllamaEmbeddings

embeddings_locales = OllamaEmbeddings(model="nomic-embed-text")
# El resto del pipeline RAG es exactamente igual, solo cambia el embedding

Aspecto	OpenAI GPT-4o-mini	Ollama (Llama 3.2)
Coste	~$0.15/M tokens	Gratis
Velocidad	Rápido (red)	Depende del hardware
Calidad	Excelente	Muy buena (7B+)
Privacidad	Datos van a OpenAI	100% local
Límite de contexto	128K tokens	Configurable
Mejor para	Producción, calidad máxima	Desarrollo, datos privados


python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.callbacks import get_openai_callback

llm = ChatOpenAI(model="gpt-4o-mini")

# 1. Medir el coste exacto de una llamada
with get_openai_callback() as cb:
    respuesta = llm.invoke("Explica qué es una API en 50 palabras")
    print(f"Tokens usados: {cb.total_tokens}")
    print(f"Tokens prompt: {cb.prompt_tokens}")
    print(f"Tokens respuesta: {cb.completion_tokens}")
    print(f"Coste: ${cb.total_cost:.6f}")

# 2. Limitar tokens de respuesta según el caso de uso
llm_economico = ChatOpenAI(
    model="gpt-4o-mini",
    max_tokens=150,     # Para respuestas cortas
    temperature=0       # Temperatura 0 = más determinista y con menos tokens
)

# 3. Usar el modelo adecuado para cada tarea
# gpt-4o-mini  → clasificación, extracción, preguntas simples (muy barato)
# gpt-4o       → razonamiento complejo, código avanzado (más caro, úsalo poco)

# 4. Comprimir el historial de conversación para no acumular tokens
def comprimir_historial(historial: list, max_mensajes: int = 10) -> list:
    """Mantiene solo los últimos N mensajes + el mensaje del sistema"""
    if len(historial) <= max_mensajes:
        return historial
    # Conservar el primer mensaje (system) y los últimos N
    return [historial[0]] + historial[-(max_mensajes-1):]

# 5. Cachear embeddings — son caros si regeneras los mismos textos
from langchain.storage import LocalFileStore
from langchain.embeddings import CacheBackedEmbeddings
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

store = LocalFileStore(".cache_embeddings/")
embeddings_base = OpenAIEmbeddings(model="text-embedding-3-small")

# Solo genera embeddings si no están en caché
embeddings_cacheados = CacheBackedEmbeddings.from_bytes_store(
    embeddings_base,
    store,
    namespace=embeddings_base.model
)

# Primera vez: llama a la API
# Segunda vez con los mismos textos: sale del caché local ← gratis


python
import pytest
from unittest.mock import AsyncMock, patch
from langchain_core.messages import AIMessage

# 1. Mockear el LLM para tests unitarios rápidos
@pytest.mark.asyncio
async def test_chat_sin_llamar_a_openai():
    respuesta_mock = AIMessage(content="Respuesta simulada del LLM")
    
    with patch("app.services.chat_service.ChatOpenAI") as MockLLM:
        MockLLM.return_value.ainvoke = AsyncMock(return_value=respuesta_mock)
        
        from app.services.chat_service import ChatService
        servicio = ChatService()
        resultado = await servicio.chat("sesion-test", "Hola")
        
        assert resultado == "Respuesta simulada del LLM"
        assert MockLLM.return_value.ainvoke.called

# 2. Tests de evaluación — verificar calidad de respuestas reales
# (estos sí llaman a la API, ejecutarlos con menos frecuencia)
@pytest.mark.slow  # Marca para ejecutar solo en CI o manualmente
@pytest.mark.asyncio
async def test_calidad_respuesta_rag():
    from app.services.rag_service import RAGService
    servicio = RAGService()
    servicio.agregar_documentos(["FastAPI fue creado por Sebastián Ramírez en 2018"])
    
    respuesta = await servicio.consultar("¿Quién creó FastAPI?")
    
    # Verificación semántica, no exacta
    assert "Sebastián" in respuesta or "Ramírez" in respuesta
    assert len(respuesta) > 10  # No está vacía

# 3. Snapshot testing — detectar regresiones en el comportamiento del LLM
import json
import os

def guardar_snapshot(nombre: str, datos: dict):
    ruta = f"tests/snapshots/{nombre}.json"
    with open(ruta, "w") as f:
        json.dump(datos, f, ensure_ascii=False, indent=2)

def cargar_snapshot(nombre: str) -> dict:
    ruta = f"tests/snapshots/{nombre}.json"
    if not os.path.exists(ruta):
        return None
    with open(ruta) as f:
        return json.load(f)

Módulo	Contenido	Nivel
1	Variables, tipos, condicionales, bucles, listas, diccionarios, funciones, type hints, f-strings pro	Principiante
2	Manejo de errores, clases, módulos, archivos, decoradores, generadores, dataclasses	Principiante-Intermedio
3	Entornos virtuales, uv, pyproject.toml, VS Code, variables de entorno	Intermedio
4	FastAPI básico: rutas, métodos HTTP, Pydantic, query params	Intermedio
5	FastAPI avanzado: async/await, lifespan, manejo global de errores, GZip, caché	Intermedio-Avanzado
6	LangChain: LLMs, Prompt Templates, few-shot, prompt engineering, Output Parsers, LCEL	Avanzado
7	Memoria, RAG avanzado (reranking, búsqueda híbrida), Agentes, Streaming	Avanzado
8	Proyecto completo: API de IA con FastAPI + LangChain	Avanzado
9	Caché, rate limiting, structured outputs, profiling, observabilidad, tests, Docker	Experto
10	LangGraph, Ollama (modelos locales), optimización de costes, testing de LLMs	Experto

python

# Sin type hints (válido, pero menos claro)
def saludar(nombre):
    return "Hola, " + nombre

# Con type hints (FastAPI los usa para validar entradas en la API)
def saludar(nombre: str) -> str:
    return "Hola, " + nombre

bash

# Instala Homebrew si no lo tienes (el gestor de paquetes de macOS)
/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"

# Luego instala Python
brew install python

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from dotenv import load_dotenv

load_dotenv()

# 1. Cargar documento
with open("conocimiento.txt", "w") as f:
    f.write("""
    FastAPI es un framework web moderno y rápido para construir APIs con Python.
    Fue creado por Sebastián Ramírez y se publicó en 2018.
    FastAPI usa Pydantic para la validación de datos y Starlette como base.
    
    LangChain es un framework para desarrollar aplicaciones con modelos de lenguaje.
    Fue fundado por Harrison Chase en 2022.
    LangChain facilita la construcción de chatbots, agentes y sistemas RAG.
    
    Los embeddings son representaciones numéricas (vectores) de texto.
    Textos similares tienen embeddings cercanos en el espacio vectorial.
    OpenAI ofrece el modelo text-embedding-3-small para generar embeddings.
    """)

loader = TextLoader("conocimiento.txt", encoding="utf-8")
documentos = loader.load()

# 2. Dividir en fragmentos
splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=50)
fragmentos = splitter.split_documents(documentos)

# 3. Crear embeddings y base de datos vectorial
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = FAISS.from_documents(fragmentos, embeddings)

# 4. Crear retriever
retriever = vector_store.as_retriever(search_kwargs={"k": 3})

# 5. Template RAG
template = ChatPromptTemplate.from_messages([
    ("system", """Responde la pregunta basándote SOLO en el siguiente contexto.
    Si no puedes responder con el contexto dado, di que no lo sabes.
    
    Contexto: {contexto}"""),
    ("human", "{pregunta}")
])

def formatear_documentos(docs):
    return "\n\n".join(doc.page_content for doc in docs)

llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

# 6. Cadena RAG completa
cadena_rag = (
    {
        "contexto": retriever | formatear_documentos,
        "pregunta": RunnablePassthrough()
    }
    | template
    | llm
    | parser
)

preguntas = [
    "¿Quién creó FastAPI?",
    "¿Para qué sirven los embeddings?",
    "¿Cuál es la capital de Francia?"
]

for pregunta in preguntas:
    respuesta = cadena_rag.invoke(pregunta)
    print(f"P: {pregunta}")
    print(f"R: {respuesta}\n")

# app/services/chat_service.py
from langchain_openai import ChatOpenAI
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from app.config import settings

class ChatService:
    def __init__(self):
        self.llm = ChatOpenAI(
            model=settings.openai_model,
            temperature=settings.temperatura,
            max_tokens=settings.max_tokens,
            api_key=settings.openai_api_key
        )
        self.store: dict[str, InMemoryChatMessageHistory] = {}
        
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """Eres un asistente de IA inteligente y amigable.
            Responde siempre en el mismo idioma que el usuario.
            Sé conciso pero completo en tus respuestas."""),
            MessagesPlaceholder(variable_name="history"),
            ("human", "{mensaje}")
        ])
        
        cadena_base = self.prompt | self.llm
        self.cadena = RunnableWithMessageHistory(
            cadena_base,
            self._obtener_historial,
            input_messages_key="mensaje",
            history_messages_key="history"
        )
    
    def _obtener_historial(self, session_id: str) -> InMemoryChatMessageHistory:
        if session_id not in self.store:
            self.store[session_id] = InMemoryChatMessageHistory()
        return self.store[session_id]
    
    async def chat(self, session_id: str, mensaje: str) -> str:
        config = {"configurable": {"session_id": session_id}}
        respuesta = await self.cadena.ainvoke(
            {"mensaje": mensaje},
            config=config
        )
        return respuesta.content
    
    async def chat_stream(self, session_id: str, mensaje: str):
        """Genera la respuesta en streaming"""
        config = {"configurable": {"session_id": session_id}}
        async for fragmento in self.cadena.astream(
            {"mensaje": mensaje},
            config=config
        ):
            if hasattr(fragmento, 'content') and fragmento.content:
                yield fragmento.content
    
    def limpiar_sesion(self, session_id: str):
        if session_id in self.store:
            del self.store[session_id]
    
    def obtener_historial(self, session_id: str) -> list:
        if session_id not in self.store:
            return []
        historial = self.store[session_id]
        return [
            {"rol": msg.type, "contenido": msg.content}
            for msg in historial.messages
        ]

# Instancia singleton
chat_service = ChatService()

# app/services/rag_service.py
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
from app.config import settings
from typing import Optional
import os

class RAGService:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-small",
            api_key=settings.openai_api_key
        )
        self.llm = ChatOpenAI(
            model=settings.openai_model,
            api_key=settings.openai_api_key
        )
        self.vector_store: Optional[FAISS] = None
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=100
        )
        self.template = ChatPromptTemplate.from_messages([
            ("system", """Responde usando SOLO la información del contexto.
            Si no encuentras la respuesta, indícalo claramente.
            
            Contexto disponible:
            {contexto}"""),
            ("human", "{pregunta}")
        ])
        self.parser = StrOutputParser()
    
    def agregar_documentos(self, textos: list[str], metadatos: list[dict] = None):
        """Añade documentos al índice vectorial"""
        docs = []
        for i, texto in enumerate(textos):
            meta = metadatos[i] if metadatos else {}
            docs.append(Document(page_content=texto, metadata=meta))
        
        fragmentos = self.splitter.split_documents(docs)
        
        if self.vector_store is None:
            self.vector_store = FAISS.from_documents(fragmentos, self.embeddings)
        else:
            self.vector_store.add_documents(fragmentos)
        
        return len(fragmentos)
    
    async def consultar(self, pregunta: str, k: int = 3) -> str:
        """Responde una pregunta usando RAG"""
        if self.vector_store is None:
            return "No hay documentos cargados aún."
        
        retriever = self.vector_store.as_retriever(search_kwargs={"k": k})
        
        def formatear(docs):
            return "\n\n".join(d.page_content for d in docs)
        
        cadena = (
            {
                "contexto": retriever | formatear,
                "pregunta": RunnablePassthrough()
            }
            | self.template
            | self.llm
            | self.parser
        )
        
        return await cadena.ainvoke(pregunta)
    
    def tiene_documentos(self) -> bool:
        return self.vector_store is not None

rag_service = RAGService()

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, List, Annotated
import operator
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 1. Definir el estado del grafo
class EstadoConversacion(TypedDict):
    mensajes: Annotated[List, operator.add]  # Se acumulan con cada paso
    iteraciones: int
    necesita_busqueda: bool

# 2. Definir los nodos (funciones que transforman el estado)
def analizar_pregunta(estado: EstadoConversacion) -> dict:
    """Decide si la pregunta necesita búsqueda externa"""
    ultimo_mensaje = estado["mensajes"][-1].content
    
    palabras_clave_busqueda = ["actual", "hoy", "ahora", "último", "reciente", "2024", "2025"]
    necesita = any(p in ultimo_mensaje.lower() for p in palabras_clave_busqueda)
    
    return {"necesita_busqueda": necesita}

def buscar_informacion(estado: EstadoConversacion) -> dict:
    """Simula una búsqueda web"""
    # En producción usarías Tavily, SerpAPI, etc.
    return {
        "mensajes": [AIMessage(content="[Búsqueda] Información actualizada recuperada.")],
        "iteraciones": estado["iteraciones"] + 1
    }

def generar_respuesta(estado: EstadoConversacion) -> dict:
    """Genera la respuesta final"""
    respuesta = llm.invoke(estado["mensajes"])
    return {
        "mensajes": [respuesta],
        "iteraciones": estado["iteraciones"] + 1
    }

def decidir_siguiente_paso(estado: EstadoConversacion) -> str:
    """Función de routing — decide qué nodo ejecutar"""
    if estado["necesita_busqueda"] and estado["iteraciones"] == 0:
        return "buscar"
    return "responder"

# 3. Construir el grafo
grafo = StateGraph(EstadoConversacion)

grafo.add_node("analizar", analizar_pregunta)
grafo.add_node("buscar", buscar_informacion)
grafo.add_node("responder", generar_respuesta)

# Definir el flujo
grafo.set_entry_point("analizar")
grafo.add_conditional_edges(
    "analizar",
    decidir_siguiente_paso,
    {
        "buscar": "buscar",
        "responder": "responder"
    }
)
grafo.add_edge("buscar", "responder")
grafo.add_edge("responder", END)

# 4. Compilar y ejecutar
app_grafo = grafo.compile()

resultado = app_grafo.invoke({
    "mensajes": [HumanMessage(content="¿Cuál es la última versión de Python?")],
    "iteraciones": 0,
    "necesita_busqueda": False
})

print(resultado["mensajes"][-1].content)

Módulo 1 — Fundamentos de Python

1.1 ¿Qué es Python y por qué usarlo para IA?

Python es de tipado dinámico — igual que JavaScript

1.2 Instalación de Python

En Windows

En macOS

En Linux (Ubuntu/Debian)

1.3 Tu primer programa

1.4 Variables y tipos de datos

¿Cómo saber el tipo de una variable?

1.5 Operaciones básicas

🔥 Trucos pro con f-strings

1.6 Condicionales (if / elif / else)

1.7 Bucles

Bucle for — repite un número conocido de veces

Bucle while — repite mientras se cumpla una condición

1.8 Listas

🔥 Trucos pro con listas

1.9 Diccionarios

1.10 Funciones

Módulo 2 — Python Intermedio

2.1 Manejo de errores con try/except

2.2 Clases y Programación Orientada a Objetos (POO)

2.3 Módulos y paquetes

2.4 Trabajar con archivos

2.5 List comprehensions y expresiones útiles

2.6 Funciones avanzadas: *args, **kwargs y lambdas

2.7 Decoradores — el patrón más usado en FastAPI

2.8 Generadores — procesar datos enormes sin llenar la memoria

2.9 Dataclasses — alternativa moderna a clases simples

Módulo 3 — Entornos de trabajo y herramientas profesionales

3.1 Entornos virtuales

3.2 Editor de código: VS Code

3.3 pip y gestión de dependencias

🔥 uv — el reemplazo moderno de pip (10-100x más rápido)

pyproject.toml — el estándar moderno

3.4 Variables de entorno y el archivo .env

Módulo 4 — Introducción a FastAPI

4.1 ¿Qué es FastAPI?

4.2 Instalación

4.3 Tu primera API

4.4 Métodos HTTP

4.5 Modelos de datos con Pydantic

4.6 Parámetros de consulta (Query Parameters)

4.7 Manejo de errores en FastAPI

Módulo 5 — FastAPI avanzado

5.1 Código asíncrono con async/await

5.2 Dependencias en FastAPI

5.3 Middleware y CORS

5.4 Background Tasks

5.5 Lifespan — inicializar recursos al arrancar

5.6 Manejo global de excepciones

5.7 Optimizaciones de rendimiento en FastAPI

Usa response_model_exclude_unset=True

Compresión GZip automática

Caché de respuestas con functools.lru_cache

5.5 Estructura de proyecto profesional

Módulo 6 — Introducción a LangChain

6.1 ¿Qué es LangChain?

6.2 Instalación

6.3 Primer contacto con un LLM

6.4 Prompt Templates

🔥 Few-shot Prompting — enseñar con ejemplos

🔥 Prompt Engineering: las reglas que más importan

6.5 Output Parsers

6.6 LCEL — LangChain Expression Language

Módulo 7 — LangChain avanzado: Chains, Agentes y Memoria

7.1 Memoria en conversaciones

7.2 Retrieval Augmented Generation (RAG)

🔥 RAG avanzado: chunking inteligente y reranking

🔥 Búsqueda híbrida: vectores + palabras clave

7.3 Agentes — LLMs que toman decisiones

7.4 Streaming de respuestas

Módulo 8 — Proyecto Final: API de IA en producción

8.1 Instalación de dependencias del proyecto

8.2 Estructura del proyecto

8.3 Configuración central

8.4 Modelos Pydantic

8.5 Servicio de Chat

8.6 Servicio RAG

Bucle `for` — repite un número conocido de veces

Bucle `while` — repite mientras se cumpla una condición

Usa `response_model_exclude_unset=True`

Caché de respuestas con `functools.lru_cache`