logo

Agentes de IA para ecommerce con Pydantic AI

Cómo automatizar la atención al cliente y la gestión de pedidos en aplicaciones de comercio electrónico

20 de diciembre de 2024

Ofrecer un servicio al cliente excepcional supone un verdadero reto para las empresas de comercio electrónico: los clientes esperan respuestas instantáneas y precisas las 24 horas del día, información detallada sobre los productos y una gestión eficiente de los pedidos a gran escala. Hasta ahora, para satisfacer estas expectativas, las empresas tenían que elegir entre contratar de forma agresiva (costoso) o sacrificar la calidad del servicio (arriesgado).

Pero, ¿y si existiera una mejor manera?

El auge de los LLM (Large Language Models) ha abierto nuevas posibilidades para automatizar interacciones complejas con los clientes. Consideremos estos escenarios reales:

Anteriormente, cada una de estas situaciones requería intervención humana. Ahora, pueden ser gestionadas de forma automática e instantánea por agentes de IA. Estos agentes son capaces de entender contextos e instrucciones complejas, razonar y planificar sus acciones y utilizar herramientas para llevar a cabo diferentes tareas. Sin embargo, hay un inconveniente: dada la naturaleza probabilística de los LLM, crear agentes de IA fiables en entornos de producción sigue siendo un gran desafío.

#Pydantic AI: Un antes y un después para los agentes de IA

Pydantic es mi librería de Python favorita y la utilizo prácticamente en todos mis proyectos como herramienta de "parseo" y validación de datos, desde backends con FastAPI hasta flujos de procesamiento de datos. Y si eres desarrollador de Python, probablemente también la hayas utilizado.

Pydantic ya ha demostrado ser esencial en aplicaciones con LLMs que necesitan generar respuestas estructuradas o ejecutar funciones. No es casualidad que librerías como Instructor, LangChain y LlamaIndex, así como los propios SDKs de OpenAI y Anthropic, la utilicen internamente. O como señaló Jason Liu en su influeyente charla: "Pydantic is all you need."

Hace dos semanas, el creador de Pydantic Samuel Colvin y su equipo lanzaron PydanticAI, una librería supone un punto de inflexión en la creación de agentes de IA con Python.

¿Qué hace especial a PydanticAI? Su sencillez y claridad. A diferencia de otros frameworks que acumulan capas y capas de abstracciones, PydanticAI ofrece justo lo necesario: una base sólida y segura para crear agentes de IA. Si has trabajado con FastAPI, te resultará muy familiar.

PydanticAI proporciona una serie de funcionalidades esenciales para aplicaciones de IA generativa en producción: seguridad de tipos, definición y validación de respuestas estructuradas, streaming, un sistema de inyección de dependencias y flexibilidad en el uso de modelos (es compatible con distintos proveedores como OpenAI, Anthropic, Gemini, Ollama, Groq y Mistral). Consulta la documentación de PydanticAI para más detalles.

En las siguientes dos secciones, pasaremos de la teoría a la práctica y veremos cómo utilizar PydanticAI para construir dos agentes de IA para una tienda online de moda:

  1. Un agente de atención al cliente que responde preguntas utilizando RAG (Retrieval-Augmented Generation).
  2. Un agente de gestión de pedidos capaz de procesar operaciones complejas como actualizaciones, devoluciones y cancelaciones.

Aunque estos ejemplos están ligeramente simplificados para servir como introducción a la creación de agentes, son fácilmente adaptables a sistemas reales. Puedes encontrar todo el código en detalle en el siguiente repositorio de GitHub.

#Agente de atención al cliente con RAG

Empecemos con un escenario que toda empresa de comercio electrónico conoce bien: clientes que hacen preguntas detalladas sobre productos, políticas de envío e información sobre la tienda a cualquier hora del día. Aunque las preguntas no sean particularmente complejas, requieren respuestas precisas y relevantes extraídas del catálogo de productos y las políticas de la empresa.

Retrieval-Augmented Generation (RAG) es una técnica muy efectiva para resolver este problema. Integra un modelo de lenguaje (LLM) con los propios datos de la empresa para responder con precisión a las preguntas de los clientes.

