diff --git a/nemoguardrails/llm/clients/_errors.py b/nemoguardrails/llm/clients/_errors.py index 988e875fd0..0d0d8de651 100644 --- a/nemoguardrails/llm/clients/_errors.py +++ b/nemoguardrails/llm/clients/_errors.py @@ -162,16 +162,19 @@ def build_error_payload( # looks like an OpenAI error object cannot truncate a stream or skip output # rails. ``downstream_error`` carries an upstream HTTP status; # ``generation_error`` is a status-less generation failure; -# ``guardrails_violation`` is emitted when a rail blocks. +# ``guardrails_violation`` is emitted when a rail blocks; +# ``internal_error`` is emitted when a rail fails to execute. GENERATION_ERROR_TYPE = "generation_error" DOWNSTREAM_ERROR_TYPE = "downstream_error" GUARDRAILS_VIOLATION_TYPE = "guardrails_violation" +INTERNAL_ERROR_TYPE = "internal_error" STREAM_ERROR_TYPES = frozenset( { GENERATION_ERROR_TYPE, DOWNSTREAM_ERROR_TYPE, GUARDRAILS_VIOLATION_TYPE, + INTERNAL_ERROR_TYPE, } ) diff --git a/nemoguardrails/server/api.py b/nemoguardrails/server/api.py index 9aefe4b732..3f1c4fa610 100644 --- a/nemoguardrails/server/api.py +++ b/nemoguardrails/server/api.py @@ -23,13 +23,13 @@ import time import uuid from contextlib import asynccontextmanager -from typing import Any, AsyncIterator, Callable, List, Literal, Optional, Union +from typing import Any, AsyncIterator, Callable, List, Optional, Union import httpx from fastapi import FastAPI, HTTPException, Request from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware -from pydantic import BaseModel, ValidationError +from pydantic import BaseModel, ValidationError, field_validator from starlette.exceptions import HTTPException as StarletteHTTPException from starlette.middleware.exceptions import ExceptionMiddleware from starlette.responses import JSONResponse, RedirectResponse, StreamingResponse @@ -53,6 +53,7 @@ from nemoguardrails.http.errors import HTTPClientError from nemoguardrails.llm.call import _prepend_think_tags from nemoguardrails.llm.clients._errors import ( + STREAM_ERROR_TYPES, build_error_payload, build_streaming_error_payload, normalize_error_status, @@ -539,10 +540,17 @@ class ChunkErrorMetadata(BaseModel): message: str # Only the internal stream markers count as a terminal error frame; see # ChunkError below. - type: Literal["generation_error", "downstream_error", "guardrails_violation"] + type: str param: Optional[str] = None code: Union[str, int, None] = None + @field_validator("type") + @classmethod + def validate_error_type(cls, value: str) -> str: + if value not in STREAM_ERROR_TYPES: + raise ValueError("Not an internal stream error type") + return value + class ChunkError(BaseModel): """A terminal error frame pushed into the stream by the guardrails runtime. diff --git a/tests/guardrails/test_iorails_streaming.py b/tests/guardrails/test_iorails_streaming.py index b89668383c..63ceb148b2 100644 --- a/tests/guardrails/test_iorails_streaming.py +++ b/tests/guardrails/test_iorails_streaming.py @@ -990,6 +990,12 @@ def test_error_json_string(self): def test_metadata_frame_with_error(self): assert _is_stream_error_chunk({"text": '{"error": {"type": "generation_error"}}'}) is True + @pytest.mark.parametrize("include_metadata", [False, True]) + def test_internal_rail_error_is_terminal(self, include_metadata): + error = '{"error": {"type": "internal_error", "code": "rail_execution_failure"}}' + chunk = {"text": error} if include_metadata else error + assert _is_stream_error_chunk(chunk) is True + def test_non_error_json_string(self): assert _is_stream_error_chunk('{"tool_calls": []}') is False diff --git a/tests/server/test_api.py b/tests/server/test_api.py index 1b805db595..22f5cc051c 100644 --- a/tests/server/test_api.py +++ b/tests/server/test_api.py @@ -25,7 +25,8 @@ pytest.importorskip("openai", reason="openai is required for server tests") from fastapi.testclient import TestClient -from nemoguardrails import RailsConfig +from nemoguardrails import Guardrails, RailsConfig +from nemoguardrails.actions import action from nemoguardrails.exceptions import InvalidModelConfigurationError from nemoguardrails.guardrails.model_engine import ModelEngine from nemoguardrails.llm.models.openai_chat import OpenAIChatModel @@ -1412,6 +1413,97 @@ async def test_openai_sse_format_basic_chunks(): assert done == "data: [DONE]\n\n" +@pytest.mark.asyncio +@pytest.mark.parametrize("as_dict", [False, True]) +async def test_streaming_rail_execution_failure_is_terminal(as_dict): + error = { + "error": { + "message": "Internal error in failing safety check rail: action failed", + "type": "internal_error", + "param": "failing safety check", + "code": "rail_execution_failure", + } + } + stream = _create_test_stream([error if as_dict else json.dumps(error), "Unchecked content"]) + + frames = [frame async for frame in _format_streaming_response(stream, model_name="test-model")] + + assert len(frames) == 2 + assert json.loads(frames[0][6:]) == error + assert frames[1] == "data: [DONE]\n\n" + + +@pytest.mark.parametrize("use_iorails", [False, True], ids=["legacy", "iorails-fallback"]) +@pytest.mark.parametrize("parallel", [False, True], ids=["sequential", "parallel"]) +@pytest.mark.parametrize("invalid_outcome", [False, True], ids=["action-failure", "invalid-outcome"]) +@pytest.mark.parametrize("stream_first", [False, True], ids=["rails-first", "stream-first"]) +def test_streaming_output_rail_failure_returns_error_frame( + monkeypatch, use_iorails, parallel, invalid_outcome, stream_first +): + @action(is_system_action=True) + def failing_rail_action(**params): + if invalid_outcome: + return False + raise RuntimeError("Action execution failed") + + config = RailsConfig.from_content( + config={ + "models": [], + "passthrough": True, + "rails": { + "output": { + "parallel": parallel, + "flows": ["failing safety check"], + "streaming": {"enabled": True, "stream_first": stream_first, "chunk_size": 4}, + } + }, + }, + colang_content=""" + define subflow failing safety check + execute failing_rail_action + """, + ) + rails = Guardrails(config, llm=FakeLLMModel(responses=["Unchecked model content"]), use_iorails=use_iorails) + assert isinstance(rails.rails_engine, LLMRails) + rails.register_action(failing_rail_action) + monkeypatch.setattr(api, "_get_rails", AsyncMock(return_value=rails)) + + response = client.post( + "/v1/chat/completions", + json={ + "model": "test-model", + "messages": [{"role": "user", "content": "Hello"}], + "stream": True, + "guardrails": {"config_id": "with_custom_llm"}, + }, + ) + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/event-stream") + frames = [line[6:] for line in response.text.splitlines() if line.startswith("data: ")] + assert frames[-1] == "[DONE]" + events = [json.loads(frame) for frame in frames[:-1]] + content_events = events[:-1] + if stream_first: + assert content_events + assert "".join(event["choices"][0]["delta"]["content"] for event in content_events) == "Unchecked model content" + else: + assert content_events == [] + error_message = ( + "Rail action must return RailOutcome, got bool" + if invalid_outcome + else "Action failing_rail_action failed with status: failed" + ) + assert events[-1] == { + "error": { + "message": f"Internal error in failing safety check rail: {error_message}", + "type": "internal_error", + "param": "failing safety check", + "code": "rail_execution_failure", + } + } + + @pytest.mark.asyncio async def test_openai_sse_format_with_model_name(): """Test that model name is properly included in the response."""