From d56624c90e775d417837845dc9be2c4b66ee3d92 Mon Sep 17 00:00:00 2001 From: Juan Date: Sun, 10 May 2026 15:15:35 +0200 Subject: [PATCH] =?UTF-8?q?feat(runtime):=20a=C3=B1ade=20build=5Fgraph=20(?= =?UTF-8?q?StateGraph=20+=20routing=20condicional=20+=20interrupt=20HITL)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 --- core/src/agentforge_core/runtime/graph.py | 72 ++++++++++++ tests/unit/test_runtime_graph.py | 128 ++++++++++++++++++++++ 2 files changed, 200 insertions(+) create mode 100644 core/src/agentforge_core/runtime/graph.py create mode 100644 tests/unit/test_runtime_graph.py diff --git a/core/src/agentforge_core/runtime/graph.py b/core/src/agentforge_core/runtime/graph.py new file mode 100644 index 0000000..b9054e9 --- /dev/null +++ b/core/src/agentforge_core/runtime/graph.py @@ -0,0 +1,72 @@ +"""Compilación del grafo LangGraph para un AgentDefinition concreto.""" + +from __future__ import annotations + +from typing import Any + +from langgraph.checkpoint.base import BaseCheckpointSaver +from langgraph.graph import END, START, StateGraph + +from agentforge_core.domain.agent import AgentDefinition +from agentforge_core.domain.policy import PolicyDefinition +from agentforge_core.guardrails.base import GuardrailEngine +from agentforge_core.llm.base import LLMProvider +from agentforge_core.runtime.nodes import ( + build_node_approve_gate, + build_node_finalize, + build_node_llm_reason, + build_node_propose_actions, + build_node_validate_input, + build_node_validate_output, +) +from agentforge_core.runtime.state import AgentState + + +def build_graph( + *, + agent_def: AgentDefinition, + policy: PolicyDefinition, + provider: LLMProvider, + engine: GuardrailEngine, + checkpointer: BaseCheckpointSaver[Any], +) -> Any: + """Construye y compila el grafo de ejecución del agente.""" + g = StateGraph(AgentState) + + g.add_node("validate_input", build_node_validate_input(engine, policy)) + g.add_node("llm_reason", build_node_llm_reason(provider, agent_def)) + g.add_node("validate_output", build_node_validate_output(engine, policy)) + g.add_node("propose_actions", build_node_propose_actions()) + g.add_node("approve_gate", build_node_approve_gate(agent_def)) + g.add_node("finalize", build_node_finalize()) + + g.add_edge(START, "validate_input") + + def _after_validate_input(state: AgentState) -> str: + return END if state.get("status") == "blocked_by_guardrail" else "llm_reason" + + g.add_conditional_edges( + "validate_input", _after_validate_input, {END: END, "llm_reason": "llm_reason"} + ) + + def _after_llm(state: AgentState) -> str: + return END if state.get("status") == "failed" else "validate_output" + + g.add_conditional_edges( + "llm_reason", _after_llm, {END: END, "validate_output": "validate_output"} + ) + + def _after_validate_output(state: AgentState) -> str: + if state.get("status") in {"blocked_by_guardrail", "failed"}: + return END + return "propose_actions" + + g.add_conditional_edges( + "validate_output", _after_validate_output, {END: END, "propose_actions": "propose_actions"} + ) + + g.add_edge("propose_actions", "approve_gate") + g.add_edge("approve_gate", "finalize") + g.add_edge("finalize", END) + + return g.compile(checkpointer=checkpointer) diff --git a/tests/unit/test_runtime_graph.py b/tests/unit/test_runtime_graph.py new file mode 100644 index 0000000..2949342 --- /dev/null +++ b/tests/unit/test_runtime_graph.py @@ -0,0 +1,128 @@ +"""Tests del grafo compilado: camino feliz con MockProvider y pausa HITL.""" + +from datetime import UTC, datetime +from pathlib import Path +from uuid import uuid4 + +from agentforge_core.domain.agent import AgentDefinition, LLMConfig +from agentforge_core.domain.policy import PolicyDefinition +from agentforge_core.guardrails.guardrails_ai import GuardrailsAIEngine +from agentforge_core.llm.mock import MockProvider +from agentforge_core.runtime.checkpointer import build_checkpointer +from agentforge_core.runtime.graph import build_graph + + +def _agent() -> AgentDefinition: + return AgentDefinition( + name="incident_analyzer", + version="v1", + owner="Juan", + purpose="Análisis de incidentes", + state="active", + guardrails=["default"], + llm=LLMConfig(provider="mock"), + system_prompt="Eres un analista. Responde SIEMPRE con JSON.", + output_schema={"type": "object"}, + risk_threshold_for_hitl=4, + updated_at=datetime.now(UTC), + ) + + +def _policy_min() -> PolicyDefinition: + return PolicyDefinition( + name="min", + version="v1", + description="t", + input_validators=[], + output_validators=[], + ) + + +def _initial_state(trace_id: str, agent: AgentDefinition, user_input: str) -> dict: + return { + "trace_id": trace_id, + "agent_name": agent.name, + "agent_version": agent.version, + "user_input": user_input, + "messages": [], + "raw_llm_output": None, + "parsed_output": None, + "proposed_actions": [], + "violations": [], + "decision_path": [], + "status": "running", + "error": None, + "human_decision": None, + "final_output": None, + } + + +async def test_grafo_completa_camino_feliz_sin_hitl(tmp_path: Path) -> None: + agent = _agent() + trace_id = str(uuid4()) + config = {"configurable": {"thread_id": trace_id}} + async with build_checkpointer(tmp_path) as cp: + graph = build_graph( + agent_def=agent, + policy=_policy_min(), + provider=MockProvider(), + engine=GuardrailsAIEngine(), + checkpointer=cp, + ) + final = await graph.ainvoke( + _initial_state(trace_id, agent, "degradación MOS pool SBC"), # risk=2 → no HITL + config=config, + ) + assert final["status"] == "completed" + assert final["final_output"]["approved_actions"] # propagó la acción no riesgosa + + +async def test_grafo_pausa_en_hitl_si_riesgo_alto(tmp_path: Path) -> None: + agent = _agent() + trace_id = str(uuid4()) + config = {"configurable": {"thread_id": trace_id}} + async with build_checkpointer(tmp_path) as cp: + graph = build_graph( + agent_def=agent, + policy=_policy_min(), + provider=MockProvider(), + engine=GuardrailsAIEngine(), + checkpointer=cp, + ) + await graph.ainvoke( + _initial_state(trace_id, agent, "caída registros sip"), # act-1 risk=4 → HITL + config=config, + ) + # En interrupt, ainvoke devuelve el snapshot; el grafo queda con un paso pendiente. + state = await graph.aget_state(config) + assert state.next # interrumpido en approve_gate, esperando resume() + assert state.values["status"] == "running" + + +async def test_grafo_bloquea_por_guardrail_de_entrada(tmp_path: Path) -> None: + agent = _agent() + policy = PolicyDefinition( + name="block-pii", + version="v1", + description="bloquea emails en la entrada", + input_validators=[ + {"type": "detect_pii", "config": {"entities": ["EMAIL_ADDRESS"], "severity_on_match": "block"}} + ], + output_validators=[], + ) + trace_id = str(uuid4()) + config = {"configurable": {"thread_id": trace_id}} + async with build_checkpointer(tmp_path) as cp: + graph = build_graph( + agent_def=agent, + policy=policy, + provider=MockProvider(), + engine=GuardrailsAIEngine(), + checkpointer=cp, + ) + final = await graph.ainvoke( + _initial_state(trace_id, agent, "manda correo a juan@example.com"), + config=config, + ) + assert final["status"] == "blocked_by_guardrail" + assert final["raw_llm_output"] is None # nunca llegó al LLM