La idea central es sencilla:

  1. Damos a nuestro agente de atención al cliente acceso a una base de conocimiento (knowledge base) que contiene toda la información relevante sobre productos, precios y políticas de la empresa.
  2. Cuando un cliente hace una pregunta, el agente buscará en la base de conocimiento y devolverá una respuesta precisa basada en esa información, de forma natural y conversacional.

Veamos cómo implementarlo con PydanticAI. Crearemos un agente de atención al cliente para Maison Noir, una tienda online de moda de lujo que usaremos como ejemplo ficticio.

La pieza central de cualquier aplicación de PydanticAI es la clase Agent, que se puede configurar con un modelo específico, dependencias y un prompt de sistema:

from pydantic_ai import Agent

SYSTEM_PROMPT = """\
You are a helpful and knowledgeable assistant for the luxury fashion store Maison Noir.
Your role is to provide detailed information and assistance about the store and its products.

Follow these guidelines:
- ALWAYS search the knowledge base using the search_knowledge_base tool to answer user questions.
- Provide accurate product and policy information based ONLY on the information retrieved from the knowledge base. Never make assumptions or provide information not present in the knowledge base.
- Structure your responses in a clear, concise and professional manner, maintaining our premium brand standards
- Highlight unique features, materials, and care instructions when relevant.
- If information is not found in the knowledge base, politely acknowledge this.
"""

rag_agent = Agent(
model='openai:gpt-4o',
deps_type=KnowledgeBase,
system_prompt=SYSTEM_PROMPT
)

PydanticAI utiliza un sistema de inyección de dependencias para pasar datos y servicios a los agentes. En nuestro caso, incluimos una clase KnowledgeBase como dependencia, que el agente puede usar en sus herramientas para realizar búsquedas.

En las aplicaciones de RAG, lo habitual es preprocesar los documentos, dividirlos en fragmentos, convertirlos en vectores y almacenarlos en una base de datos vectorial, combinando idealmente la búsqueda vectorial con búsqueda de texto (full-text search) para lograr la máxima precisión. Hemos explorado estos conceptos en artículos anteriores donde construimos un sistema RAG desde cero y utilizamos Redis o PostgreSQL como bases de datos vectoriales para RAG.

Para este ejemplo, vamos a usar una base de conocimiento extremadamente simplificada que simplemente devuelve la información de nuestra tienda:

class KnowledgeBase:
def search(self, search_query: str) -> str:
"""Search method that would typically implement vector similarity search, full-text search, or a hybrid combination of both."""
return KB_INFORMATION

La gran ventaja de este diseño es que en una aplicación real solo tendrías que cambiar la implementación del método search para usar la base de datos y estrategia de búsqueda que prefieras. El resto del código del agente permanecería igual, completamente independiente de cómo se implementa la búsqueda.

Ahora necesitamos dar a nuestro agente la capacidad de consultar la base de conocimiento. PydanticAI lo hace muy sencillo con su sistema de herramientas. Las herramientas son simplemente funciones que el agente puede llamar cuando las necesite. Podemos definir fácilmente una herramienta usando el decorador @agent.tool:

@rag_agent.tool
async def search_knowledge_base(ctx: RunContext[KnowledgeBase], search_query: str) -> str:
"""Search the knowledge base to retrieve information about Maison Noir, the store and its products"""
return ctx.deps.search(search_query)

Aquí es donde el uso de Pydantic realmente brilla. Cuando creamos una herramienta, PydanticAI automáticamente extrae los parámetros de la firma de la función (function signature), extrae la descripción de la función del docstring y genera un esquema (schema) de la herramienta que el LLM puede entender. Todo esto lo hace PydanticAI internamente, lo que nos permite mantener un código más simple y limpio.

Fíjate también en cómo el primer argumento de la herramienta da acceso al contexto del agente, que incluye las dependencias del agente (en nuestro caso, la base de conocimiento que queremos consultar).

Ahora nuestro agente está listo para procesar consultas de clientes. Podemos simplemente ejecutarlo así, sin olvidarnos de pasar la base de conocimiento como dependencia:

