En este tutorial, creamos un flujo de trabajo agentic de estilo de producción de un extremo a otro utilizando GraphBit que demuestra cómo la ejecución estructurada en gráficos, la llamada de herramientas y los agentes opcionales controlados por LLM pueden coexistir en un solo sistema. Comenzamos inicializando e inspeccionando el tiempo de ejecución de GraphBit, luego definimos un dominio de ticket de atención al cliente realista con estructuras de datos tipificadas y herramientas deterministas ejecutables fuera de línea. Mostramos cómo estas herramientas se pueden componer en una canalización confiable basada en reglas para clasificación, enrutamiento y redacción de respuestas, y luego elevamos esa misma lógica a un flujo de trabajo GraphBit validado en el que los nodos de agentes orquestan el uso de herramientas a través de un gráfico dirigido. A lo largo del tutorial, mantenemos el sistema ejecutándose en modo fuera de línea y al mismo tiempo permitimos una promoción perfecta a la ejecución en línea simplemente proporcionando una configuración LLM, lo que ilustra cómo GraphBit admite la adopción gradual de inteligencia agente sin sacrificar la reproducibilidad o el control operativo. Consulte los códigos completos aquí.
!pip -q instalar graphbit rico pydantic numpy importar os importar tiempo importar json importar aleatorio desde clases de datos importar clase de datos desde escribir importar Dict, Cualquiera, Lista, Opcional importar numpy como np desde rico importar imprimir como rprint desde rich.panel importar Panel desde rich.table importar tabla
Comenzamos instalando todas las dependencias necesarias e importando las bibliotecas principales de Python, numéricas y de visualización necesarias para el tutorial. Configuramos el entorno de ejecución para que el portátil siga siendo autónomo y reproducible en Google Colab. Consulte los códigos completos aquí.
desde Graphbit import init, apagado, configure_runtime, get_system_info, health_check, versión desde Graphbit import Workflow, Node, Executor, LlmConfig de la herramienta de importación Graphbit, ToolExecutor, ExecutorConfig de Graphbit import get_tool_registry, clear_tools configure_runtime(worker_threads=4, max_blocking_threads=8, thread_stack_size_mb=2) init(log_level="warn", enable_tracing=False, debug=False) info = get_system_info() salud = salud_check() sys_table = Table(title="Información del sistema/Salud") sys_table.add_column("Clave", estilo="bold") sys_table.add_column("Valor") para k en ["versión", "python_binding_version", "cpu_count", "runtime_worker_threads", "runtime_initialized", "build_target", "build_profile"]: sys_table.add_row(k, str(info.get(k))) sys_table.add_row("graphbit_version()", str(version())) sys_table.add_row("overall_healthy", str(health.get("overall_healthy"))) rprint(sys_table)
Inicializamos el tiempo de ejecución de GraphBit y configuramos explícitamente sus parámetros de ejecución para controlar el uso de subprocesos y recursos. Luego consultamos los metadatos del sistema y realizamos una verificación de estado para verificar que el tiempo de ejecución esté inicializado correctamente. Consulte los códigos completos aquí.
@dataclass class Ticket: ticket_id: str user_id: str text: str create_at: float def make_tickets(n: int = 10) -> List[Ticket]: seeds = [ "El pago de mi tarjeta falló dos veces, ¿qué debo hacer?", "Quiero cancelar mi suscripción inmediatamente", "Su aplicación falla cuando abro el panel", "Actualice mi dirección de correo electrónico en la cuenta", "Reembolso no recibido después de 7 días", "Mi entrega se retrasa y el seguimiento está atascado.", "Sospecho que hay actividad fraudulenta en mi cuenta", "¿Cómo puedo cambiar la fecha de mi ciclo de facturación?", "El sitio web es muy lento y se agota el tiempo de espera", "Olvidé mi contraseña y no puedo iniciar sesión", "Detalles del proceso de devolución de cargo, por favor", "Necesito la factura del pago del mes pasado". ] aleatorio.shuffle(semillas) fuera =[]para i en rango(n): out.append( Ticket( ticket_id=f"T-{1000+i}", user_id=f"U-{random.randint(100,999)}", text=seeds[i % len(seeds)], creado_at=time.time() – random.randint(0, 7 * 24 * 3600), ) ) devolver boletos = make_tickets(10) rprint(Panel.fit("n".join([f"- {t.ticket_id}: {t.text}" for t in tickets]), title="Boletos de muestra"))
Definimos un modelo de datos fuertemente tipado para tickets de soporte y generamos un conjunto de datos sintéticos que simula problemas realistas de los clientes. Construimos tickets con marcas de tiempo e identificadores para reflejar los insumos de producción. Este conjunto de datos sirve como entrada compartida entre canalizaciones tanto fuera de línea como impulsadas por agentes. Consulte los códigos completos aquí.
clear_tools() @tool(_description="Clasificar un ticket de soporte en una categoría aproximada.") def classify_ticket(text: str) -> Dict[str, Any]: t = text.lower() if "fraude" en t o "fraudulento" en t: return {"category": "fraud", "priority": "p0"} if "cancelar" en t: return {"category": "cancellation", "priority": "p1"} si "reembolso" en t o "contracargo" en t: devolver {"categoría": "reembolsos", "prioridad": "p1"} si "contraseña" en t o "iniciar sesión" en t: devolver {"categoría": "account_access", "prioridad": "p2"} si "caída" en t o "lento" en t o "tiempo de espera" en t: devolver {"categoría": "error", "prioridad": "p2"} si "pago" en t o "facturación" en t o "factura" en t: return {"category": "billing", "priority": "p2"} if "entrega" en t o "seguimiento" en t: return {"category": "delivery", "priority": "p3"} return {"category": "general", "priority": "p3"} @tool(_description="Enrutar un ticket a una cola (devuelve ID de cola y horas de SLA).") def route_ticket(categoría: str, prioridad: str) -> Dict[str, Any]: queue_map = { "fraud": ("risk_ops", 2), "cancellation": ("retention", 8), "refunds": ("paids_ops", 12), "account_access": ("identity", 12), "bug": ("engineering_support", 24), "facturación": ("billing_support", 24), "entrega": ("logistics_support", 48), "general": ("support_general", 48), } q, sla = queue_map.get(categoría, ("support_general", 48)) if prioridad == "p0": sla = min(sla, 2) elif prioridad == "p1": sla = min(sla, 8) return {"queue": q, "sla_hours": sla} @tool(_description="Generar una respuesta de manual basada en categoría + prioridad.") def draft_response(categoría: str, prioridad: str, ticket_text: str) -> Dict[str, Any]: templates = { "fraud": "Hemos asegurado temporalmente su cuenta. Confirme las últimas 3 transacciones y restablezca las credenciales.", "cancellation": "Podemos ayudarlo a cancelar su suscripción. Confirme su plan y la fecha de vigencia desea.", "refunds": "Estamos verificando el estado del reembolso. Comparta la fecha y la referencia del pedido/pago.", "account_access": "Volvamos a ingresar. Utilice el enlace para restablecer la contraseña; si está bloqueado, verificaremos la identidad.", "bug": "Gracias por informar. Comparta el dispositivo/navegador + una captura de pantalla"., "billing": "Podemos ayudarlo con la facturación. "delivery": "Estamos verificando el estado del envío. Comparta su ID de seguimiento y dirección de entrega PIN/ZIP.", "general": "Gracias por comunicarse". } base = templates.get(categoría, plantillas["general"]) tono = "urgente" si prioridad == "p0" else ("rápido" si prioridad == "p1" else "estándar") return { "tono": tono, "mensaje": f"{base}nnContexto que recibimos: '{ticket_text}'", "next_steps": ["request_missing_info", "log_case", "route_to_queue"] } registro = get_tool_registry() herramientas_lista = registro.list_tools() si hasattr(registro, "lista_herramientas") más[]rprint(Panel.fit(f"Herramientas registradas: {tools_list}", título="Registro de herramientas"))
Registramos herramientas comerciales deterministas para clasificación, enrutamiento y redacción de respuestas de tickets utilizando la interfaz de herramientas de GraphBit. Codificamos la lógica de dominio directamente en estas herramientas para que puedan ejecutarse sin ninguna dependencia de LLM. Esto establece una base confiable y comprobable para la posterior orquestación de agentes. Consulte los códigos completos aquí.
tool_exec_cfg = ExecutorConfig( max_execution_time_ms=10_000, max_tool_calls=50, continue_on_error=False, store_results=True, enable_logging=False ) tool_executor = ToolExecutor(config=tool_exec_cfg) si "config" en ToolExecutor.__init__.__code__.co_varnames else ToolExecutor() def offline_triage(ticket: Ticket) -> Dict[str, Any]: c = classify_ticket(ticket.text) rt = route_ticket(c["categoría"], c["prioridad"]) dr = draft_response(c["categoría"], c["prioridad"], ticket.text) return { "ticket_id": ticket.ticket_id, "user_id": ticket.user_id, "categoría": c["categoría"], "prioridad": c["prioridad"], "cola": rt["cola"], "sla_hours": rt["sla_hours"], "borrador": dr["mensaje"], "tono": dr["tono"], "pasos": [ ("classify_ticket", c), ("route_ticket", rt), ("draft_response", dr), ] } offline_results = [offline_triage res_table = Table(title="Resultados de canalización sin conexión") res_table.add_column("Ticket", style="bold") res_table.add_column("Categoría") res_table.add_column("Prioridad") res_table.add_column("Cola") res_table.add_column("SLA (h)") para r en offline_results: res_table.add_row(r["ticket_id"], r["category"], r["priority"], r["queue"], str(r["sla_hours"])) rprint(res_table) prio_counts: Dict[str, int] = {} sla_vals: Lista[int] =[]para r en offline_results: prio_counts[r["priority"]] = prio_counts.get(r["priority"], 0) + 1 sla_vals.append(int(r["sla_hours"])) metrics = { "offline_mode": True, "tickets": len(offline_results), "priority_distribution": prio_counts, "sla_mean": float(np.mean(sla_vals)) if sla_vals else Ninguno, "sla_p95": float(np.percentile(sla_vals, 95)) if sla_vals else Ninguno, } rprint(Panel.fit(json.dumps(metrics, indent=2), title="Métricas sin conexión"))
Reunimos las herramientas registradas en un proceso de ejecución fuera de línea y lo aplicamos en todos los tickets para producir resultados de clasificación estructurados. Agregamos resultados en tablas y calculamos métricas de prioridad y SLA para evaluar el comportamiento del sistema. Demuestra cómo la lógica basada en GraphBit se puede validar de manera determinista antes de introducir agentes. Consulte los códigos completos aquí.
SYSTEM_POLICY = "Es un agente de operaciones de soporte confiable. Devuelve solo JSON ESTRICTO." flujo de trabajo = Workflow("Flujo de trabajo de clasificación de tickets (GraphBit)") resumen = Node.agent( name="Summarizer", agent_id="summarizer", system_prompt=SYSTEM_POLICY, Prompt="Resuma este ticket en 1-2 líneas. Devuelva JSON: {"summary":"…"}nTicket: {input}", temperatura=0.2, max_tokens=200) router_agent = Node.agent( name="RouterAgent", agent_id="router", system_prompt=SYSTEM_POLICY, Prompt=( "DEBES usar herramientas.n" "Llamar a classify_ticket(texto), route_ticket(categoría, prioridad), draft_response(categoría, prioridad, ticket_text).n" "Devolver JSON con campos: categoría, prioridad, cola, sla_hours, mensaje.n" "Ticket: {input}" ), herramientas=[classify_ticket, route_ticket, draft_response], temperatura=0.1, max_tokens=700 ) formatter = Node.agent( name="FinalFormatter", agent_id="final_formatter", system_prompt=SYSTEM_POLICY, Prompt=( "Validar el JSON y generar solo JSON ESTRICTO:n" "{"ticket_id":"…","category":"…","priority":"…","queue":"…","sla_hours":0,"customer_message":"…"}n" "Entrada: {entrada}" ), temperatura=0.0, max_tokens=500 ) sid = flujo de trabajo.add_node(summarizer) rid = flujo de trabajo.add_node(router_agent) fid = flujo de trabajo.add_node(formatter) flujo de trabajo.connect(sid, rid) flujo de trabajo.connect(rid, fid) flujo de trabajo.validate() rprint(Panel.fit("Flujo de trabajo validado: Summarizer -> RouterAgent -> FinalFormatter", title="Gráfico de flujo de trabajo"))
Construimos un flujo de trabajo GraphBit dirigido compuesto por múltiples nodos de agentes con responsabilidades claramente definidas y contratos JSON estrictos. Conectamos estos nodos en un gráfico de ejecución validado que refleja la lógica fuera de línea anterior a nivel de agente. Consulte los códigos completos aquí.
def pick_llm_config() -> Opcional[Cualquiera]: if os.getenv("OPENAI_API_KEY"): regresa LlmConfig.openai(os.getenv("OPENAI_API_KEY"), "gpt-4o-mini") if os.getenv("ANTHROPIC_API_KEY"): regresa LlmConfig.anthropic(os.getenv("ANTHROPIC_API_KEY"), "claude-sonnet-4-20250514") si os.getenv("DEEPSEEK_API_KEY"): devuelve LlmConfig.deepseek(os.getenv("DEEPSEEK_API_KEY"), "deepseek-chat") si os.getenv("MISTRALAI_API_KEY"): return LlmConfig.mistralai(os.getenv("MISTRALAI_API_KEY"), "mistral-large-latest") return Ninguno def run_agent_flow_once(ticket_text: str) -> Dict[str, Any]: llm_cfg = pick_llm_config() si llm_cfg es Ninguno: return { "mode": "fuera de línea", "nota": "Establezca OPENAI_API_KEY / ANTHROPIC_API_KEY / DEEPSEEK_API_KEY / MISTRALAI_API_KEY para habilitar la ejecución.", "input": ticket_text } executor = Executor(llm_cfg, Lightweight_mode=True, timeout_segundos=90, debug=False) si "lightweight_mode" está en Ejecutor.__init__.__code__.co_varnames else Ejecutor(llm_cfg) if hasattr(ejecutor, "configure"): executor.configure(timeout_segundos=90, max_retries=2, enable_metrics=True, debug=False) wf = Workflow("Ejecución de ticket único") s = Node.agent( name="Summarizer", agent_id="summarizer", system_prompt=SYSTEM_POLICY, Prompt=f"Resuma este ticket en 1-2 líneas. Devuelve JSON: {{"summary":"…"}}nTicket: {ticket_text}", temperatura=0.2, max_tokens=200 ) r = Node.agent( name="RouterAgent", agent_id="router", system_prompt=SYSTEM_POLICY, Prompt=( "DEBES usar herramientas.n" "Llamar a classify_ticket(text), route_ticket(categoría, prioridad), draft_response(categoría, prioridad, ticket_text).n" "Devolver JSON con campos: categoría, prioridad, cola, sla_hours, mensaje.n" f"Ticket: {ticket_text}" ), herramientas=[classify_ticket, route_ticket, draft_response], temperatura=0.1, max_tokens=700 ) f = Node.agent( name="FinalFormatter", agent_id="final_formatter", system_prompt=SYSTEM_POLICY, Prompt=( "Validar el JSON y generar solo JSON ESTRICTO:n" "{"ticket_id":"…","category":"…","priority":"…","queue":"…","sla_hours":0,"customer_message":"…"}n" "Entrada: {entrada}" ), temperatura=0.0, max_tokens=500 ) sid = wf.add_node(s) rid = wf.add_node(r) fid = wf.add_node(f) wf.connect(sid, rid) wf.connect(rid, fid) wf.validate() t0 = time.time() resultado = executor.execute(wf) dt_ms = int((time.time() – t0) * 1000) out = {"modo": "online", "execution_time_ms": dt_ms, "success": bool(result.is_success()) if hasattr(resultado, "is_success") else Ninguno} if hasattr(resultado, "get_all_variables"): out["variables"] = result.get_all_variables() else: out["raw"] = str(resultado)[:3000] return muestra = tickets[0]agent_run = run_agent_flow_once(sample.text) rprint(Panel.fit(json.dumps(agent_run, indent=2)[:3000], title="Ejecución del flujo de trabajo del agente")) rprint(Panel.fit("Listo", título="Completo"))
Agregamos configuración LLM opcional y lógica de ejecución que permite que el mismo flujo de trabajo se ejecute de forma autónoma cuando una clave de proveedor está disponible. Ejecutamos el flujo de trabajo en un solo ticket y capturamos el estado de ejecución y los resultados. Este último paso ilustra cómo el sistema pasa sin problemas del determinismo fuera de línea a la ejecución totalmente agente.
En conclusión, implementamos un flujo de trabajo completo de GraphBit que abarca la configuración del tiempo de ejecución, el registro de herramientas, la ejecución determinista fuera de línea, la agregación de métricas y la orquestación opcional basada en agentes con proveedores de LLM externos. Demostramos cómo la misma lógica de negocios se puede ejecutar tanto manualmente a través de herramientas como automáticamente a través de nodos de agentes conectados en un gráfico validado, destacando la fortaleza de GraphBit como sustrato de ejecución en lugar de simplemente un contenedor LLM. Demostramos que se pueden diseñar sistemas agentes complejos para fallar sin problemas, ejecutarse sin dependencias externas y aún escalar a flujos de trabajo totalmente autónomos cuando los LLM están habilitados.
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.
Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como emprendedor e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.
