🛡️ Forja — Walkthrough

00 Qué es y por qué

Poner agentes de IA en producción sin una capa de gobierno produce sistemas opacos: prompts que cambian sin historial, validaciones inconsistentes, acciones de alto impacto sin supervisión y ninguna auditoría de lo que decidió el agente.

Forja es el "plano de control" que pones delante de tus agentes antes de dejarlos tocar nada importante. No es un framework para construir agentes; es la capa que los cataloga, versiona, valida, ejecuta de forma supervisada y audita.

Este documento va de lo general a lo concreto: primero las ideas y la forma del sistema, luego cada módulo y sus interrelaciones, y por último el cableado de bajo nivel (grafo, persistencia, API). Si solo quieres arrancarlo, ve al README.md; si quieres la narrativa completa, docs/explicacion.md; si quieres firmas exactas y cadenas de llamada, docs/componentes.md.

01 Las seis ideas grandes

Si entiendes estas seis ideas, entiendes el proyecto. Todo lo demás son detalles de implementación.

02 Vista de pájaro: dos servicios

Forja son dos procesos que se hablan por HTTP/JSON, levantados por docker-compose:

• forja-core (FastAPI, puerto 8000) — todo el dominio: registry de agentes, motor de guardrails, runtime de ejecución, persistencia. No tiene UI.
• forja-dashboard (Streamlit, puerto 8501) — una consola visual con cinco páginas. No contiene lógica de negocio: es un cliente HTTP del core.

La separación importa: el core podría servir a una CLI, a otro servicio, a un pipeline… el dashboard es solo una de las caras posibles. (Y este documento es otra cara más: el HTML que estás leyendo.)

03 Las capas del core, de fuera hacia dentro

Una petición HTTP atraviesa estas capas en orden. Cada una solo conoce a la de dentro; nada de dentro conoce a las de fuera (eso es lo que mantiene el sistema desacoplado y testeable).

04 El grafo de módulos (quién depende de quién)

El código del core vive bajo core/src/forja_core/. Es un DAG: las capas de abajo no importan nada de las de arriba. El "nivel" es la profundidad topológica. Lee de abajo hacia arriba: el vocabulario primero, la composición de la app al final.

El paquete dashboard/ no aparece aquí: no comparte código con el core, solo lo llama por HTTP.

05 El patrón que se repite: Protocol + factory + Settings

Tres veces (LLM, guardrails, registry) verás la misma estructura. Permite cambiar el comportamiento sin tocar el código: pones LLM_PROVIDER=azure en .env y el factory te da el provider de Azure; pones mock y tienes un demo determinista sin claves. Es el principio de "configuración antes que código".

06 El dominio — el vocabulario del sistema

Modelos Pydantic puros bajo domain/. No dependen de nada del proyecto; todo lo demás depende de ellos. Si quieres entender el sistema rápido, empieza leyendo domain/execution.py: te dice exactamente qué información se produce y se guarda.

07 Guardrails — el motor de validación

Aquí está el corazón del "gobierno". La estructura sigue otra vez Protocol + implementaciones + composite + factory. Una política lista qué validar; el motor lo aplica. Por defecto, fail-closed: si un validador peta, cuenta como bloqueo (seguridad antes que disponibilidad).

Los validadores concretos (guardrails/validators.py)

Así se conecta una política con un validador (extracto de policies/default/versions/v1.yaml):

name: default
version: v1
input_validators:
  - type: detect_pii
    config: { entities: [EMAIL_ADDRESS, ES_NIF, IP_ADDRESS, IBAN_CODE], severity_on_match: block }
  - type: prompt_injection
    config: { severity_on_match: block }
  - type: toxic_language
    config: { threshold: 0.7, severity_on_match: warning }
  - type: forbidden_topics
    config: { topics: ["instrucciones de explotación", "credenciales", "código malicioso"], severity_on_match: block }
output_validators:
  - type: schema_match          # valida el JSON del LLM contra un JSON Schema
    config: { severity_on_mismatch: block, schema: { … } }
  - type: pii_leakage
  - type: forbidden_action_keywords
    config: { keywords: ["DROP TABLE", "rm -rf", "shutdown -h now", "delete production", "format c:"] }
  - type: telco_safety_rules
    config: { rules: [never_propose_action_targeting_production_without_rollback, never_propose_mass_action_without_canary] }
on_validator_error: fail_closed   # si un validador lanza una excepción → cuenta como bloqueo

08 El grafo de ejecución (LangGraph)