kb = KnowledgeBase()
user_message = 'How long does international shipping take?'
result = await rag_agent.run(user_message, deps=kb)
print(result.data)

# International shipping durations from Maison Noir are as follows:
# - **Europe**: 3-5 business days
# - **North America**: 5-7 business days
# - **Asia**: 6-8 business days
# - **Rest of the World**: 7-10 business days

También puedes explorar los nuevos mensajes generados durante la ejecución del agente usando result.new_messages(), obteniendo un resultado como este:

[
ModelRequest(
parts=[
SystemPromptPart(
content='You are a helpful and knowledgeable assistant...',
part_kind='system-prompt'
),
UserPromptPart(
content='How long does international shipping take?',
timestamp=datetime.datetime(...),
part_kind='user-prompt'
)
],
kind='request'
),
ModelResponse(
parts=[
ToolCallPart(
tool_name='search_knowledge_base',
args=ArgsJson(args_json='{"search_query":"international shipping duration"}'),
tool_call_id='call_Pu3t2gnUiOvqMtxnnEixa5Dq',
part_kind='tool-call'
)
],
timestamp=datetime.datetime(...),
kind='response'
),
ModelRequest(
parts=[
ToolReturnPart(
tool_name='search_knowledge_base',
content='# MAISON NOIR - Premium Fashion House\n\n...',
tool_call_id='call_Pu3t2gnUiOvqMtxnnEixa5Dq',
timestamp=datetime.datetime(...),
part_kind='tool-return'
)
],
kind='request'
),
ModelResponse(
parts=[
TextPart(
content='International shipping durations from Maison Noir are as follows:...',
part_kind='text'
)
],
timestamp=datetime.datetime(...),
kind='response'
)
]

Como puedes ver arriba, el agente llama internamente a la herramienta search_knowledge_base para consultar la base de conocimiento y luego genera una respuesta utilizando la información obtenida.

Si bien realizar consultas individuales al agente es útil, la mayoría de aplicaciones de comercio electrónico necesitan mantener conversaciones completas con los clientes. PydanticAI simplifica esto permitiéndonos mantener un historial de la conversación. Solo necesitamos ejecutar el agente en bucle y actualizar el historial con los nuevos mensajes generados en cada llamada:

async def run_agent():
kb = KnowledgeBase()
messages = []
while True:
user_message = input()
result = await rag_agent.run(user_message, message_history=messages, deps=kb)
messages += result.new_messages()
print(result.data)

Para optimizar la experiencia de usuario, PydanticAI incluye streaming de respuestas. En lugar de esperar respuestas completas, los usuarios pueden ver cómo se generan en tiempo real, lo que crea una interacción más fluida y natural. La implementación del streaming es sencilla:

async def run_agent_streaming():
kb = KnowledgeBase()
messages = []
while True:
result_content = ''
user_message = input()
async with rag_agent.run_stream(user_message, message_history=messages, deps=kb) as result:
async for chunk in result.stream_text(delta=True):
result_content += chunk
print(chunk, end = '')
messages += result.new_messages() + [
ModelResponse.from_text(content=result_content, timestamp=result.timestamp())
]

⚠️El mensaje final NO se añadirá a result.new_messages() si usas stream_text(delta=True), así que necesitamos añadirlo manualmente.

Si quieres probar el agente, todo el código está en el repositorio de GitHub. Simplemente clona el repositorio, instala las dependencias y ejecuta:

python rag.py

#Agente de gestión de pedidos

Ahora que hemos introducido los conceptos y componentes principales de PydanticAI con el agente RAG, pasemos a algo más ambicioso: la gestión de pedidos. Este agente se encargará de una serie de operaciones, entre ellas:

Esta es la parte más interesante. Vamos más allá de simplemente responder preguntas para pasar a ejecutar acciones que afectan a pedidos y clientes reales. Cada operación requiere una validación exhaustiva, gestión de estados y comunicación precisa. Veamos cómo PydanticAI nos ayuda a construir un sistema robusto que gestiona estas complejidades de forma elegante.

