En este tutorial, creamos un flujo de trabajo agente listo para producción que prioriza la confiabilidad sobre la generación del mejor esfuerzo al aplicar resultados estrictos y escritos en cada paso. Usamos PydanticAI para definir esquemas de respuesta claros, conectar herramientas mediante inyección de dependencia y garantizar que el agente pueda interactuar de forma segura con sistemas externos, como una base de datos, sin interrumpir la ejecución. Al ejecutar todo en una configuración asincrónica primero compatible con portátiles, demostramos cómo ir más allá de los frágiles patrones de chatbot hacia sistemas agentes robustos adecuados para flujos de trabajo empresariales reales.
!pip -q instalar “pydantic-ai-slim[openai]”pydantic import os, json, sqlite3 desde clases de datos importar clase de datos desde fecha y hora importar fecha y hora, zona horaria desde escribir importar Literal, Opcional, Lista desde pydantic importar BaseModel, Campo, field_validator desde pydantic_ai importar Agente, RunContext, ModelRetry si no os.environ.get(“OPENAI_API_KEY”): prueba: desde google.colab importar datos de usuario os.environ[“OPENAI_API_KEY”] = (userdata.get(“OPENAI_API_KEY”) o “”).strip() excepto Excepción: pasar si no os.environ.get(“OPENAI_API_KEY”): importar getpass os.environ[“OPENAI_API_KEY”] = getpass.getpass(“Pegue su OPENAI_API_KEY: “).strip() afirmar os.environ.get(“OPENAI_API_KEY”), “Se requiere OPENAI_API_KEY”.
Configuramos el entorno de ejecución y nos aseguramos de que todas las bibliotecas necesarias estén disponibles para que el agente se ejecute correctamente. Cargamos de forma segura la clave API de OpenAI de forma compatible con Colab para que el tutorial funcione sin cambios de configuración manuales. También importamos todas las dependencias principales que se compartirán entre esquemas, herramientas y lógica de agente.
Prioridad = Literal[“low”, “medium”, “high”, “critical”]
Tipo de acción = Literal[“create_ticket”, “update_ticket”, “query_ticket”, “list_open_tickets”, “no_action”]
Confianza = Literal[“low”, “medium”, “high”]
clase TicketDraft(BaseModel): título: str = Field(…, min_length=8, max_length=120) cliente: str = Field(…, min_length=2, max_length=60) prioridad: Categoría de prioridad: Literal[“billing”, “bug”, “feature_request”, “security”, “account”, “other”]
descripción: str = Field(…, min_length=20, max_length=1000) resultado_esperado: str = Field(…, min_length=10, max_length=250) clase AgentDecision(BaseModel): acción: ActionType motivo: str = Field(…, min_length=20, max_length=400) confianza: Ticket de confianza: Opcional[TicketDraft] = Ninguno ticket_id: Opcional[int] = Ninguno follow_up_questions: Lista[str] = Field(default_factory=list, max_length=5) @field_validator(“follow_up_questions”) @classmethod def short_questions(cls, v): para q en v: si len(q) > 140: elevar ValueError(“Cada pregunta de seguimiento debe tener <= 140 caracteres.") return v
Definimos los modelos de datos estrictos que actúan como contrato entre el agente y el resto del sistema. Usamos campos escritos y reglas de validación para garantizar que la respuesta de cada agente siga una estructura predecible. Al aplicar estos esquemas, evitamos que resultados con formato incorrecto se propaguen silenciosamente a través del flujo de trabajo.
@dataclass class SupportDeps: db: sqlite3.Connection inquilino: str política: dict def utc_now_iso() -> str: return datetime.now(timezone.utc).isoformat() def init_db() -> sqlite3.Connection: conn = sqlite3.connect(“:memory:”, check_same_thread=False) conn.execute(“”” CREAR TABLA tickets ( id INTEGER AUTOINCREMENTO DE CLAVE PRIMARIA, TEXTO DEL INQUILINO NO NULO, TEXTO DEL TÍTULO NO NULO, TEXTO DEL CLIENTE NO NULO, TEXTO NO NULO de prioridad, TEXTO NO NULO de la categoría, TEXTO NO NULO de la descripción, TEXTO NO NULO esperado, TEXTO NO NULO del estado, TEXTO NO NULO creado, TEXTO NO NULO actualizado, “””) conn.commit() return conn def seed_ticket(db: sqlite3.Connection, inquilino: str, ticket: TicketDraft, estado: str = “open”) -> int: now = utc_now_iso() cur = db.execute( “”” INSERTAR EN tickets (inquilino, título, cliente, prioridad, categoría, descripción, resultado_esperado, estado, creado_at, actualizado_at) VALORES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) “””, ( inquilino, ticket.title, ticket.customer, ticket.priority, ticket.category, ticket.description, ticket.expected_outcome, estado, ahora, ahora, ), ) db.commit() return int(cur.lastrowid)
Construimos la capa de dependencia e inicializamos una base de datos SQLite liviana para persistencia. Modelamos dependencias de tiempo de ejecución del mundo real, como conexiones de bases de datos y políticas de inquilinos, y las hacemos inyectables en el agente. También definimos funciones auxiliares que insertan y administran de forma segura los datos del ticket durante la ejecución.
def build_agent(nombre_modelo: str) -> Agente[SupportDeps, AgentDecision]: agente = Agente( f”openai:{model_name}”, salida_tipo=AgentDecision, salida_retries=2, instrucciones=( “Usted es un agente de clasificación de soporte de producción.\n” “Devuelve un resultado que coincide con el esquema de AgentDecision.\n” “Utilice herramientas cuando necesite el estado de la base de datos.\n” “Nunca invente ID de ticket.\n” “Si la intención del usuario no está clara, haga preguntas de seguimiento concisas.\n” ), ) @agent.tool def create_ticket(ctx: RunContext[SupportDeps]ticket: TicketDraft) -> int: deps = ctx.deps if ticket.priority in (“critical”, “high”) y deps.policy.get(“require_security_phrase_for_critical”, False): if ticket.category == “security” and “incident” not in ticket.description.lower(): rise ModelRetry(“Para seguridad alta/crítica, incluya la palabra ‘incidente’ en la descripción y reintentar.”) return seed_ticket(deps.db, deps.tenant, ticket, status=”open”) @agent.tool def update_ticket_status( ctx: RunContext[SupportDeps]ticket_id: int, estado: Literal[“open”, “in_progress”, “resolved”, “closed”]) -> dict: deps = ctx.deps now = utc_now_iso() cur = deps.db.execute(“SELECT id FROM tickets DONDE inquilino=? AND id=?”, (deps.tenant, ticket_id)) if not cur.fetchone(): rise ModelRetry(f”Ticket {ticket_id} no encontrado para este inquilino. Solicite el ticket_id correcto.”) deps.db.execute( “ACTUALIZAR tickets ESTABLECER estado=?, actualizado_at=? ¿DÓNDE inquilino=? Y id=?”, (estado, ahora, deps.tenant, ticket_id), ) deps.db.commit() return {“ticket_id”: ticket_id, “status”: estado, “updated_at”: ahora} @agent.tool def query_ticket(ctx: RunContext[SupportDeps]ticket_id: int) -> dict: deps = ctx.deps cur = deps.db.execute( “”” SELECCIONE id, título, cliente, prioridad, categoría, estado, creado_en, actualizado_at DESDE los tickets DONDE inquilino=? AND id=? “””, (deps.tenant, ticket_id),) fila = cur.fetchone() si no es fila: aumente ModelRetry(f”Ticket {ticket_id} no encontrado. Solicite al usuario un ticket_id válido.””) claves = [“id”, “title”, “customer”, “priority”, “category”, “status”, “created_at”, “updated_at”]
return dict(zip(teclas, fila)) @agent.tool def list_open_tickets(ctx: RunContext[SupportDeps]límite: int = 5) -> lista: deps = ctx.deps límite = max(1, min(int(límite), 20)) cur = deps.db.execute( “”” SELECCIONE ID, título, prioridad, categoría, estado, actualizado_en DESDE los tickets DONDE inquilino=? Y estado EN (‘abierto’, ‘en_progreso’) ORDEN POR actualizado_en DESC LÍMITE ? “””, (deps.tenant, límite), ) filas = cur.fetchall() retorno [
{“id”: r[0]”título”: r[1]”prioridad”: r[2]”categoría”: r[3]”estado”: r[4]”actualizado_en”: r[5]} para r en filas]@agent.output_validator def validar_decisión(ctx: RunContext[SupportDeps]out: AgentDecision) -> AgentDecision: deps = ctx.deps if out.action == “create_ticket” y out.ticket es Ninguno: elevar ModelRetry (“Eligió crear_ticket pero no proporcionó el ticket. Proporcione los campos del ticket y vuelva a intentarlo.”) if out.action in (“update_ticket”, “query_ticket”) y out.ticket_id es Ninguno: aumente ModelRetry (“Eligió actualizar/consultar pero no proporcionó ticket_id. Solicite ticket_id y vuelva a intentarlo.”) if out.ticket y out.ticket.priority == “crítico” y no deps.policy.get(“allow_critical”, True): aumente ModelRetry(“Este inquilino no permite ‘crítico’. Baje a ‘alto’ y vuelva a intentarlo.”) devuelva el agente de devolución
Contiene la lógica del agente central para ensamblar un agente PydanticAI independiente del modelo. Registramos herramientas escritas para crear, consultar, actualizar y enumerar tickets, permitiendo al agente interactuar con el estado externo de forma controlada. También aplicamos la validación de resultados para que el agente pueda autocorregirse siempre que sus decisiones violen las reglas comerciales.
db = init_db() deps = SupportDeps( db=db, inquilino=”acme_corp”, Policy={“allow_critical”: True, “require_security_phrase_for_critical”: True}, ) seed_ticket( db, deps.tenant, TicketDraft( title=”Doble cargo en la factura 8831″, cliente=”Riya”, prioridad=”alta”, categoría=”facturación”, descripción=”El cliente informa que se le facturó dos veces la factura 8831 y quiere un reembolso y un correo electrónico de confirmación.”, expected_outcome=”Emitir un reembolso y confirmar la resolución al cliente.”, ), ) seed_ticket( db, deps.tenant, TicketDraft( title=”La aplicación falla al iniciar sesión después de la actualización”, cliente=”Sam”, prioridad=”alta”, categoría=”error”, descripción=”Después de la última actualización, la aplicación falla inmediatamente al iniciar sesión. Reproducible en dos dispositivos; necesita investigación.”, expected_outcome=”Proporcionar una solución o solución alternativa y restaurar inicios de sesión exitosos.”, ), ) agent = build_agent(“gpt-4o-mini”) async def run_case(prompt: str): res = await agent.run(prompt, deps=deps) out = res.output print(json.dumps(out.model_dump(), indent=2)) return out case_a = await run_case( “Sospechamos que se ha apoderado de la cuenta: múltiples correos electrónicos de restablecimiento de contraseña e inicios de sesión no autorizados”. “Cliente=Leila. Prioridad=crítica. Abra un ticket de seguridad.” ) case_b = await run_case(“Enumere nuestros tickets abiertos y resuma qué abordar primero.”) case_c = await run_case(“¿Cuál es el estado del ticket 1? Si está abierto, muévalo a in_progress.”) agent_alt = build_agent(“gpt-4o”) alt_res = await agent_alt.run( “Crear un ticket de solicitud de función: cliente=Noah quiere ‘exportar a CSV’ en el panel de análisis; prioridad=media.”, deps=deps, ) print(json.dumps(alt_res.output.model_dump(), sangría=2))
Conectamos todo sembrando datos iniciales y ejecutando el agente de forma asincrónica, de forma segura para un portátil. Ejecutamos múltiples escenarios del mundo real para mostrar cómo el agente razona, llama a las herramientas y devuelve resultados válidos para el esquema. También demostramos con qué facilidad podemos cambiar el modelo subyacente manteniendo intactos los mismos flujos de trabajo y garantías.
En conclusión, mostramos cómo un agente con seguridad de tipos puede razonar, llamar a herramientas, validar sus propios resultados y recuperarse de errores sin intervención manual. Mantuvimos la lógica independiente del modelo, lo que nos permitió intercambiar LLM subyacentes y al mismo tiempo preservar los mismos esquemas y herramientas, lo cual es fundamental para la mantenibilidad a largo plazo. En general, demostramos cómo la combinación de una estricta aplicación de esquemas, inyección de dependencias y ejecución asíncrona cierra la brecha de confiabilidad en la IA agente y proporciona una base sólida para construir sistemas de producción confiables.
Consulte los códigos completos aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.