Una ejecución del agente se modela como un grafo dirigido de estados con checkpoints. No es "llama al LLM y ya": es un flujo con aristas condicionales y una pausa real para Human-in-the-Loop. Cada nodo añade un DecisionStep con su duración al decision_path.

Los nodos (runtime/nodes.py) y el estado (runtime/state.py)

El estado que fluye por el grafo es un TypedDict. decision_path usa un reducer (Annotated[list, operator.add]) para que cada nodo añada pasos en vez de sobrescribir:

class AgentState(TypedDict, total=False):
    trace_id: str; agent_name: str; agent_version: str; user_input: str
    messages: list[dict]; raw_llm_output: str | None; parsed_output: dict | None
    proposed_actions: list[dict]; violations: list[dict]
    decision_path: Annotated[list[dict], operator.add]   # ← reducer: cada nodo AÑADE pasos
    status: str; error: str | None
    human_decision: dict | None; final_output: dict | None

g.add_edge(START, "validate_input")

def _after_validate_input(state):
    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"})

# … _after_llm → END si status=="failed"  ·  _after_validate_output → END si status in {blocked, failed} …
g.add_edge("propose_actions", "approve_gate")
g.add_edge("approve_gate", "finalize")
g.add_edge("finalize", END)
return g.compile(checkpointer=checkpointer)

def build_node_approve_gate(agent_def):
    async def approve_gate(state):
        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)]}
        decision = interrupt({"awaiting_actions": risky})   # ← pausa; reanuda con Command(resume=decision)
        return {"human_decision": decision,
                "decision_path": [_step("approve_gate", started, hitl=True, resumed=True)]}
    return approve_gate

09 El viaje de UNA petición, de principio a fin

Esta es la sección que conviene leer despacio: aquí se ve cómo encaja todo. Seguimos POST /agents/incident_analyzer/invoke con el escenario SIP — el que dispara HITL — y luego la aprobación.

El mismo recorrido, con curl (provider mock, sin claves):

# 1) Invocar — el escenario SIP dispara HITL
TRACE=$(curl -s localhost:8000/agents/incident_analyzer/invoke \
  -H 'content-type: application/json' \
  -d '{"input": "Caída de registros SIP tras desplegar la imagen 4.7.2 en CSCF aravaca-01"}' \
  | python -c 'import sys,json; print(json.load(sys.stdin)["trace_id"])')
# → status: "awaiting_approval", needs_human_for: [act-1]

# 2) Revisar la cola HITL
curl -s localhost:8000/executions | python -m json.tool   # incluye las awaiting_approval reconstruidas del índice + checkpointer

# 3) Aprobar la acción segura act-1
curl -s localhost:8000/executions/$TRACE/approve \
  -H 'content-type: application/json' \
  -d '{"approved_action_ids": ["act-1"], "comment": "rollback ok"}'
# → status: "completed", final_output: { ..., "approved_actions": [act-1] }   ·   ya escrito en executions.jsonl

# (el dashboard hace exactamente esto desde las páginas "Ejecutar" y "Aprobaciones")

10 Versionado tipo Git

Un agente no es una fila en una BD: es una carpeta con un index.yaml (el "catálogo de commits") y versions/v1.yaml, versions/v2.yaml… (los "commits"). Cada versión tiene un hash SHA-256 del contenido normalizado, y dos versiones son comparables con un diff unificado (difflib). Lo mismo para las políticas. Cambias un YAML → nueva versión; promueves cambiando active_version. Sin redeploy.

name: incident_analyzer
versions:
  - id: v1
    hash: pending
    author: Juan
    message: Versión inicial; SIP/IMS y MOS básicos.
    created_at: 2026-04-12T10:00:00Z
  - id: v2
    hash: pending
    author: Juan
    message: Añade codec mismatch y refuerza prompts.
    created_at: 2026-05-01T12:00:00Z
active_version: v2     # ← invoke usa esta si no pides otra

name: incident_analyzer
version: v2
owner: Juan
state: active                # draft | active | deprecated
guardrails: [default]         # nombres de política
llm:
  provider: mock
  model: gpt-4o
  temperature: 0.1
  max_tokens: 2000
system_prompt: |
  Eres un analista senior de operaciones de plataforma de voz
  virtualizada (IMS, CSCF, SBC, HSS). Devuelves SIEMPRE un JSON
  con severity, root_cause_hypothesis y proposed_actions. …
output_schema:
  type: object
  required: [severity, root_cause_hypothesis, proposed_actions]
  …
risk_threshold_for_hitl: 4    # risk_score ≥ 4 (o requires_approval) → HITL