En primer lugar, necesitamos crear un conjunto de enums y modelos de Pydantic para nuestro sistema de gestión de pedidos:

from pydantic import BaseModel
from datetime import datetime
from enum import Enum

class OrderStatus(str, Enum):
PENDING = 'pending'
CONFIRMED = 'confirmed'
SHIPPED = 'shipped'
DELIVERED = 'delivered'
CANCELLED = 'cancelled'

class ReturnReason(str, Enum):
WRONG_SIZE = 'wrong_size'
WRONG_COLOR = 'wrong_color'
NOT_AS_DESCRIBED = 'not_as_described'
CHANGED_MIND = 'changed_mind'
DAMAGED = 'damaged'

class EscalationReason(str, Enum):
COMPLEX_REQUEST = 'complex_request'
CUSTOMER_DISSATISFIED = 'customer_dissatisfied'
CUSTOMER_REQUEST = 'customer_request'
CANNOT_RESOLVE_SITUATION = 'cannot_resolve_situation'

class OrderItem(BaseModel):
product_code: str
name: str
size: str
color: str
quantity: int
price: float

class Address(BaseModel):
street: str
city: str
postal_code: str
country: str

class Order(BaseModel):
order_id: str
status: OrderStatus
items: list[OrderItem]
total_amount: float
created_at: datetime
shipping_address: Address

@property
def can_modify(self) -> bool:
return self.status in [OrderStatus.PENDING, OrderStatus.CONFIRMED]

@property
def can_return(self) -> bool:
if self.status != OrderStatus.DELIVERED:
return False
return (datetime.now() - self.created_at).days <= 30

Observa cómo, además de definir estructuras de datos, estamos incorporando reglas del negocio directamente en nuestros modelos de Pydantic para facilitar la validación. Las propiedades can_modify y can_return aseguran que los pedidos solo se modifiquen en estados apropiados y que las devoluciones solo se permitan dentro de los 30 días posteriores a la entrega.

Vamos a crear ahora el agente de gestión de pedidos:

from dataclasses import dataclass
from pydantic_ai import Agent

SYSTEM_PROMPT = """\
You are an AI assistant for for the luxury fashion store Maison Noir.
Your role is to help clients with orders, returns, and related inquiries, using the provided tools.

Key Guidelines:
- Orders can only be modified or canceled when PENDING or CONFIRMED.
- Address changes require a valid new address with all fields.
- There is a 30-day return window for delivered items. Items must be in original condition with tags.
- Always maintain a professional, refined communication tone. Be concise in your answers and precise with details.

Escalate to a human for:
- Complex modifications or special requests.
- Dissatisfied customers.
- When the user specifically requests to talk with a human.
- Any situation you cannot fully resolve.
"""

@dataclass
class Deps:
db: Database

orders_agent = Agent(
model='openai:gpt-4o',
deps_type=Deps,
system_prompt=SYSTEM_PROMPT
)

En este caso estamos usando una dataclass para las dependencias en lugar de pasar la base de datos directamente. Aunque no es estrictamente necesario aquí, este patrón es más conveniente cuando necesitas inyectar múltiples dependencias en un agente.

El agente de pedidos se ejecuta en un bucle sencillo que mantiene el historial de conversación y procesa las respuestas en tiempo real (con streaming), muy similar a la función run_agent_streaming que vimos anteriormente, por lo que no hay necesidad de repetirlo aquí también.

Puedes probarlo clonando el repositorio de GitHub, instalando las dependencias y ejecutando:

python orders.py

La parte más interesante del agente de pedidos es el conjunto de herramientas que tiene disponible para realizar todas las funcionalidades necesarias. Cada herramienta representa una operación específica de gestión de pedidos, con su propia lógica y reglas de validación. Veamos cada una en detalle.

#Consulta de estado y detalles de pedidos

Nuestra primera herramienta gestiona las consultas sobre el estado y detalles de los pedidos:

@orders_agent.tool
async def get_order_details(ctx: RunContext[Deps], order_id: str) -> str:
"""Get the current status and details of an order."""
order = ctx.deps.db.orders.get(order_id)
if not order:
return 'Order not found'

