Add pipeline flow builder UI for guardrail policies

litellm/proxy/policy_engine/pipeline_executor.py

@@ -56,7 +56,9 @@ async def execute_steps(
             PipelineExecutionResult with terminal action and step results
         """
         step_results: List[PipelineStepResult] = []
-        working_data = copy.deepcopy(data)
+        working_data = data.copy()
+        if "metadata" in working_data:
+            working_data["metadata"] = working_data["metadata"].copy()
 
         for i, step in enumerate(steps):
             start_time = time.perf_counter()
@@ -148,6 +150,12 @@ async def _run_step(
             return ("error", None, f"Guardrail '{step.guardrail}' not found")
 
         try:
+            # Inject guardrail name into metadata so should_run_guardrail() allows it
+            if "metadata" not in data:
+                data["metadata"] = {}
+            original_guardrails = data["metadata"].get("guardrails")
+            data["metadata"]["guardrails"] = [step.guardrail]
+
             # Use unified_guardrail path if callback implements apply_guardrail
             target = callback
             use_unified = "apply_guardrail" in type(callback).__dict__

litellm/proxy/policy_engine/policy_endpoints.py

@@ -10,8 +10,11 @@
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
 from litellm.proxy.policy_engine.attachment_registry import get_attachment_registry
+from litellm.proxy.policy_engine.pipeline_executor import PipelineExecutor
 from litellm.proxy.policy_engine.policy_registry import get_policy_registry
 from litellm.types.proxy.policy_engine import (
+    GuardrailPipeline,
+    PipelineTestRequest,
     PolicyAttachmentCreateRequest,
     PolicyAttachmentDBResponse,
     PolicyAttachmentListResponse,
@@ -349,6 +352,69 @@ async def get_resolved_guardrails(policy_id: str):
         raise HTTPException(status_code=500, detail=str(e))
 
 
+# ─────────────────────────────────────────────────────────────────────────────
+# Pipeline Test Endpoint
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@router.post(
+    "/policies/test-pipeline",
+    tags=["Policies"],
+    dependencies=[Depends(user_api_key_auth)],
+)
+async def test_pipeline(
+    request: PipelineTestRequest,
+    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
+):
+    """
+    Test a guardrail pipeline with sample messages.
+
+    Executes the pipeline steps against the provided test messages and returns
+    step-by-step results showing which guardrails passed/failed, actions taken,
+    and timing information.
+
+    Example Request:
+    ```bash
+    curl -X POST "http://localhost:4000/policies/test-pipeline" \\
+        -H "Authorization: Bearer <your_api_key>" \\
+        -H "Content-Type: application/json" \\
+        -d '{
+            "pipeline": {
+                "mode": "pre_call",
+                "steps": [
+                    {"guardrail": "pii-guard", "on_pass": "next", "on_fail": "block"}
+                ]
+            },
+            "test_messages": [{"role": "user", "content": "My SSN is 123-45-6789"}]
+        }'
+    ```
+    """
+    try:
+        validated_pipeline = GuardrailPipeline(**request.pipeline)
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=f"Invalid pipeline: {e}")
+
+    data = {
+        "messages": request.test_messages,
+        "model": "test",
+        "metadata": {},
+    }
+
+    try:
+        result = await PipelineExecutor.execute_steps(
+            steps=validated_pipeline.steps,
+            mode=validated_pipeline.mode,
+            data=data,
+            user_api_key_dict=user_api_key_dict,
+            call_type="completion",
+            policy_name="test-pipeline",
+        )
+        return result.model_dump()
+    except Exception as e:
+        verbose_proxy_logger.exception(f"Error testing pipeline: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
 # ─────────────────────────────────────────────────────────────────────────────
 # Policy Attachment CRUD Endpoints
 # ─────────────────────────────────────────────────────────────────────────────

litellm/proxy/policy_engine/policy_registry.py

@@ -10,6 +10,8 @@
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
+from prisma import Json as PrismaJson
+
 from litellm._logging import verbose_proxy_logger
 from litellm.types.proxy.policy_engine import (
     GuardrailPipeline,
@@ -248,7 +250,10 @@ async def add_policy_to_db(
                 data["created_by"] = created_by
                 data["updated_by"] = created_by
             if policy_request.condition is not None:
-                data["condition"] = policy_request.condition.model_dump()
+                data["condition"] = PrismaJson(policy_request.condition.model_dump())
+            if policy_request.pipeline is not None:
+                validated_pipeline = GuardrailPipeline(**policy_request.pipeline)
+                data["pipeline"] = PrismaJson(validated_pipeline.model_dump())
 
             created_policy = await prisma_client.db.litellm_policytable.create(
                 data=data
@@ -267,6 +272,7 @@ async def add_policy_to_db(
                     "condition": policy_request.condition.model_dump()
                     if policy_request.condition
                     else None,
+                    "pipeline": policy_request.pipeline,
                 },
             )
             self.add_policy(policy_request.policy_name, policy)
@@ -279,6 +285,7 @@ async def add_policy_to_db(
                 guardrails_add=created_policy.guardrails_add or [],
                 guardrails_remove=created_policy.guardrails_remove or [],
                 condition=created_policy.condition,
+                pipeline=created_policy.pipeline,
                 created_at=created_policy.created_at,
                 updated_at=created_policy.updated_at,
                 created_by=created_policy.created_by,
@@ -325,7 +332,10 @@ async def update_policy_in_db(
             if policy_request.guardrails_remove is not None:
                 update_data["guardrails_remove"] = policy_request.guardrails_remove
             if policy_request.condition is not None:
-                update_data["condition"] = policy_request.condition.model_dump()
+                update_data["condition"] = PrismaJson(policy_request.condition.model_dump())
+            if policy_request.pipeline is not None:
+                validated_pipeline = GuardrailPipeline(**policy_request.pipeline)
+                update_data["pipeline"] = PrismaJson(validated_pipeline.model_dump())
 
             updated_policy = await prisma_client.db.litellm_policytable.update(
                 where={"policy_id": policy_id},
@@ -343,6 +353,7 @@ async def update_policy_in_db(
                         "remove": updated_policy.guardrails_remove,
                     },
                     "condition": updated_policy.condition,
+                    "pipeline": updated_policy.pipeline,
                 },
             )
             self.add_policy(updated_policy.policy_name, policy)
@@ -355,6 +366,7 @@ async def update_policy_in_db(
                 guardrails_add=updated_policy.guardrails_add or [],
                 guardrails_remove=updated_policy.guardrails_remove or [],
                 condition=updated_policy.condition,
+                pipeline=updated_policy.pipeline,
                 created_at=updated_policy.created_at,
                 updated_at=updated_policy.updated_at,
                 created_by=updated_policy.created_by,
@@ -432,6 +444,7 @@ async def get_policy_by_id_from_db(
                 guardrails_add=policy.guardrails_add or [],
                 guardrails_remove=policy.guardrails_remove or [],
                 condition=policy.condition,
+                pipeline=policy.pipeline,
                 created_at=policy.created_at,
                 updated_at=policy.updated_at,
                 created_by=policy.created_by,
@@ -468,6 +481,7 @@ async def get_all_policies_from_db(
                     guardrails_add=p.guardrails_add or [],
                     guardrails_remove=p.guardrails_remove or [],
                     condition=p.condition,
+                    pipeline=p.pipeline,
                     created_at=p.created_at,
                     updated_at=p.updated_at,
                     created_by=p.created_by,
@@ -503,6 +517,7 @@ async def sync_policies_from_db(
                             "remove": policy_response.guardrails_remove,
                         },
                         "condition": policy_response.condition,
+                        "pipeline": policy_response.pipeline,
                     },
                 )
                 self.add_policy(policy_response.policy_name, policy)
@@ -551,6 +566,7 @@ async def resolve_guardrails_from_db(
                             "remove": policy_response.guardrails_remove,
                         },
                         "condition": policy_response.condition,
+                        "pipeline": policy_response.pipeline,
                     },
                 )
                 temp_policies[policy_response.policy_name] = policy

litellm/proxy/schema.prisma

@@ -917,6 +917,7 @@ model LiteLLM_PolicyTable {
   guardrails_add   String[]  @default([])
   guardrails_remove String[] @default([])
   condition        Json?     @default("{}")  // Policy conditions (e.g., model matching)
+  pipeline         Json?     // Optional guardrail pipeline (mode + steps[])
   created_at       DateTime  @default(now())
   created_by       String?
   updated_at       DateTime  @default(now()) @updatedAt

litellm/proxy/utils.py

@@ -1222,10 +1222,7 @@ def _handle_pipeline_result(
                     },
                 }
             }
-            if HTTPException is not None:
-                raise HTTPException(status_code=400, detail=error_detail)
-            else:
-                raise Exception(str(error_detail))
+            raise HTTPException(status_code=400, detail=error_detail)
 
         if result.terminal_action == "modify_response":
             raise ModifyResponseException(
@@ -1236,9 +1233,6 @@ def _handle_pipeline_result(
                 detection_info=None,
             )
 
-        verbose_proxy_logger.warning(
-            f"Pipeline '{policy_name}': unrecognized terminal_action '{result.terminal_action}', defaulting to allow"
-        )
         return data
 
     # The actual implementation of the function

litellm/types/proxy/policy_engine/__init__.py

@@ -26,6 +26,7 @@
 )
 from litellm.types.proxy.policy_engine.resolver_types import (
     AttachmentImpactResponse,
+    PipelineTestRequest,
     PolicyAttachmentCreateRequest,
     PolicyAttachmentDBResponse,
     PolicyAttachmentListResponse,
@@ -90,6 +91,8 @@
     "PolicyAttachmentCreateRequest",
     "PolicyAttachmentDBResponse",
     "PolicyAttachmentListResponse",
+    # Pipeline test types
+    "PipelineTestRequest",
     # Resolve types
     "PolicyResolveRequest",
     "PolicyResolveResponse",

litellm/types/proxy/policy_engine/resolver_types.py

@@ -154,6 +154,10 @@ class PolicyCreateRequest(BaseModel):
         default=None,
         description="Condition for when this policy applies.",
     )
+    pipeline: Optional[Dict[str, Any]] = Field(
+        default=None,
+        description="Optional guardrail pipeline for ordered execution. Contains 'mode' and 'steps'.",
+    )
 
 
 class PolicyUpdateRequest(BaseModel):
@@ -183,6 +187,10 @@ class PolicyUpdateRequest(BaseModel):
         default=None,
         description="Condition for when this policy applies.",
     )
+    pipeline: Optional[Dict[str, Any]] = Field(
+        default=None,
+        description="Optional guardrail pipeline for ordered execution. Contains 'mode' and 'steps'.",
+    )
 
 
 class PolicyDBResponse(BaseModel):
@@ -201,6 +209,9 @@ class PolicyDBResponse(BaseModel):
     condition: Optional[Dict[str, Any]] = Field(
         default=None, description="Policy condition."
     )
+    pipeline: Optional[Dict[str, Any]] = Field(
+        default=None, description="Optional guardrail pipeline."
+    )
     created_at: Optional[datetime] = Field(
         default=None, description="When the policy was created."
     )
@@ -291,6 +302,17 @@ class PolicyAttachmentListResponse(BaseModel):
 # ─────────────────────────────────────────────────────────────────────────────
 
 
+class PipelineTestRequest(BaseModel):
+    """Request body for testing a guardrail pipeline with sample messages."""
+
+    pipeline: Dict[str, Any] = Field(
+        description="Pipeline definition with 'mode' and 'steps'.",
+    )
+    test_messages: List[Dict[str, str]] = Field(
+        description="Test messages to run through the pipeline, e.g. [{'role': 'user', 'content': '...'}].",
+    )
+
+
 class PolicyResolveRequest(BaseModel):
     """Request body for resolving effective policies/guardrails for a context."""