11 Persistencia: cuatro formas, cuatro razones

Forja no usa una sola base de datos; usa la herramienta adecuada para cada cosa.

12 Trazabilidad: el hilo del trace_id

Cada petición lleva un trace_id (UUID) que recorre todo el sistema. structlog emite JSON, una línea por evento, con ese trace_id en el contexto. Y cada nodo del grafo deja un DecisionStep en el decision_path con su duration_ms — la "caja negra" de la ejecución.

Detalles del logging: observability/logging.py configura structlog con JSONRenderer, TimeStamper(iso), niveles, y merge_contextvars. bind_trace_id / clear_trace_id manejan el contexto por request.

13 Estados de una ejecución

El status de un AgentExecution es uno de cinco. Tres son terminales (completed, failed, blocked_by_guardrail); awaiting_approval es el único que persiste indefinidamente esperando a un humano.

14 Referencia de la API REST

El core expone una API limpia. Todos los endpoints pasan por TraceIdMiddleware (lee/crea X-Trace-Id, lo devuelve en la respuesta). Los response_model son los modelos del dominio. Es exactamente lo que llama el dashboard (vía CoreClient).

Helpers internos de executions.py alrededor del índice (data/execution_index.json): _record_execution (en cada invoke), _resolve (reconstruye agent_def/policy; 404/500), _ensure_awaiting (404/409).

15 El agente de ejemplo: incident_analyzer

El repo trae un agente de operaciones de telco y un proveedor LLM mock determinista para que el demo funcione out-of-the-box, sin API keys ni red. MockProvider mira el texto de entrada y elige una de tres respuestas canónicas buscando subcadenas ("sip", "mos", "hss", en ese orden); si no encuentra ninguna, una respuesta genérica (o, en tests con texto aleatorio, un fallback por hash). Cada respuesta es un JSON con severity, root_cause_hypothesis y proposed_actions.

