From 683cc69edc67e56763148722737a8823f0c38756 Mon Sep 17 00:00:00 2001 From: Juan Date: Sun, 10 May 2026 10:56:25 +0200 Subject: [PATCH] =?UTF-8?q?feat(runtime):=20a=C3=B1ade=20AgentState=20y=20?= =?UTF-8?q?nodos=20LangGraph=20(validate,=20llm,=20propose,=20approve,=20f?= =?UTF-8?q?inalize)?= 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/nodes.py | 193 ++++++++++++++++++++++ core/src/agentforge_core/runtime/state.py | 25 +++ tests/unit/test_runtime_nodes.py | 66 ++++++++ 3 files changed, 284 insertions(+) create mode 100644 core/src/agentforge_core/runtime/nodes.py create mode 100644 core/src/agentforge_core/runtime/state.py create mode 100644 tests/unit/test_runtime_nodes.py diff --git a/core/src/agentforge_core/runtime/nodes.py b/core/src/agentforge_core/runtime/nodes.py new file mode 100644 index 0000000..3c884e7 --- /dev/null +++ b/core/src/agentforge_core/runtime/nodes.py @@ -0,0 +1,193 @@ +"""Factory de nodos LangGraph parametrizados por engine, policy, llm, agent_def.""" + +from __future__ import annotations + +import json +import time +from collections.abc import Awaitable, Callable +from datetime import UTC, datetime +from typing import Any +from uuid import UUID + +import structlog +from langgraph.types import interrupt + +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, Message +from agentforge_core.runtime.state import AgentState + +log = structlog.get_logger(__name__) + +NodeFn = Callable[[AgentState], Awaitable[dict[str, Any]]] + + +def _now() -> datetime: + return datetime.now(UTC) + + +def _step(name: str, started: float, **detail: Any) -> dict[str, Any]: + return { + "step": name, + "timestamp": _now().isoformat(), + "duration_ms": int((time.perf_counter() - started) * 1000), + "detail": detail, + } + + +def build_node_validate_input(engine: GuardrailEngine, policy: PolicyDefinition) -> NodeFn: + async def validate_input(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + trace_id = UUID(state["trace_id"]) + violations = await engine.validate_input(state["user_input"], policy, trace_id) + any_blocked = any(v.blocked for v in violations) + new_violations = state.get("violations", []) + [ + v.model_dump(mode="json") for v in violations + ] + update: dict[str, Any] = { + "violations": new_violations, + "decision_path": [_step("validate_input", started, n_violations=len(violations))], + } + if any_blocked: + update["status"] = "blocked_by_guardrail" + return update + + return validate_input + + +def build_node_llm_reason(provider: LLMProvider, agent_def: AgentDefinition) -> NodeFn: + async def llm_reason(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + messages = [ + Message(role="system", content=agent_def.system_prompt), + Message(role="user", content=state["user_input"]), + ] + try: + result = await provider.complete( + messages=messages, + temperature=agent_def.llm.temperature, + max_tokens=agent_def.llm.max_tokens, + ) + except Exception as exc: + log.error("llm_failed", trace_id=state["trace_id"], error=str(exc)) + return { + "status": "failed", + "error": "llm_unavailable", + "decision_path": [_step("llm_reason", started, ok=False, error=str(exc))], + } + return { + "raw_llm_output": result.content, + "messages": [m.model_dump() for m in messages], + "decision_path": [ + _step( + "llm_reason", + started, + model=result.model, + tokens_in=result.tokens_in, + tokens_out=result.tokens_out, + latency_ms=result.latency_ms, + ) + ], + } + + return llm_reason + + +def build_node_validate_output(engine: GuardrailEngine, policy: PolicyDefinition) -> NodeFn: + async def validate_output(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + if state.get("status") in {"failed", "blocked_by_guardrail"}: + return {} + try: + parsed = json.loads(state.get("raw_llm_output") or "{}") + except json.JSONDecodeError as exc: + return { + "status": "failed", + "error": "output_schema_mismatch", + "decision_path": [_step("validate_output", started, ok=False, error=str(exc))], + } + trace_id = UUID(state["trace_id"]) + violations = await engine.validate_output(parsed, policy, trace_id) + any_blocked = any(v.blocked for v in violations) + new_violations = state.get("violations", []) + [ + v.model_dump(mode="json") for v in violations + ] + update: dict[str, Any] = { + "parsed_output": parsed, + "violations": new_violations, + "decision_path": [_step("validate_output", started, n_violations=len(violations))], + } + if any_blocked: + update["status"] = "blocked_by_guardrail" + return update + + return validate_output + + +def build_node_propose_actions() -> NodeFn: + async def propose_actions(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + if state.get("status") in {"failed", "blocked_by_guardrail"}: + return {} + parsed = state.get("parsed_output") or {} + actions = parsed.get("proposed_actions", []) + return { + "proposed_actions": actions, + "decision_path": [_step("propose_actions", started, n_actions=len(actions))], + } + + return propose_actions + + +def build_node_approve_gate(agent_def: AgentDefinition) -> NodeFn: + async def approve_gate(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + if state.get("status") in {"failed", "blocked_by_guardrail"}: + return {} + actions = state.get("proposed_actions", []) + risky = [ + a + for a in actions + if int(a.get("risk_score", 1)) >= agent_def.risk_threshold_for_hitl + or bool(a.get("requires_approval")) + ] + if not risky: + return {"decision_path": [_step("approve_gate", started, hitl=False)]} + # Pausa la ejecución hasta que llegue resume(human_decision={...}) + decision = interrupt({"awaiting_actions": risky}) + return { + "human_decision": decision, + "decision_path": [_step("approve_gate", started, hitl=True, resumed=True)], + } + + return approve_gate + + +def build_node_finalize() -> NodeFn: + async def finalize(state: AgentState) -> dict[str, Any]: + started = time.perf_counter() + if state.get("status") in {"failed", "blocked_by_guardrail"}: + return {"decision_path": [_step("finalize", started, skipped=True)]} + parsed = state.get("parsed_output") or {} + decision = state.get("human_decision") or {} + approved_ids = set(decision.get("approved_action_ids", [])) if decision else None + actions = state.get("proposed_actions", []) + if approved_ids is not None: + final_actions = [a for a in actions if a.get("id") in approved_ids] + if decision.get("rejected"): + return { + "status": "failed", + "error": "rejected_by_human", + "decision_path": [_step("finalize", started, rejected=True)], + } + else: + final_actions = actions + final_output = {**parsed, "approved_actions": final_actions} + return { + "final_output": final_output, + "status": "completed", + "decision_path": [_step("finalize", started, n_approved=len(final_actions))], + } + + return finalize diff --git a/core/src/agentforge_core/runtime/state.py b/core/src/agentforge_core/runtime/state.py new file mode 100644 index 0000000..cfb5c83 --- /dev/null +++ b/core/src/agentforge_core/runtime/state.py @@ -0,0 +1,25 @@ +"""Estado tipado del grafo LangGraph.""" + +from __future__ import annotations + +import operator +from typing import Annotated, Any, TypedDict + + +class AgentState(TypedDict, total=False): + """Estado mutable que fluye por el grafo; `decision_path` se acumula.""" + + trace_id: str + agent_name: str + agent_version: str + user_input: str + messages: list[dict[str, Any]] + raw_llm_output: str | None + parsed_output: dict[str, Any] | None + proposed_actions: list[dict[str, Any]] + violations: list[dict[str, Any]] + decision_path: Annotated[list[dict[str, Any]], operator.add] + status: str + error: str | None + human_decision: dict[str, Any] | None + final_output: dict[str, Any] | None diff --git a/tests/unit/test_runtime_nodes.py b/tests/unit/test_runtime_nodes.py new file mode 100644 index 0000000..a5bfb8f --- /dev/null +++ b/tests/unit/test_runtime_nodes.py @@ -0,0 +1,66 @@ +"""Tests de los nodos del grafo (lógica pura, sin compilación de grafo).""" + +from uuid import uuid4 + +import pytest + +from agentforge_core.domain.policy import PolicyDefinition, PolicyValidator +from agentforge_core.guardrails.guardrails_ai import GuardrailsAIEngine +from agentforge_core.runtime.nodes import build_node_validate_input + + +@pytest.fixture +def engine() -> GuardrailsAIEngine: + return GuardrailsAIEngine() + + +@pytest.fixture +def policy_block_email() -> PolicyDefinition: + return PolicyDefinition( + name="t", + version="v1", + description="t", + input_validators=[ + PolicyValidator( + type="detect_pii", + config={"entities": ["EMAIL_ADDRESS"], "severity_on_match": "block"}, + ) + ], + output_validators=[], + ) + + +def _base_state(user_input: str) -> dict: + return { + "trace_id": str(uuid4()), + "agent_name": "x", + "agent_version": "v1", + "user_input": user_input, + "messages": [], + "raw_llm_output": None, + "parsed_output": None, + "proposed_actions": [], + "violations": [], + "decision_path": [], + "status": "running", + "error": None, + "human_decision": None, + } + + +async def test_validate_input_marca_blocked_si_pii( + engine: GuardrailsAIEngine, policy_block_email: PolicyDefinition +) -> None: + node = build_node_validate_input(engine, policy_block_email) + out = await node(_base_state("manda correo a juan@example.com")) + assert out["status"] == "blocked_by_guardrail" + assert any(v["blocked"] for v in out["violations"]) + + +async def test_validate_input_pasa_sin_pii( + engine: GuardrailsAIEngine, policy_block_email: PolicyDefinition +) -> None: + node = build_node_validate_input(engine, policy_block_email) + out = await node(_base_state("incidente sin pii")) + # El nodo solo escribe `status` cuando bloquea; si no, el estado sigue "running". + assert out.get("status", "running") == "running"