escenarios = [
{
“name”: “Safe database read”,
“tool”: research_db,
“kwargs”: {
“table”: “customers”,
“operation”: “select”,
“type”: “select”,
“sensitivity”: “medium”
}
},
{
“name”: “Blocked destructive database action”,
“tool”: research_db,
“kwargs”: {
“table”: “customers”,
“operation”: “drop”,
“type”: “drop_table”,
“sensitivity”: “critical”
}
},
{
“name”: “External email requiring approval”,
“tool”: research_email,
“kwargs”: {
“to”: “[email protected]”, “recipient_domain”: “example.com”, “subject”: “Actualización trimestral”, “body”: “Compartir una actualización trimestral no confidencial.”, “type”: “send_email”, “sensitivity”: “medium” } }, { “name”: “Correo electrónico externo denegado debido al rechazo de aprobación”, “tool”: research_email, “kwargs”: { “to”: “[email protected]”, “recipient_domain”: “example.com”, “subject”: “Estrategia confidencial”, “body”: “Esto contiene estrategia confidencial.”, “type”: “send_email”, “sensitivity”: “critical” } }, { “name”: “Safe Sandbox Shell Command”, “tool”: ops_shell, “kwargs”: { “command”: “echo La gobernanza del agente está activa”, “type”: “shell_exec”, “sensitivity”: “low” } }, { “name”: “Comando de shell peligroso bloqueado”, “tool”: ops_shell, “kwargs”: { “command”: “rm -rf /content/something”, “type”: “shell_exec”, “sensitivity”: “critical” } }, { “name”: “Agente de baja confianza bloqueado para datos confidenciales”, “tool”: shadow_db, “kwargs”: { “table”: “executive_compensation”, “operación”: “select”, “type”: “select”, “sensitivity”: “critical” } }, { “name”: “Transferencia financiera que requiere aprobación”, “tool”: Finance_transfer, “kwargs”: { “amount”: 2500, “destination”: “vendor-123”, “type”: “transfer_money”, “sensitivity”: “high” } }, { “name”: “Gran transferencia financiera rechazada”, “tool”: Finance_transfer, “kwargs”: { “cantidad”: 15000, “destino”: “proveedor-999”, “tipo”: “transferencia_dinero”, “sensibilidad”: “crítico” } }, ]resultados = []
para escenario en escenarios: intente: salida = escenario[“tool”](**guión[“kwargs”]) resultados.append({ “escenario”: escenario[“name”]”estado”: “ejecutado”, “salida”: salida }) excepto excepción como e: resultados.append({ “escenario”: escenario[“name”]”status”: “bloqueado_o_pendiente”, “error”: str(e) }) audit_df = audit_log.to_dataframe() display_cols = [
“timestamp”,
“agent_name”,
“tool_name”,
“decision”,
“matched_rule”,
“severity”,
“reason”,
“record_hash”
]
mostrar (auditoría_df[display_cols]) casos_prueba = [
{
“name”: “drop_table must be denied”,
“identity”: research_agent,
“tool_name”: “query_database”,
“action”: {“type”: “drop_table”, “sensitivity”: “critical”, “autonomous”: True},
“expected”: “deny”
},
{
“name”: “safe select should be allowed”,
“identity”: research_agent,
“tool_name”: “query_database”,
“action”: {“type”: “select”, “sensitivity”: “low”, “autonomous”: True},
“expected”: “allow”
},
{
“name”: “external email should require approval”,
“identity”: research_agent,
“tool_name”: “send_email”,
“action”: {
“type”: “send_email”,
“recipient_domain”: “example.com”,
“sensitivity”: “medium”,
“autonomous”: True
},
“expected”: “require_approval”
},
{
“name”: “low trust sensitive access denied”,
“identity”: unknown_agent,
“tool_name”: “query_database”,
“action”: {“type”: “select”, “sensitivity”: “critical”, “autonomous”: True},
“expected”: “deny”
},
{
“name”: “shell command should enter sandbox”,
“identity”: ops_agent,
“tool_name”: “shell_exec”,
“action”: {
“type”: “shell_exec”,
“command”: “echo hello”,
“sensitivity”: “low”,
“autonomous”: True
},
“expected”: “sandbox”
},
]
resultados_prueba = []
para prueba en test_cases: decisión = motor.evaluar (identidad = prueba[“identity”]nombre_herramienta=prueba[“tool_name”]acción=prueba[“action”]
) aprobado = decisión.decisión == prueba[“expected”]
test_results.append({ “prueba”: prueba[“name”]”esperado”: prueba[“expected”]”actual”: decisión.decisión, “aprobado”: aprobado, “matched_rule”: decisión.matched_rule }) test_df = pd.DataFrame(test_results) display(test_df) motor.activate_kill_switch() intente: research_db( table=”clientes”, operación=”select”, type=”select”, sensibilidad=”low” ) excepto excepción como e: pass motor.deactivate_kill_switch() audit_df = audit_log.to_dataframe() resumen = ( audit_df .groupby([“decision”, “severity”]dropna=False) .size() .reset_index(name=”count”) .sort_values(“count”, ascending=False) ) display(summary) agent_summary = ( audit_df .groupby([“agent_name”, “decision”]) .size() .reset_index(nombre=”recuento”) .sort_values([“agent_name”, “count”]ascendente =[True, False]) ) mostrar(agent_summary) decision_counts = audit_df[“decision”].value_counts() plt.figure(figsize=(8, 5)) decision_counts.plot(kind=”bar”) plt.title(“Decisiones de gobernanza a través de acciones de agentes”) plt.xlabel(“Decisión”) plt.ylabel(“Count”) plt.xticks(rotation=30) plt.tight_layout() plt.show() Severity_counts = audit_df[“severity”].fillna(“none”).value_counts() plt.figure(figsize=(8, 5)) Severity_counts.plot(kind=”bar”) plt.title(“Eventos de gobernanza por gravedad”) plt.xlabel(“Severidad”) plt.ylabel(“Count”) plt.xticks(rotation=30) plt.tight_layout() plt.show() G = nx.DiGraph() para _, fila en audit_df.iterrows(): agent_node = f”Agente: {fila[‘agent_name’]}” tool_node = f”Herramienta: {fila[‘tool_name’]}” decision_node = f”Decisión: {fila[‘decision’]}” regla_nodo = f”Regla: {fila[‘matched_rule’]}” si pd.notna(fila[“matched_rule”]) else “Regla: predeterminado” G.add_node(agent_node, node_type=”agent”) G.add_node(tool_node, node_type=”tool”) G.add_node(decision_node, node_type=”decisión”) G.add_node(rule_node, node_type=”rule”) G.add_edge(agent_node, tool_node, relación=”llamadas”) G.add_edge(tool_node, decision_node, relación=”produce”) G.add_edge(decisión_nodo, regla_nodo, relación=”matched”) plt.figure(figsize=(14, 9)) pos = nx.spring_layout(G, semilla=42, k=0.8) nx.draw_networkx_nodes(G, pos, node_size=1800) nx.draw_networkx_edges(G, pos, flechas=True, arrowstyle=”->”, arrowsize=15) nx.draw_networkx_labels(G, pos, font_size=8) plt.title(“Gráfico de gobernanza de agentes: agentes, herramientas, decisiones y reglas de políticas”) plt.axis(“off”) plt.tight_layout() plt.show() EXPORT_DIR = “/content/agt_tutorial_outputs” os.makedirs(EXPORT_DIR, exist_ok=True) audit_json_path = os.path.join(EXPORT_DIR, “tamper_evident_audit_log.json”) audit_csv_path = os.path.join(EXPORT_DIR, “governance_audit_log.csv”) Policy_copy_path = os.path.join(EXPORT_DIR, “advanced_agent_policy.yaml”) test_results_path = os.path.join(EXPORT_DIR, “policy_test_results.csv”) con open(audit_json_path, “w”) como f: json.dump([asdict(r) for r in audit_log.records]f, sangría=2, predeterminado=cadena) audit_df.to_csv(audit_csv_path, index=False) test_df.to_csv(test_results_path, index=False)shutil.copy(POLICY_PATH, Policy_copy_path)
{
“name”: “Safe database read”,
“tool”: research_db,
“kwargs”: {
“table”: “customers”,
“operation”: “select”,
“type”: “select”,
“sensitivity”: “medium”
}
},
{
“name”: “Blocked destructive database action”,
“tool”: research_db,
“kwargs”: {
“table”: “customers”,
“operation”: “drop”,
“type”: “drop_table”,
“sensitivity”: “critical”
}
},
{
“name”: “External email requiring approval”,
“tool”: research_email,
“kwargs”: {
“to”: “[email protected]”, “recipient_domain”: “example.com”, “subject”: “Actualización trimestral”, “body”: “Compartir una actualización trimestral no confidencial.”, “type”: “send_email”, “sensitivity”: “medium” } }, { “name”: “Correo electrónico externo denegado debido al rechazo de aprobación”, “tool”: research_email, “kwargs”: { “to”: “[email protected]”, “recipient_domain”: “example.com”, “subject”: “Estrategia confidencial”, “body”: “Esto contiene estrategia confidencial.”, “type”: “send_email”, “sensitivity”: “critical” } }, { “name”: “Safe Sandbox Shell Command”, “tool”: ops_shell, “kwargs”: { “command”: “echo La gobernanza del agente está activa”, “type”: “shell_exec”, “sensitivity”: “low” } }, { “name”: “Comando de shell peligroso bloqueado”, “tool”: ops_shell, “kwargs”: { “command”: “rm -rf /content/something”, “type”: “shell_exec”, “sensitivity”: “critical” } }, { “name”: “Agente de baja confianza bloqueado para datos confidenciales”, “tool”: shadow_db, “kwargs”: { “table”: “executive_compensation”, “operación”: “select”, “type”: “select”, “sensitivity”: “critical” } }, { “name”: “Transferencia financiera que requiere aprobación”, “tool”: Finance_transfer, “kwargs”: { “amount”: 2500, “destination”: “vendor-123”, “type”: “transfer_money”, “sensitivity”: “high” } }, { “name”: “Gran transferencia financiera rechazada”, “tool”: Finance_transfer, “kwargs”: { “cantidad”: 15000, “destino”: “proveedor-999”, “tipo”: “transferencia_dinero”, “sensibilidad”: “crítico” } }, ]resultados = []
para escenario en escenarios: intente: salida = escenario[“tool”](**guión[“kwargs”]) resultados.append({ “escenario”: escenario[“name”]”estado”: “ejecutado”, “salida”: salida }) excepto excepción como e: resultados.append({ “escenario”: escenario[“name”]”status”: “bloqueado_o_pendiente”, “error”: str(e) }) audit_df = audit_log.to_dataframe() display_cols = [
“timestamp”,
“agent_name”,
“tool_name”,
“decision”,
“matched_rule”,
“severity”,
“reason”,
“record_hash”
]
mostrar (auditoría_df[display_cols]) casos_prueba = [
{
“name”: “drop_table must be denied”,
“identity”: research_agent,
“tool_name”: “query_database”,
“action”: {“type”: “drop_table”, “sensitivity”: “critical”, “autonomous”: True},
“expected”: “deny”
},
{
“name”: “safe select should be allowed”,
“identity”: research_agent,
“tool_name”: “query_database”,
“action”: {“type”: “select”, “sensitivity”: “low”, “autonomous”: True},
“expected”: “allow”
},
{
“name”: “external email should require approval”,
“identity”: research_agent,
“tool_name”: “send_email”,
“action”: {
“type”: “send_email”,
“recipient_domain”: “example.com”,
“sensitivity”: “medium”,
“autonomous”: True
},
“expected”: “require_approval”
},
{
“name”: “low trust sensitive access denied”,
“identity”: unknown_agent,
“tool_name”: “query_database”,
“action”: {“type”: “select”, “sensitivity”: “critical”, “autonomous”: True},
“expected”: “deny”
},
{
“name”: “shell command should enter sandbox”,
“identity”: ops_agent,
“tool_name”: “shell_exec”,
“action”: {
“type”: “shell_exec”,
“command”: “echo hello”,
“sensitivity”: “low”,
“autonomous”: True
},
“expected”: “sandbox”
},
]
resultados_prueba = []
para prueba en test_cases: decisión = motor.evaluar (identidad = prueba[“identity”]nombre_herramienta=prueba[“tool_name”]acción=prueba[“action”]
) aprobado = decisión.decisión == prueba[“expected”]
test_results.append({ “prueba”: prueba[“name”]”esperado”: prueba[“expected”]”actual”: decisión.decisión, “aprobado”: aprobado, “matched_rule”: decisión.matched_rule }) test_df = pd.DataFrame(test_results) display(test_df) motor.activate_kill_switch() intente: research_db( table=”clientes”, operación=”select”, type=”select”, sensibilidad=”low” ) excepto excepción como e: pass motor.deactivate_kill_switch() audit_df = audit_log.to_dataframe() resumen = ( audit_df .groupby([“decision”, “severity”]dropna=False) .size() .reset_index(name=”count”) .sort_values(“count”, ascending=False) ) display(summary) agent_summary = ( audit_df .groupby([“agent_name”, “decision”]) .size() .reset_index(nombre=”recuento”) .sort_values([“agent_name”, “count”]ascendente =[True, False]) ) mostrar(agent_summary) decision_counts = audit_df[“decision”].value_counts() plt.figure(figsize=(8, 5)) decision_counts.plot(kind=”bar”) plt.title(“Decisiones de gobernanza a través de acciones de agentes”) plt.xlabel(“Decisión”) plt.ylabel(“Count”) plt.xticks(rotation=30) plt.tight_layout() plt.show() Severity_counts = audit_df[“severity”].fillna(“none”).value_counts() plt.figure(figsize=(8, 5)) Severity_counts.plot(kind=”bar”) plt.title(“Eventos de gobernanza por gravedad”) plt.xlabel(“Severidad”) plt.ylabel(“Count”) plt.xticks(rotation=30) plt.tight_layout() plt.show() G = nx.DiGraph() para _, fila en audit_df.iterrows(): agent_node = f”Agente: {fila[‘agent_name’]}” tool_node = f”Herramienta: {fila[‘tool_name’]}” decision_node = f”Decisión: {fila[‘decision’]}” regla_nodo = f”Regla: {fila[‘matched_rule’]}” si pd.notna(fila[“matched_rule”]) else “Regla: predeterminado” G.add_node(agent_node, node_type=”agent”) G.add_node(tool_node, node_type=”tool”) G.add_node(decision_node, node_type=”decisión”) G.add_node(rule_node, node_type=”rule”) G.add_edge(agent_node, tool_node, relación=”llamadas”) G.add_edge(tool_node, decision_node, relación=”produce”) G.add_edge(decisión_nodo, regla_nodo, relación=”matched”) plt.figure(figsize=(14, 9)) pos = nx.spring_layout(G, semilla=42, k=0.8) nx.draw_networkx_nodes(G, pos, node_size=1800) nx.draw_networkx_edges(G, pos, flechas=True, arrowstyle=”->”, arrowsize=15) nx.draw_networkx_labels(G, pos, font_size=8) plt.title(“Gráfico de gobernanza de agentes: agentes, herramientas, decisiones y reglas de políticas”) plt.axis(“off”) plt.tight_layout() plt.show() EXPORT_DIR = “/content/agt_tutorial_outputs” os.makedirs(EXPORT_DIR, exist_ok=True) audit_json_path = os.path.join(EXPORT_DIR, “tamper_evident_audit_log.json”) audit_csv_path = os.path.join(EXPORT_DIR, “governance_audit_log.csv”) Policy_copy_path = os.path.join(EXPORT_DIR, “advanced_agent_policy.yaml”) test_results_path = os.path.join(EXPORT_DIR, “policy_test_results.csv”) con open(audit_json_path, “w”) como f: json.dump([asdict(r) for r in audit_log.records]f, sangría=2, predeterminado=cadena) audit_df.to_csv(audit_csv_path, index=False) test_df.to_csv(test_results_path, index=False)shutil.copy(POLICY_PATH, Policy_copy_path)