Lo que no es código: agents/incident_analyzer/ (el YAML del agente + examples/*.txt), policies/default/ (la política), y data/ (estado runtime gitignored: checkpoints.sqlite, executions.jsonl, violations.jsonl, execution_index.json — se crea sola). Para usar Azure OpenAI o OpenAI reales: edita .env (LLM_PROVIDER=azure|openai + las claves). Los providers reales tienen retry exponencial 1s/2s/4s.

16 El dashboard (Streamlit)

Una consola visual con cinco páginas. No contiene lógica de negocio: client.py (CoreClient, httpx síncrono con 2 retries) envuelve todos los endpoints del core y mapea 404/409/422 a {"api_error": …} para que las páginas puedan distinguir "la API rechazó la petición" de "ejecución sin error". app.py es la raíz: sidebar de branding + health-check del core. Nadie del core depende del dashboard.

Componentes reutilizables de UI: components/diff_view (pinta +/-/@@), components/trace_view (el timeline del decision_path), components/violation_view (badges de severidad). El dashboard no tiene tests unitarios (mal coste/beneficio para Streamlit); su verificación es la checklist manual de docs/manual_qa.md. Demo guiada en 3 pasos: Registro → comparar v1 vs v2 · Ejecutar → escenario SIP → pausa en HITL · Aprobaciones → aprobar las seguras → ejecución completa.

17 Arranque y configuración

Out-of-the-box, sin claves: cp .env.example .env y docker compose up → dashboard en localhost:8501 y API en localhost:8000. (Este documento se consulta abriendo docs/walkthrough.html en el navegador — no requiere servidor.)

18 Tests

Hay una suite de unit + integration (~90 tests). make test corre solo los unit; make test-all añade los de integración; make lint es ruff + mypy (estricto); make smoke levanta docker compose y hace curl /health.

• tests/unit/ — dominio (agent, execution, guardrail, policy), config, llm (base, mock, factory), guardrails (ai, composite), registry (repository, policy_store, versioning, factory), runtime (state-graph, nodes, orchestrator), observability/logging, y los routers de la API (agents, executions, policies/violations, health) — estos usan tests/fixtures/ (versiones mínimas) y hacen deps.*.cache_clear() tras apuntar DATA_DIR a un tmp_path.
• tests/integration/ — montan la app con TestClient contra los assets reales del repo (agents/, policies/): test_invoke_happy_path, test_invoke_hitl (el ciclo completo en ~40 líneas — el mejor sitio para entenderlo), test_invoke_pii_block, test_invoke_resume_after_restart (prueba que un awaiting_approval sobrevive a recrear el orchestrator).
• El dashboard Streamlit no se prueba con unit tests; se verifica con la checklist de docs/manual_qa.md.

19 Este documento — un HTML autocontenido

Lo que estás leyendo es un único fichero HTML autocontenido — docs/walkthrough.html en el repositorio. Todo está embebido: los estilos, los nueve diagramas SVG, la navegación con scrollspy, el tema claro/oscuro y los bloques de código. No depende de ninguna red ni de ningún CDN, y no necesita que ningún servidor esté corriendo: se abre directamente en cualquier navegador.

xdg-open docs/walkthrough.html        # Linux
open docs/walkthrough.html            # macOS
# o arrástralo al navegador · o file://<ruta-del-repo>/docs/walkthrough.html

20 Glosario

21 Lo que aún no hace (a propósito)

Es un MVP. En el roadmap (docs/futuro.md): NeMo Guardrails completo (Colang + KB embedding + dialog rails) · autenticación (OAuth2/OIDC, MSAL para Azure AD) · OpenTelemetry (spans por nodo, métricas de violaciones por validador) · multi-tenant (tenant_id aislado) · evaluadores LLM-as-judge (regression suite sobre escenarios canónicos, mide deriva) · UI de aprobación con SLA (cola Kanban, reasignación, alertas) · persistencia migrable a Postgres + pgvector · esquema de prompts mejorado (variables tipadas tipo Jinja, valores por entorno) · promoción explícita draft → active vía PR/aprobación. La detección de PII más fina (modelos grandes, español), también pendiente.

22 Por dónde empezar a leer (ruta de ~1 hora)

1. domain/execution.py y domain/agent.py — qué datos hay. (§06)
2. policies/default/versions/v1.yaml y agents/incident_analyzer/versions/v2.yaml — cómo se declara todo. (§07, §10)
3. runtime/nodes.py y runtime/graph.py — el flujo de ejecución. (§08)
4. api/executions.py — cómo se expone (invoke / approve / reject). (§14)
5. tests/integration/test_invoke_hitl.py — el ciclo completo, en ~40 líneas. (§18)
6. Levanta docker compose up, haz clic por las cinco páginas del dashboard (§16), y abre docs/walkthrough.html en el navegador (§19).

23 Ruta de codificación manual — de cero a sistema funcionando

Si quieres entender el proyecto escribiéndolo, este es el orden que maximiza la comprensión. La lógica: sigue el grafo de dependencias hacia afuera desde el centro — empieza por lo que no depende de nada, termina por lo que depende de todo.

Fase 1 — El vocabulario: domain/ (~30 min)

Estos cuatro ficheros no importan nada del proyecto — solo Pydantic. Son el vocabulario de todo lo demás. Escribirlos primero te obliga a responder "¿qué maneja este sistema?" antes de pensar en cómo lo procesa.

1. domain/agent.py — AgentDefinition, LLMConfig, AgentVersionMeta. Al terminar sabrás exactamente qué es "un agente" en este sistema: una declaración Pydantic, no código Python.
2. domain/guardrail.py — GuardrailViolation. Qué devuelve un guardrail cuando detecta un problema: una violación tipada, no una excepción.
3. domain/policy.py — PolicyDefinition. La lista de reglas que rigen un agente concreto.
4. domain/execution.py — el modelo más rico: AgentExecution, ProposedAction, ExecutionStatus, DecisionStep. Contiene el estado completo de una ejecución. Fíjate en que AgentExecution tiene definición del agente separada de su instancia en runtime.

Fase 2 — Las costuras: los dos Protocols (~15 min)

Los Protocols son las interfaces estructurales del sistema — el contrato que separa "qué hace" de "cómo lo hace". El grafo solo importa estos dos ficheros, nunca OpenAI ni guardrails-ai directamente.

1. llm/base.py — LLMProvider (un método async complete()), Message, CompletionResult. Dos clases Pydantic y un Protocol de cuatro líneas.
2. guardrails/base.py — GuardrailEngine: dos métodos (validate_input y validate_output), ambos devuelven list[GuardrailViolation].

Fase 3 — Primera ejecución posible: mocks + YAML del agente (~45 min)

Con dominios e interfaces ya puedes ejecutar el sistema sin ninguna clave de API. Los mocks son tu andamio.

1. llm/mock.py — implementa LLMProvider devolviendo un JSON hardcodeado. ~30 líneas. Permite correr el grafo completo sin tocar OpenAI ni Azure.
2. guardrails/validators.py — validadores keyword-based simples. Implementan GuardrailEngine sin dependencias externas.
3. El primer YAML del agente: agents/incident_analyzer/versions/v1.yaml. Escribe la definición declarativa (nombre, owner, purpose, guardrails, config LLM, system_prompt, output_schema).

Fase 4 — El corazón del sistema: estado → nodos → grafo (~1.5 h)

Esta es la parte más densa e importante. El grafo LangGraph es la máquina de estados que orquesta todo. El orden dentro de la fase también importa.

1. runtime/state.py primero — AgentState: un TypedDict con todos los campos del estado mutable que fluye por el grafo. El detalle a entender: decision_path: Annotated[list, operator.add] — no se sobreescribe, se acumula; LangGraph llama al reducer automáticamente cada vez que un nodo devuelve un decision_path.
2. runtime/nodes.py — los seis nodos, en el orden en que aparecerán en el grafo:
   • build_node_validate_input — llama al engine, acumula violations, decide si poner blocked_by_guardrail.
   • build_node_llm_reason — llama al provider, captura el fallo y lo transforma en status="failed".
   • build_node_validate_output — parsea el JSON del LLM y valida el output contra la política.
   • build_node_propose_actions — extrae proposed_actions del output parseado.
   • build_node_approve_gate — el nodo HITL: llama a interrupt({"awaiting_actions": risky}) si hay acciones con riesgo ≥ threshold. La ejecución se pausa aquí hasta que llegue un Command(resume=...).
   • build_node_finalize — filtra acciones por las aprobadas, construye final_output.

   Nota el patrón: cada builder captura sus dependencias por clausura y devuelve una función async (state) → dict. Las dependencias no viajan en el estado — están fijas en el cierre.

3. runtime/graph.py — conecta los nodos. Los tres routers condicionales (_after_validate_input, _after_llm, _after_validate_output) hacen que cualquier error cortocircuite hacia END sin pasar por nodos posteriores.

Fase 5 — Orquestador y checkpointer (~30 min)

1. runtime/checkpointer.py — configura SQLite como backend de checkpoints LangGraph. Sin esto, interrupt() no puede persistir el estado entre la llamada a invoke y la llamada a approve.
2. runtime/orchestrator.py — el objeto que la API usará. Recibe AgentDefinition + PolicyDefinition + providers ya construidos, compila el grafo y expone dos métodos: invoke() (nueva ejecución) y approve() (reanudar tras HITL).

Punto de control: en este momento tienes un sistema funcionando de extremo a extremo. Puedes llamar directamente al orquestador en un test y observar el decision_path completo, sin API ni dashboard.

Fase 6 — La API: de adentro hacia afuera (~1 h)

Dentro de api/, el orden también importa.

1. api/deps.py — el armario de cableado: los @lru_cache que construyen el registry, el policy store, el LLM provider y el guardrail engine. Escríbelo después de tener todo lo anterior — solo entonces sabrás exactamente qué estás conectando.
2. api/executions.py — los tres endpoints críticos: POST /executions/invoke, POST /executions/{trace_id}/approve, GET /executions/{trace_id}.
3. api/agents.py — CRUD del registry (listar agentes, leer versión concreta).
4. api/middlewares.py — inyecta X-Trace-Id y bind-ea el contexto de structlog.
5. main.py — monta los routers, configura CORS, arranca la app FastAPI.

Fase 7 — Registry y versionado (~45 min)

Si has estado usando el mock registry, ahora añades la persistencia real basada en YAML.

1. registry/repository.py — carga definiciones de agentes desde YAML, cachea en memoria.
2. registry/versioning.py — hash SHA-256 + index.yaml: el sistema de versionado tipo-Git.
3. registry/policy_store.py — análogo al repository, pero para políticas.
4. registry/factory.py — el factory que elige qué implementación de registry usar según Settings.

Fase 8 — El dashboard (~1.5 h)

El dashboard es un cliente HTTP del core — puedes empezarlo en cualquier momento después de tener los endpoints.

1. dashboard/client.py — primero. Te fuerza a pensar en qué necesita la UI antes de diseñar las páginas. Es el contrato del core visto desde el consumidor.
2. Las páginas por complejidad creciente:
   • pages/1_Registro.py — solo lectura: lista agentes y versiones.
   • pages/5_Politicas.py — solo lectura: muestra políticas activas.
   • pages/4_Historial.py — lectura + filtros: lista ejecuciones pasadas.
   • pages/2_Ejecutar.py — escritura: invoke + polling de estado + visualización del decision_path.
   • pages/3_Aprobaciones.py — la más compleja: muestra acciones pendientes, permite aprobar o rechazar individual o masivamente.