@tool def sql_investigate(query: str) -> dict: try: df = con.execute(query).df() head = df.head(30) return { “rows”: int(len(df)), “columns”: list(df.columns), “preview”: head.to_dict(orient=”records”) } excepto Excepción como e: return {“error”: str(e)} @tool def log_pattern_scan(window_start_iso: str, window_end_iso: str, top_k: int = 8) -> dict: ws = pd.to_datetime(window_start_iso) we = pd.to_datetime(window_end_iso) df = logs_df[(logs_df[“ts”] >= ws) & (logs_df[“ts”] <= nosotros)].copy() si df.empty: devuelve {"filas": 0, "top_error_kinds": []"servicios_superiores": []"top_endpoints": []} df["error_kind_norm"] = gl["error_kind"].fillna("").replace("", "NINGUNO") err = df[df["level"].isin(["WARN","ERROR"])].copiar() top_err = errar["error_kind_norm"].value_counts().head(int(top_k)).to_dict() top_svc = err["service"].value_counts().head(int(top_k)).to_dict() top_ep = errar["endpoint"].value_counts().head(int(top_k)).to_dict() by_region = err.groupby("region").size().sort_values(ascending=False).head(int(top_k)).to_dict() p95_latency = float(np.percentile(df["latency_ms"].values, 95)) return { "rows": int(len(df)), "warn_error_rows": int(len(err)), "p95_latency_ms": p95_latency, "top_error_kinds": top_err, "top_services": top_svc, "top_endpoints": top_ep, "error_by_region": by_region } @tool def proponer_mitigaciones (hipótesis: str) -> dict: h = hipótesis.lower() mitigaciones = []
si “conn” en h o “pool” en h o “db” en h: mitigaciones += [
{“action”: “Increase DB connection pool size (bounded) and add backpressure at db-proxy”, “owner”: “Platform”, “eta_days”: 3},
{“action”: “Add circuit breaker + adaptive timeouts between api-gateway and db-proxy”, “owner”: “Backend”, “eta_days”: 5},
{“action”: “Tune query hotspots; add indexes for top offending endpoints”, “owner”: “Data/DBA”, “eta_days”: 7},
]
si “tiempo de espera” en h o “upstream” en h: mitigaciones += [
{“action”: “Implement hedged requests for idempotent calls (carefully) and tighten retry budgets”, “owner”: “Backend”, “eta_days”: 6},
{“action”: “Add upstream SLO-aware load shedding at api-gateway”, “owner”: “Platform”, “eta_days”: 7},
]
si “caché” en h: mitigaciones += [
{“action”: “Add request coalescing and negative caching to prevent cache-miss storms”, “owner”: “Backend”, “eta_days”: 6},
{“action”: “Prewarm cache for top endpoints during deploys”, “owner”: “SRE”, “eta_days”: 4},
]
si no son mitigaciones: mitigaciones += [
{“action”: “Add targeted dashboards and alerts for the suspected bottleneck metric”, “owner”: “SRE”, “eta_days”: 3},
{“action”: “Run controlled load test to reproduce and validate the hypothesis”, “owner”: “Perf Eng”, “eta_days”: 5},
]
mitigaciones = mitigaciones[:10]
return {“hipótesis”: hipótesis, “mitigaciones”: mitigaciones} @tool def draft_postmortem(título: cadena, window_start_iso: cadena, ventana_end_iso: cadena, impacto_cliente: cadena, causa_raíz_sospechada: cadena, key_facts_json: cadena, mitigaciones_json: cadena) -> dict: try: hechos = json.loads(key_facts_json) excepto Excepción: hechos = {“nota”: “key_facts_json no era un JSON válido”} intente: mits = json.loads(mitigations_json) excepto Excepción: mits = {“note”: “mitigations_json no era un JSON válido”} doc = { “title”: title, “date_utc”: datetime.utcnow().strftime(“%Y-%m-%d”), “incident_window_utc”: {“start”: window_start_iso, “end”: window_end_iso}, “customer_impact”: customer_impact, “suspected_root_cause”: sospecha_root_cause, “detection”: { “how_detected”: “Detección automática de anomalías + clasificación de picos de tasa de errores”, “brechas”: [“Add earlier saturation alerting”, “Improve symptom-to-cause correlation dashboards”]
}, “línea de tiempo”: [
{“t”: window_start_iso, “event”: “Symptoms begin (latency/error anomalies)”},
{“t”: “T+10m”, “event”: “On-call begins triage; identifies top services/endpoints”},
{“t”: “T+25m”, “event”: “Mitigation actions initiated (throttling/backpressure)”},
{“t”: window_end_iso, “event”: “Customer impact ends; metrics stabilize”},
]”key_facts”: hechos, “corrective_actions”: mits.get(“mitigaciones”, mits), “seguimientos”: [
{“area”: “Reliability”, “task”: “Add saturation signals + budget-based retries”, “priority”: “P1”},
{“area”: “Observability”, “task”: “Add golden signals per service/endpoint”, “priority”: “P1”},
{“area”: “Performance”, “task”: “Reproduce with load test and validate fix”, “priority”: “P2″},
]”appendix”: {“notes”: “Generado por un flujo de trabajo de múltiples agentes de Haystack (no RAG).”} } return {“postmortem_json”: doc} llm = OpenAIChatGenerator(model=”gpt-4o-mini”) state_schema = { “metrics_csv_path”: {“type”: str}, “logs_csv_path”: {“type”: str}, “metrics_summary”: {“type”: dict}, “logs_summary”: {“type”: dict}, “incident_window”: {“type”: dict}, “investigation_notes”: {“type”: list, “handler”: merge_lists}, “hypothesis”: {“type”: str}, “key_facts”: {“type”: dict}, “mitigation_plan”: {“type”: dict}, “postmortem”: {“type”: dict}, } perfiler_prompt = “””Usted es un perfilador de incidentes especializado. Objetivo: convertir métricas sin procesar/resúmenes de registros en hallazgos nítidos y de alta señal. Reglas: – Prefiera llamar a herramientas antes que adivinar. – La salida debe ser un objeto JSON con claves: ventana, síntomas, principales_contribuyentes, hipótesis, hechos_clave. – La hipótesis debe ser falsificable y mencionar al menos un servicio y mecanismo específico. “”” escritor_prompt = “””Es un escritor postmortem especializado. Objetivo: producir un JSON postmortem de alta calidad (no en prosa) utilizando la evidencia proporcionada y el plan de mitigación. Reglas: – Llame a las herramientas solo si es necesario. – Mantenga la ‘causa_raíz_sospechosa’ específica y no genérica. – Asegúrese de que las acciones correctivas tengan propietarios y días de eta. “”” coordinador_prompt = “””Usted es un comandante de incidentes que coordina un flujo de trabajo de múltiples agentes que no es RAG. Debe: 1) Cargar entradas 2) Encontrar una ventana de incidente (use p95_ms o error_rate) 3) Investigar con SQL dirigido y escaneo de patrones de registro 4) Solicitar al perfilador especialista que sintetice evidencia 5) Proponer mitigaciones 6) Solicitar al escritor especialista que redacte un JSON post mortem Devuelva una respuesta final con: – Un breve resumen ejecutivo (máximo 10 líneas) – El JSON post mortem – Una lista de verificación de runbook compacto (con viñetas) “”” agente_perfil = Agente( chat_generator=llm, herramientas=[load_inputs, detect_incident_window, sql_investigate, log_pattern_scan]system_prompt=profiler_prompt, condiciones_salida=[“text”]state_schema=state_schema ) escritor_agent = Agente( chat_generator=llm, herramientas=[draft_postmortem]system_prompt=escritor_prompt, condiciones_salida=[“text”]state_schema=state_schema ) perfil_herramienta = ComponentTool( componente=profiler_agent, nombre=”profiler_specialist”, descripción=”Sintetiza la evidencia del incidente en una hipótesis falsificable y hechos clave (salida JSON)”, outputs_to_string={“source”: “last_message”} ) escritor_herramienta = ComponentTool( componente=escritor_agent, nombre=”postmortem_writer_specialist”, descripción=”Redacta un JSON post mortem usando title/window/impact/rca/facts/mitigations.”, outputs_to_string={“source”: “last_message”} ) coordinador_agent = Agente( chat_generator=llm, herramientas=[
load_inputs,
detect_incident_window,
sql_investigate,
log_pattern_scan,
propose_mitigations,
profiler_tool,
writer_tool,
draft_postmortem
]system_prompt=coordinador_prompt, condiciones_salida=[“text”]esquema_estado=esquema_estado )
si “conn” en h o “pool” en h o “db” en h: mitigaciones += [
{“action”: “Increase DB connection pool size (bounded) and add backpressure at db-proxy”, “owner”: “Platform”, “eta_days”: 3},
{“action”: “Add circuit breaker + adaptive timeouts between api-gateway and db-proxy”, “owner”: “Backend”, “eta_days”: 5},
{“action”: “Tune query hotspots; add indexes for top offending endpoints”, “owner”: “Data/DBA”, “eta_days”: 7},
]
si “tiempo de espera” en h o “upstream” en h: mitigaciones += [
{“action”: “Implement hedged requests for idempotent calls (carefully) and tighten retry budgets”, “owner”: “Backend”, “eta_days”: 6},
{“action”: “Add upstream SLO-aware load shedding at api-gateway”, “owner”: “Platform”, “eta_days”: 7},
]
si “caché” en h: mitigaciones += [
{“action”: “Add request coalescing and negative caching to prevent cache-miss storms”, “owner”: “Backend”, “eta_days”: 6},
{“action”: “Prewarm cache for top endpoints during deploys”, “owner”: “SRE”, “eta_days”: 4},
]
si no son mitigaciones: mitigaciones += [
{“action”: “Add targeted dashboards and alerts for the suspected bottleneck metric”, “owner”: “SRE”, “eta_days”: 3},
{“action”: “Run controlled load test to reproduce and validate the hypothesis”, “owner”: “Perf Eng”, “eta_days”: 5},
]
mitigaciones = mitigaciones[:10]
return {“hipótesis”: hipótesis, “mitigaciones”: mitigaciones} @tool def draft_postmortem(título: cadena, window_start_iso: cadena, ventana_end_iso: cadena, impacto_cliente: cadena, causa_raíz_sospechada: cadena, key_facts_json: cadena, mitigaciones_json: cadena) -> dict: try: hechos = json.loads(key_facts_json) excepto Excepción: hechos = {“nota”: “key_facts_json no era un JSON válido”} intente: mits = json.loads(mitigations_json) excepto Excepción: mits = {“note”: “mitigations_json no era un JSON válido”} doc = { “title”: title, “date_utc”: datetime.utcnow().strftime(“%Y-%m-%d”), “incident_window_utc”: {“start”: window_start_iso, “end”: window_end_iso}, “customer_impact”: customer_impact, “suspected_root_cause”: sospecha_root_cause, “detection”: { “how_detected”: “Detección automática de anomalías + clasificación de picos de tasa de errores”, “brechas”: [“Add earlier saturation alerting”, “Improve symptom-to-cause correlation dashboards”]
}, “línea de tiempo”: [
{“t”: window_start_iso, “event”: “Symptoms begin (latency/error anomalies)”},
{“t”: “T+10m”, “event”: “On-call begins triage; identifies top services/endpoints”},
{“t”: “T+25m”, “event”: “Mitigation actions initiated (throttling/backpressure)”},
{“t”: window_end_iso, “event”: “Customer impact ends; metrics stabilize”},
]”key_facts”: hechos, “corrective_actions”: mits.get(“mitigaciones”, mits), “seguimientos”: [
{“area”: “Reliability”, “task”: “Add saturation signals + budget-based retries”, “priority”: “P1”},
{“area”: “Observability”, “task”: “Add golden signals per service/endpoint”, “priority”: “P1”},
{“area”: “Performance”, “task”: “Reproduce with load test and validate fix”, “priority”: “P2″},
]”appendix”: {“notes”: “Generado por un flujo de trabajo de múltiples agentes de Haystack (no RAG).”} } return {“postmortem_json”: doc} llm = OpenAIChatGenerator(model=”gpt-4o-mini”) state_schema = { “metrics_csv_path”: {“type”: str}, “logs_csv_path”: {“type”: str}, “metrics_summary”: {“type”: dict}, “logs_summary”: {“type”: dict}, “incident_window”: {“type”: dict}, “investigation_notes”: {“type”: list, “handler”: merge_lists}, “hypothesis”: {“type”: str}, “key_facts”: {“type”: dict}, “mitigation_plan”: {“type”: dict}, “postmortem”: {“type”: dict}, } perfiler_prompt = “””Usted es un perfilador de incidentes especializado. Objetivo: convertir métricas sin procesar/resúmenes de registros en hallazgos nítidos y de alta señal. Reglas: – Prefiera llamar a herramientas antes que adivinar. – La salida debe ser un objeto JSON con claves: ventana, síntomas, principales_contribuyentes, hipótesis, hechos_clave. – La hipótesis debe ser falsificable y mencionar al menos un servicio y mecanismo específico. “”” escritor_prompt = “””Es un escritor postmortem especializado. Objetivo: producir un JSON postmortem de alta calidad (no en prosa) utilizando la evidencia proporcionada y el plan de mitigación. Reglas: – Llame a las herramientas solo si es necesario. – Mantenga la ‘causa_raíz_sospechosa’ específica y no genérica. – Asegúrese de que las acciones correctivas tengan propietarios y días de eta. “”” coordinador_prompt = “””Usted es un comandante de incidentes que coordina un flujo de trabajo de múltiples agentes que no es RAG. Debe: 1) Cargar entradas 2) Encontrar una ventana de incidente (use p95_ms o error_rate) 3) Investigar con SQL dirigido y escaneo de patrones de registro 4) Solicitar al perfilador especialista que sintetice evidencia 5) Proponer mitigaciones 6) Solicitar al escritor especialista que redacte un JSON post mortem Devuelva una respuesta final con: – Un breve resumen ejecutivo (máximo 10 líneas) – El JSON post mortem – Una lista de verificación de runbook compacto (con viñetas) “”” agente_perfil = Agente( chat_generator=llm, herramientas=[load_inputs, detect_incident_window, sql_investigate, log_pattern_scan]system_prompt=profiler_prompt, condiciones_salida=[“text”]state_schema=state_schema ) escritor_agent = Agente( chat_generator=llm, herramientas=[draft_postmortem]system_prompt=escritor_prompt, condiciones_salida=[“text”]state_schema=state_schema ) perfil_herramienta = ComponentTool( componente=profiler_agent, nombre=”profiler_specialist”, descripción=”Sintetiza la evidencia del incidente en una hipótesis falsificable y hechos clave (salida JSON)”, outputs_to_string={“source”: “last_message”} ) escritor_herramienta = ComponentTool( componente=escritor_agent, nombre=”postmortem_writer_specialist”, descripción=”Redacta un JSON post mortem usando title/window/impact/rca/facts/mitigations.”, outputs_to_string={“source”: “last_message”} ) coordinador_agent = Agente( chat_generator=llm, herramientas=[
load_inputs,
detect_incident_window,
sql_investigate,
log_pattern_scan,
propose_mitigations,
profiler_tool,
writer_tool,
draft_postmortem
]system_prompt=coordinador_prompt, condiciones_salida=[“text”]esquema_estado=esquema_estado )