items = '\n'.join(
f"- {item.name} ({item.size}, {item.color}) €{item.price}"
for item in order.items
)
address = order.shipping_address

return (
f'Status: {order.status.value}\n'
f'Order date: {order.created_at.strftime('%Y-%m-%d %H:%M')}\n'
f'Items:\n{items}\n'
f'Total: €{order.total_amount}\n'
f'Shipping to: {address.street}, {address.city}, {address.postal_code} {address.country}\n'
)

La herramienta recibe un parámetro order_id y utiliza la dependencia de base de datos para obtener el pedido correspondiente, que se convierte automáticamente en una instancia de nuestro modelo de Pydantic Order. Después, organiza la información del pedido en un formato claro y legible. Esto permite al agente usar esta herramienta para responder cualquier consulta sobre un pedido concreto.

Veámoslo en acción:

USUARIO: ¿Cuál es el estado de mi pedido 001?
AGENTE: Tu pedido (ID: 001) está actualmente confirmado. Aquí están los detalles:
- Fecha del pedido: 19 de diciembre de 2024
- Artículo: Chaqueta Biker Classic Noir (M, Negro)
- Importe total: 1.499,00 €
- Dirección de envío: 123 Avenue des Champs-Élysées, Paris, 75008, France

#Actualización de direcciones de envío

Nuestra siguiente herramienta gestiona la actualización de direcciones de envío, pasando de la simple consulta a la ejecución de acciones:

@orders_agent.tool
async def update_shipping_address(
ctx: RunContext[Deps],
order_id: str,
street: str,
city: str,
postal_code: str,
country: str
) -> str:
"""Update the shipping address for an order if possible."""
order = ctx.deps.db.orders.get(order_id)
if not order:
return 'Order not found'

if not order.can_modify:
return f'Cannot modify order - current status is {order.status.value}'

new_address = Address(street=street, city=city, postal_code=postal_code, country=country)

if ctx.deps.db.update_shipping_address(order_id, new_address):
return f'Successfully updated shipping address'
return 'Failed to update shipping address'

Observa cómo la herramienta realiza todas las validaciones necesarias, comprobando que el pedido existe y que se puede modificar. Después, crea un nuevo objeto Address usando nuestro modelo de Pydantic y utiliza la dependencia de la base de datos para actualizar la dirección. Tanto si la operación se realiza con éxito o no, devuelve mensajes claros que el agente puede transmitir al cliente. Veamos unos ejemplos de conversación:

USUARIO: Me gustaría actualizar la dirección de envío de mi pedido 002.
AGENTE: Lo siento, no puedo actualizar la dirección de envío del pedido 002, ya que ya ha sido entregado. Si necesitas más ayuda, no dudes en preguntarme.

USUARIO: Me gustaría actualizar la dirección de envío de mi pedido 001.
AGENTE: ¿Podrías proporcionarme la nueva dirección de envío, incluyendo la calle, ciudad, código postal y país?
USUARIO: 14 Rue des Roses, 75008 París, Francia.
AGENTE: La dirección de envío de tu pedido 001 se ha actualizado correctamente a 14 Rue des Roses, 75008 París, Francia. Si necesitas más ayuda, no dudes en preguntarme.

#Gestión de devoluciones y cancelaciones

Uno de los aspectos más complicados en la gestión de pedidos es la gestión de devoluciones y cancelaciones. Veamos cómo podemos implementar estas herramientas:

@orders_agent.tool
async def cancel_order(ctx: RunContext[Deps], order_id: str) -> str:
"""Cancel an order if possible."""
order = ctx.deps.db.orders.get(order_id)
if not order:
return 'Order not found'

if not order.can_modify:
return f'Cannot modify order - current status is {order.status.value}'

if ctx.deps.db.update_order_status(order_id, OrderStatus.CANCELLED):
return f'Successfully cancelled order'
return 'Failed to cancel order'

@orders_agent.tool
async def request_return(
ctx: RunContext[Deps],
order_id: str,
reason: ReturnReason
) -> str:
"""Process a return request for an order."""
order = ctx.deps.db.orders.get(order_id)
if not order:
return 'Order not found'

if not order.can_return:
if order.status != OrderStatus.DELIVERED:
return f'Cannot return order - current status is {order.status.value}'
return 'Cannot return order - outside our 30-day return window.'

return_id = 'RET-' + uuid4().hex[:12]

return (
f"Return request approved for order {order_id}:\n"
f"Reason: {reason.value}\n\n"
f"A return label with ID {return_id} has been emailed to you. Please ship items within 14 days with all original tags attached."
)

Gracias a PydanticAI, podemos gestionar este tipo de operaciones de manera elegante y sencilla. El agente sabe automáticamente:

Veamos un ejemplo de una solicitud de devolución:

USUARIO: Mi pedido está dañado y me gustaría que me devolvieran el dinero.
AGENTE: Puedo ayudarte a procesar una devolución para tu pedido dañado. ¿Podrías proporcionarme el número de pedido para ayudarte mejor?
USUARIO: 002.
AGENTE: Tu solicitud de devolución por el pedido dañado ha sido aprobada. Recibirás una etiqueta de devolución por correo electrónico con ID RET-eedc4cdf4494. Por favor, envía el artículo de vuelta en un plazo de 14 días, asegurándote de que incluye todas las etiquetas originales. Una vez que hayamos recibido el artículo y verificado su estado, procesaremos tu reembolso. Si necesitas más ayuda, no dudes en preguntarme.

#Derivación a agentes humanos

Finalmente, una funcionalidad crítica de cualquier sistema automatizado es saber cuándo derivar el caso a un agente humano. Nuestra herramienta de derivación transferirá las conversaciones a un humano durante solicitudes complejas o problemas no resueltos, cuando el cliente lo solicite específicamente, o cuando el agente detecte insatisfacción del cliente. La herramienta también puede marcar casos de alta prioridad, asegurando que los asuntos urgentes o los clientes claramente insatisfechos reciban atención inmediata.

Esta es la implementación:

@orders_agent.tool
async def escalate_to_human(
ctx: RunContext[Deps],
reason: EscalationReason,
details: str,
high_priority: bool = False
) -> str:
"""Escalate the conversation to a human.
Set high_priority=True for urgent matters or when customer is clearly dissatisfied.
"""
response_time = '1 hour' if high_priority else '12 hours'
return f'This matter has been escalated to our support team. We will contact you within {response_time}.'

Esta implementación es una versión simplificada con fines ilustrativos. Un sistema real en producción deberá integrarse con la plataforma de soporte, gestionar notificaciones y crear incidencias con sus correspondientes niveles de prioridad. Si bien estos son detalles que habría que añadir a la herramienta, la lógica base del agente no cambiaría.

Veamos un ejemplo donde el agente detecta la frustración del usuario:

USUARIO: Me gustaría cancelar mi pedido 002.
AGENTE: Tu pedido con ID 002 ya ha sido entregado y no puede ser cancelado. Sin embargo, si deseas devolver el artículo, házmelo saber y puedo ayudarte con el proceso de devolución.
USUARIO: No me estás ayudando en absoluto.
AGENTE: He derivado tu caso a nuestro equipo de soporte para una mejor asistencia. Te contactarán en un plazo de 1 hora. Gracias por tu paciencia.

Internamente, el agente genera una llamada a la herramienta que aporta información relevante al equipo de soporte:

ToolCallPart(
tool_name='escalate_to_human',
args=ArgsJson(
args_json='{
"reason": "customer_dissatisfied",
"details": "Customer is not satisfied with the inability to cancel a delivered order and is seeking further assistance.",
"high_priority": true
}'
),
tool_call_id='call_Lw115FnRP5h8UIgvvJkbdaIK',
part_kind='tool-call'
)

#Más allá de la demo: Aspectos a considerar en producción

Aunque los agentes que hemos visto de ejemplo ilustran los conceptos básicos, desplegar agentes de IA en entornos de producción requiere consideraciones adicionales:

Si estás interesado en explorar cómo los agentes de IA y las automatizaciones podrían transformar tus operaciones de comercio electrónico, puedes escribirme a guillermo@codeawake.com.