From aeacb49dc5cb80e718cc59f8d252a37da01885a2 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 9 Sep 2026 21:08:28 -0700 Subject: [PATCH 1/2] feat(guardrails): scan and rewrite streamed /v1/completions in post_call pipelines Gives OpenAITextCompletionHandler a process_output_streaming_response that assembles the buffered TextCompletionResponse chunks per choice index at end of stream, runs the guardrail over the assembled text, and writes text rewrites back into the first text-carrying chunk of each choice. Tests and live proof still pending. --- .../guardrail_translation/handler.py | 88 ++++++++++++++++++- 1 file changed, 85 insertions(+), 3 deletions(-) diff --git a/litellm/llms/openai/completion/guardrail_translation/handler.py b/litellm/llms/openai/completion/guardrail_translation/handler.py index f3557d4017e..8bba5eba931 100644 --- a/litellm/llms/openai/completion/guardrail_translation/handler.py +++ b/litellm/llms/openai/completion/guardrail_translation/handler.py @@ -5,16 +5,18 @@ This module provides guardrail translation support for OpenAI's text completion The handler processes the 'prompt' parameter for guardrails. """ +from collections.abc import Sequence from typing import TYPE_CHECKING, Any, Final from litellm._logging import verbose_proxy_logger -from litellm.llms.base_llm.guardrail_translation.base_translation import BaseTranslation -from litellm.types.utils import GenericGuardrailAPIInputs +from litellm.llms.base_llm.guardrail_translation.base_translation import BaseTranslation, StreamTransformSink +from litellm.llms.base_llm.guardrail_translation.utils import stream_item_field, stream_item_items +from litellm.types.utils import GenericGuardrailAPIInputs, TextChoices, TextCompletionResponse if TYPE_CHECKING: from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj - from litellm.types.utils import TextCompletionResponse + from litellm.proxy._types import UserAPIKeyAuth class OpenAITextCompletionHandler(BaseTranslation): @@ -30,6 +32,9 @@ class OpenAITextCompletionHandler(BaseTranslation): - A list of strings (for batch completions) """ + delivers_ended_stream_rewrites = True + assembles_streamed_response = True + async def process_input_messages( self, data: dict, @@ -193,3 +198,80 @@ class OpenAITextCompletionHandler(BaseTranslation): ) return response + + async def process_output_streaming_response( + self, + responses_so_far: list[TextCompletionResponse], # mutable-ok: rewrites the caller's buffered chunks in place + guardrail_to_apply: "CustomGuardrail", + litellm_logging_obj: "LiteLLMLoggingObj | None" = None, + user_api_key_dict: "UserAPIKeyAuth | None" = None, + request_data: dict | None = None, + stream_transform_sink: StreamTransformSink | None = None, + deliver_ended_stream_rewrites: bool = False, + ) -> list[TextCompletionResponse]: + if not self._check_streaming_has_ended(responses_so_far): + return responses_so_far + assembled: Final = self._assemble_ended_stream(responses_so_far) + pre_guardrail_texts: Final = tuple(choice.text for choice in assembled.choices) + await self.process_output_response( + response=assembled, + guardrail_to_apply=guardrail_to_apply, + litellm_logging_obj=litellm_logging_obj, + user_api_key_dict=user_api_key_dict, + request_data=request_data, + ) + if not deliver_ended_stream_rewrites: + return responses_so_far + for choice, pre_guardrail_text in zip(assembled.choices, pre_guardrail_texts): + if choice.text != pre_guardrail_text: + self._write_choice_text(responses_so_far, choice.index, choice.text) + return responses_so_far + + def _check_streaming_has_ended(self, responses_so_far: Sequence[object]) -> bool: + return any( + stream_item_field(choice, "finish_reason") is not None + for item in responses_so_far + for choice in stream_item_items(item, "choices") + ) + + @staticmethod + def _assemble_ended_stream(responses_so_far: Sequence[TextCompletionResponse]) -> TextCompletionResponse: + first: Final = responses_so_far[0] + streamed_choices: Final = tuple(choice for item in responses_so_far for choice in item.choices) + choice_indices: Final = tuple(dict.fromkeys(choice.index for choice in streamed_choices)) + assembled_choices: Final = [ # mutable-ok: TextCompletionResponse only accepts a list of choices + TextChoices( + index=index, + text="".join( + choice.text for choice in streamed_choices if choice.index == index and isinstance(choice.text, str) + ), + finish_reason=next( + ( + choice.finish_reason + for choice in reversed(streamed_choices) + if choice.index == index and choice.finish_reason is not None + ), + None, + ), + ) + for index in choice_indices + ] + return TextCompletionResponse( + id=first.id, + created=first.created, + model=first.model, + object="text_completion", + choices=assembled_choices, + usage=next((item.usage for item in reversed(responses_so_far) if item.usage is not None), None), + ) + + @staticmethod + def _write_choice_text(responses_so_far: Sequence[TextCompletionResponse], choice_index: int, text: str) -> None: + carriers: Final = tuple( + choice + for item in responses_so_far + for choice in item.choices + if choice.index == choice_index and isinstance(choice.text, str) + ) + for position, carrier in enumerate(carriers): + carrier.text = text if position == 0 else "" From 6fe79c6a30f4a6a0c1dea7a9f239ced278c398a5 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:50:05 -0700 Subject: [PATCH 2/2] fix(guardrails): scan and rewrite streamed /v1/completions under post_call guardrails Assemble each choice's streamed text once the stream ends, run the guardrail on it, write a mask back into the buffered chunks, and end the stream with the guardrail's error on a block. User key metadata now goes through the shared metadata bucket so applied guardrails reach spend logs. --- .../guardrail_translation/handler.py | 93 +++++++---- ...ext_completion_stream_guardrail_rebuild.py | 95 +++++++++-- .../test_text_completion_guardrail_handler.py | 147 +++++++++++++++++- .../proxy_logging/test_guardrail_pipeline.py | 50 +++++- 4 files changed, 337 insertions(+), 48 deletions(-) diff --git a/litellm/llms/openai/completion/guardrail_translation/handler.py b/litellm/llms/openai/completion/guardrail_translation/handler.py index ff1510d4c45..a9fc2ea5a70 100644 --- a/litellm/llms/openai/completion/guardrail_translation/handler.py +++ b/litellm/llms/openai/completion/guardrail_translation/handler.py @@ -5,15 +5,23 @@ This module provides guardrail translation support for OpenAI's text completion The handler processes the 'prompt' parameter for guardrails. """ +import json from collections.abc import Mapping, Sequence +from itertools import chain from typing import TYPE_CHECKING, Final from litellm._logging import verbose_proxy_logger -from litellm.llms.base_llm.guardrail_translation.base_translation import BaseTranslation, StreamTransformSink +from litellm.llms.base_llm.guardrail_translation.base_translation import ( + BaseTranslation, + StreamingScanKey, + StreamTransformSink, +) from litellm.llms.base_llm.guardrail_translation.utils import stream_item_field, stream_item_items from litellm.types.utils import GenericGuardrailAPIInputs, TextChoices, TextCompletionResponse if TYPE_CHECKING: + from fastapi import HTTPException + from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth @@ -128,7 +136,7 @@ class OpenAITextCompletionHandler(BaseTranslation): guardrail_to_apply: "CustomGuardrail", litellm_logging_obj: "LiteLLMLoggingObj | None" = None, user_api_key_dict: "UserAPIKeyAuth | None" = None, - request_data: dict | None = None, + request_data: dict[str, object] | None = None, ) -> "TextCompletionResponse": """ Process output response by applying guardrails to completion text. @@ -165,11 +173,7 @@ class OpenAITextCompletionHandler(BaseTranslation): if "response" not in request_data: request_data["response"] = response - # Add user API key metadata with prefixed keys - if "litellm_metadata" not in request_data: - user_metadata: Final = self.transform_user_api_key_dict_to_metadata(user_api_key_dict) - if user_metadata: - request_data["litellm_metadata"] = user_metadata + self.merge_user_api_key_metadata_into_request(request_data, user_api_key_dict) inputs: Final = GenericGuardrailAPIInputs(texts=texts_to_check) # Include model information from the response if available @@ -205,14 +209,14 @@ class OpenAITextCompletionHandler(BaseTranslation): guardrail_to_apply: "CustomGuardrail", litellm_logging_obj: "LiteLLMLoggingObj | None" = None, user_api_key_dict: "UserAPIKeyAuth | None" = None, - request_data: dict | None = None, + request_data: dict[str, object] | None = None, # mutable-ok: the hooks share and mutate the request payload stream_transform_sink: StreamTransformSink | None = None, deliver_ended_stream_rewrites: bool = False, - ) -> list[TextCompletionResponse]: + ) -> Sequence[TextCompletionResponse]: if not self._check_streaming_has_ended(responses_so_far): return responses_so_far assembled: Final = self._assemble_ended_stream(responses_so_far) - pre_guardrail_texts: Final = tuple(choice.text for choice in assembled.choices) + pre_guardrail_texts: Final = _choice_texts(assembled) await self.process_output_response( response=assembled, guardrail_to_apply=guardrail_to_apply, @@ -222,40 +226,54 @@ class OpenAITextCompletionHandler(BaseTranslation): ) if not deliver_ended_stream_rewrites: return responses_so_far - for choice, pre_guardrail_text in zip(assembled.choices, pre_guardrail_texts): - if choice.text != pre_guardrail_text: - self._write_choice_text(responses_so_far, choice.index, choice.text) + rewrites: Final = tuple( + after for before, after in zip(pre_guardrail_texts, _choice_texts(assembled)) if after != before + ) + for choice_index, text in rewrites: + self._write_choice_text(responses_so_far, choice_index, text) return responses_so_far + def get_streaming_scan_key(self, responses_so_far: Sequence[object]) -> StreamingScanKey | None: + chunks: Final = tuple(chunk for chunk in responses_so_far if isinstance(chunk, TextCompletionResponse)) + if not self._check_streaming_has_ended(chunks): + return None + assembled: Final = self._assemble_ended_stream(chunks) + return StreamingScanKey(texts=tuple(text for _, text in _choice_texts(assembled)), stream_ended=True) + + def build_stream_error_items( + self, + exc: "HTTPException", + responses_so_far: Sequence[object] | None = None, + ) -> Sequence[bytes] | None: + from litellm.proxy.common_request_processing import sse_error_payload + + _, error_obj = sse_error_payload(exc) + return (f'data: {{"error": {json.dumps(error_obj)}}}\n\n'.encode(),) + def _check_streaming_has_ended(self, responses_so_far: Sequence[object]) -> bool: - return any( - stream_item_field(choice, "finish_reason") is not None - for item in responses_so_far - for choice in stream_item_items(item, "choices") - ) + streamed_choices: Final = chain.from_iterable(stream_item_items(item, "choices") for item in responses_so_far) + return any(stream_item_field(choice, "finish_reason") is not None for choice in streamed_choices) @staticmethod def _assemble_ended_stream(responses_so_far: Sequence[TextCompletionResponse]) -> TextCompletionResponse: - first: Final = responses_so_far[0] - streamed_choices: Final = tuple(choice for item in responses_so_far for choice in item.choices) + streamed_choices: Final = tuple(chain.from_iterable(item.choices for item in responses_so_far)) choice_indices: Final = tuple(dict.fromkeys(choice.index for choice in streamed_choices)) - assembled_choices: Final = [ # mutable-ok: TextCompletionResponse only accepts a list of choices + assembled_choices: Final = [ TextChoices( index=index, - text="".join( - choice.text for choice in streamed_choices if choice.index == index and isinstance(choice.text, str) - ), + text="".join(_choice_text(choice) for choice in streamed_choices if choice.index == index), finish_reason=next( ( choice.finish_reason - for choice in reversed(streamed_choices) - if choice.index == index and choice.finish_reason is not None + for choice in streamed_choices + if choice.index == index and choice.finish_reason ), None, ), ) for index in choice_indices ] + first: Final = responses_so_far[0] return TextCompletionResponse( id=first.id, created=first.created, @@ -267,11 +285,18 @@ class OpenAITextCompletionHandler(BaseTranslation): @staticmethod def _write_choice_text(responses_so_far: Sequence[TextCompletionResponse], choice_index: int, text: str) -> None: - carriers: Final = tuple( - choice - for item in responses_so_far - for choice in item.choices - if choice.index == choice_index and isinstance(choice.text, str) - ) - for position, carrier in enumerate(carriers): - carrier.text = text if position == 0 else "" + streamed_choices: Final = chain.from_iterable(item.choices for item in responses_so_far) + head, *tail = (choice for choice in streamed_choices if choice.index == choice_index) + head.text = text + for choice in tail: + if isinstance(stream_item_field(choice, "text"), str): + choice.text = "" + + +def _choice_text(choice: TextChoices) -> str: + text: Final = stream_item_field(choice, "text") + return text if isinstance(text, str) else "" + + +def _choice_texts(response: TextCompletionResponse) -> tuple[tuple[int, str], ...]: + return tuple((choice.index, _choice_text(choice)) for choice in response.choices) diff --git a/tests/integration/streaming/test_text_completion_stream_guardrail_rebuild.py b/tests/integration/streaming/test_text_completion_stream_guardrail_rebuild.py index 229a9be6d20..ea0a820a705 100644 --- a/tests/integration/streaming/test_text_completion_stream_guardrail_rebuild.py +++ b/tests/integration/streaming/test_text_completion_stream_guardrail_rebuild.py @@ -27,7 +27,12 @@ _PROXY_CONFIG: Final = Path(__file__).resolve().parents[1] / "proxy_config.yaml" _PROVIDER_KEY: Final = "synthetic-provider-key" _GUARDRAIL: Final = "audit-text-rebuild-tool-permission" _TEXT_MODEL: Final = f"audit-text-instruct-{uuid.uuid4().hex}" +_GOVERNED_MODEL: Final = f"audit-text-governed-{uuid.uuid4().hex}" +_KEYWORD_FILTER: Final = "audit-text-keyword-filter" +_POLICY: Final = "audit-text-post-call-governance" _PIECES: Final = ("Hello", " streamed", " world") +_MASKED_PIECES: Final = ("A pers", "im", "mon tart") +_BLOCKED_PIECES: Final = ("A kum", "quat tart") _PROMPT_TOKENS: Final = 11 _COMPLETION_TOKENS: Final = 3 _USAGE: Final[dict[str, JsonValue]] = { @@ -97,6 +102,14 @@ def _text_choice(text: str, finish_reason: str | None) -> dict[str, JsonValue]: return {"text": text, "index": 0, "logprobs": None, "finish_reason": finish_reason} +def _pieces_for(identity: str) -> tuple[str, ...]: + if identity.startswith("cmpl-mask-"): + return _MASKED_PIECES + if identity.startswith("cmpl-block-"): + return _BLOCKED_PIECES + return _PIECES + + def _respond(request: Request) -> Reply: body: Final = JSON_OBJECT.validate_json(request.body) assert body.get("stream") is True, body @@ -106,7 +119,7 @@ def _respond(request: Request) -> Reply: return Reply( content_type="text/event-stream", chunks=( - *(_text_frame(identity, [_text_choice(piece, None)]) for piece in _PIECES), + *(_text_frame(identity, [_text_choice(piece, None)]) for piece in _pieces_for(identity)), _text_frame(identity, [_text_choice("", "length")]), _text_frame(identity, [], usage=_USAGE), b"data: [DONE]\n\n", @@ -114,20 +127,25 @@ def _respond(request: Request) -> Reply: ) +def _text_deployment(model_name: str, wire: Wire) -> dict[str, JsonValue]: + return { + "model_name": model_name, + "litellm_params": { + "model": "text-completion-openai/gpt-3.5-turbo-instruct", + "api_base": f"{wire.url}/v1", + "api_key": _PROVIDER_KEY, + }, + } + + def _config(directory: Path, wire: Wire) -> Path: base: Final = object_value(yaml.safe_load(_PROXY_CONFIG.read_text())) config: Final = { **base, "model_list": [ *list_value(base["model_list"]), - { - "model_name": _TEXT_MODEL, - "litellm_params": { - "model": "text-completion-openai/gpt-3.5-turbo-instruct", - "api_base": f"{wire.url}/v1", - "api_key": _PROVIDER_KEY, - }, - }, + _text_deployment(_TEXT_MODEL, wire), + _text_deployment(_GOVERNED_MODEL, wire), ], "guardrails": [ { @@ -139,8 +157,30 @@ def _config(directory: Path, wire: Wire) -> Path: "default_action": "allow", "rules": [], }, - } + }, + { + "guardrail_name": _KEYWORD_FILTER, + "litellm_params": { + "guardrail": "litellm_content_filter", + "mode": "post_call", + "default_on": False, + "blocked_words": [ + {"keyword": "persimmon", "action": "MASK"}, + {"keyword": "kumquat", "action": "BLOCK"}, + ], + }, + }, ], + "policies": { + _POLICY: { + "guardrails": {"add": [_KEYWORD_FILTER]}, + "pipeline": { + "mode": "post_call", + "steps": [{"guardrail": _KEYWORD_FILTER, "on_pass": "allow", "on_fail": "block"}], + }, + } + }, + "policy_attachments": [{"policy": _POLICY, "models": [_GOVERNED_MODEL]}], } path: Final = directory / f"text-rebuild-{uuid.uuid4().hex}.yaml" path.write_text(yaml.safe_dump(config)) @@ -163,14 +203,14 @@ def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_Rig]: yield _Rig(wire, owned) -def _stream(rig: _Rig, identity: str) -> _Stream: +def _stream(rig: _Rig, identity: str, model: str = _TEXT_MODEL) -> _Stream: gateway: Final = rig.proxy.gateway with gateway.client.stream( "POST", "/v1/completions", headers={"Authorization": f"Bearer {gateway.key}"}, json={ - "model": _TEXT_MODEL, + "model": model, "prompt": identity, "max_tokens": 16, "stream": True, @@ -212,6 +252,37 @@ def test_guarded_text_completion_stream_reaches_the_caller_whole_with_its_usage( assert _GUARDRAIL in list_value(object_value(row["metadata"])["applied_guardrails"]), row +def test_post_call_pipeline_masks_a_streamed_text_completion(rig: _Rig) -> None: + identity: Final = f"cmpl-mask-{uuid.uuid4().hex}" + + stream: Final = _stream(rig, identity, _GOVERNED_MODEL) + + assert stream.status == 200, stream.body + assert stream.body.rstrip().endswith(b"data: [DONE]"), stream.body[-400:] + assert stream.text == "A [KEYWORD_REDACTED] tart", stream.frames + assert b"persimmon" not in stream.body, stream.body + assert stream.finish_reasons == ("length",), stream.frames + assert [(usage["prompt_tokens"], usage["completion_tokens"]) for usage in stream.usages] == [ + (_PROMPT_TOKENS, _COMPLETION_TOKENS) + ], stream.frames + (row,) = eventually( + lambda: rig.proxy.gateway.rows('SELECT metadata FROM "LiteLLM_SpendLogs" WHERE request_id = %s', (identity,)), + lambda rows: len(rows) == 1, + seconds=70, + ) + assert _KEYWORD_FILTER in list_value(object_value(row["metadata"])["applied_guardrails"]), row + + +def test_post_call_pipeline_blocks_a_streamed_text_completion(rig: _Rig) -> None: + identity: Final = f"cmpl-block-{uuid.uuid4().hex}" + + stream: Final = _stream(rig, identity, _GOVERNED_MODEL) + + assert stream.status == 400, stream.body + assert b"Content blocked: keyword 'kumquat' detected" in stream.body, stream.body + assert b"quat tart" not in stream.body, stream.body + + def _together_chunk(choices: list[JsonValue], **extra: JsonValue) -> bytes: return _frame( { diff --git a/tests/unit/llms/openai/completion/test_text_completion_guardrail_handler.py b/tests/unit/llms/openai/completion/test_text_completion_guardrail_handler.py index c96fbf34fe1..33e6a9887c2 100644 --- a/tests/unit/llms/openai/completion/test_text_completion_guardrail_handler.py +++ b/tests/unit/llms/openai/completion/test_text_completion_guardrail_handler.py @@ -6,6 +6,7 @@ from typing import List, Optional, Tuple from unittest.mock import MagicMock import pytest +from fastapi import HTTPException from litellm.integrations.custom_guardrail import CustomGuardrail @@ -13,7 +14,11 @@ from litellm.llms import get_guardrail_translation_mapping from litellm.llms.openai.completion.guardrail_translation.handler import ( OpenAITextCompletionHandler, ) -from litellm.types.utils import CallTypes, TextChoices, TextCompletionResponse +from litellm.proxy._types import UserAPIKeyAuth +from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import ContentFilterGuardrail +from litellm.proxy.guardrails.guardrail_hooks.unified_guardrail.unified_guardrail import UnifiedLLMGuardrails +from litellm.types.guardrails import BlockedWord, ContentFilterAction, GuardrailEventHooks +from litellm.types.utils import CallTypes, TextChoices, TextCompletionResponse, Usage class MockGuardrail(CustomGuardrail): @@ -341,3 +346,143 @@ class TestPIIMaskingScenario: assert "[EMAIL_REDACTED]" in result["prompt"][1] assert "alice@example.com" not in result["prompt"][0] assert "bob@test.org" not in result["prompt"][1] + + +class _RecordingKeywordFilter(ContentFilterGuardrail): + def __init__(self): + super().__init__( + guardrail_name="keyword-filter", + blocked_words=[ + BlockedWord(keyword="persimmon", action=ContentFilterAction.MASK), + BlockedWord(keyword="kumquat", action=ContentFilterAction.BLOCK), + ], + event_hook=GuardrailEventHooks.post_call, + default_on=True, + ) + self.streaming_sampling_rate = 1 + self.scanned_texts: tuple[tuple[str, ...], ...] = () + + async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None): + self.scanned_texts = (*self.scanned_texts, tuple(inputs.get("texts") or ())) + return await super().apply_guardrail(inputs, request_data, input_type, logging_obj) + + +def _text_chunk(index: int, text: str | None, finish_reason: str | None = None) -> TextCompletionResponse: + return TextCompletionResponse( + id="cmpl-stream", + created=1589478378, + model="gpt-3.5-turbo-instruct", + object="text_completion", + choices=[TextChoices(index=index, text=text, finish_reason=finish_reason)], + ) + + +def _usage_chunk() -> TextCompletionResponse: + return TextCompletionResponse( + id="cmpl-stream", + created=1589478378, + model="gpt-3.5-turbo-instruct", + object="text_completion", + choices=[], + usage=Usage(prompt_tokens=3, completion_tokens=5, total_tokens=8), + ) + + +def _delivered(chunks: list[TextCompletionResponse]) -> list[tuple[int, str | None, str | None]]: + return [(choice.index, choice.text, choice.finish_reason) for chunk in chunks for choice in chunk.choices] + + +async def _drive_flat_stream(guardrail: CustomGuardrail, chunks: list[TextCompletionResponse]) -> list[object]: + async def upstream(): + for chunk in chunks: + yield chunk + + return [ + item + async for item in UnifiedLLMGuardrails().async_post_call_streaming_iterator_hook( + user_api_key_dict=UserAPIKeyAuth(api_key="test-key", request_route="/v1/completions"), + response=upstream(), + request_data={"guardrail_to_apply": guardrail, "model": "gpt-3.5-turbo-instruct"}, + ) + ] + + +class TestStreamingOutputProcessing: + @pytest.mark.asyncio + async def test_ended_stream_delivers_each_choice_masked_text(self): + guardrail = _RecordingKeywordFilter() + chunks = [ + _text_chunk(0, "A pers"), + _text_chunk(1, "kiwi "), + _text_chunk(0, "im"), + _text_chunk(1, "pie", "stop"), + _text_chunk(0, "mon tart"), + _text_chunk(0, None, "length"), + _usage_chunk(), + ] + + await OpenAITextCompletionHandler().process_output_streaming_response( + responses_so_far=chunks, + guardrail_to_apply=guardrail, + request_data={}, + deliver_ended_stream_rewrites=True, + ) + + assert guardrail.scanned_texts == (("A persimmon tart", "kiwi pie"),) + assert _delivered(chunks) == [ + (0, "A [KEYWORD_REDACTED] tart", None), + (1, "kiwi ", None), + (0, "", None), + (1, "pie", "stop"), + (0, "", None), + (0, None, "length"), + ] + assert chunks[-1].usage.total_tokens == 8 + + @pytest.mark.asyncio + async def test_ended_stream_scanned_without_delivery_keeps_chunks_and_stores_the_assembled_response(self): + chunks = [_text_chunk(0, "A pers"), _text_chunk(0, "immon tart"), _text_chunk(0, "", "stop"), _usage_chunk()] + request_data: dict[str, object] = {} + + await OpenAITextCompletionHandler().process_output_streaming_response( + responses_so_far=chunks, guardrail_to_apply=_RecordingKeywordFilter(), request_data=request_data + ) + + assert _delivered(chunks) == [(0, "A pers", None), (0, "immon tart", None), (0, "", "stop")] + assembled = request_data["response"] + assert isinstance(assembled, TextCompletionResponse) + assert [(c.text, c.finish_reason) for c in assembled.choices] == [("A [KEYWORD_REDACTED] tart", "stop")] + assert assembled.usage.total_tokens == 8 + + @pytest.mark.asyncio + async def test_ended_stream_block_raises_the_guardrail_error(self): + chunks = [_text_chunk(0, "A kum"), _text_chunk(0, "quat tart"), _text_chunk(0, "", "stop")] + + with pytest.raises(HTTPException) as exc_info: + await OpenAITextCompletionHandler().process_output_streaming_response( + responses_so_far=chunks, guardrail_to_apply=_RecordingKeywordFilter(), deliver_ended_stream_rewrites=True + ) + + assert exc_info.value.status_code == 400 + assert "kumquat" in str(exc_info.value.detail) + + @pytest.mark.asyncio + async def test_flat_stream_is_scanned_once_after_it_ends(self): + guardrail = _RecordingKeywordFilter() + chunks = [_text_chunk(0, "A pers"), _text_chunk(0, "immon tart"), _text_chunk(0, "", "stop"), _usage_chunk()] + + delivered = await _drive_flat_stream(guardrail, chunks) + + assert guardrail.scanned_texts == (("A persimmon tart",),) + assert len(delivered) == len(chunks) + + @pytest.mark.asyncio + async def test_flat_stream_block_ends_with_an_in_band_error_frame(self): + chunks = [_text_chunk(0, "A kum"), _text_chunk(0, "quat tart"), _text_chunk(0, "", "stop"), _usage_chunk()] + + delivered = await _drive_flat_stream(_RecordingKeywordFilter(), chunks) + + error_frame = delivered[-1] + assert isinstance(error_frame, bytes) + assert error_frame.startswith(b'data: {"error": ') + assert b"kumquat" in error_frame diff --git a/tests/unit/proxy/utils/proxy_logging/test_guardrail_pipeline.py b/tests/unit/proxy/utils/proxy_logging/test_guardrail_pipeline.py index acc1463b072..51d22fa93ce 100644 --- a/tests/unit/proxy/utils/proxy_logging/test_guardrail_pipeline.py +++ b/tests/unit/proxy/utils/proxy_logging/test_guardrail_pipeline.py @@ -41,6 +41,7 @@ from litellm.types.proxy.policy_engine.pipeline_types import ( GuardrailPipeline, PipelineStep, ) +from litellm.types.utils import TextChoices, TextCompletionResponse @pytest.fixture(autouse=True) @@ -1886,7 +1887,7 @@ def test_streamable_post_call_pipelines_keeps_hook_guardrails_and_drops_iterator @pytest.mark.parametrize( "request_route", - ["/v1/completions", "/v1beta/models/gemini-2.5-flash:streamGenerateContent", "/a2a/agent"], + ["/v1beta/models/gemini-2.5-flash:streamGenerateContent", "/a2a/agent"], ) def test_streamable_post_call_pipelines_keeps_legacy_hooks_off_routes_that_assemble_no_response( make_user_api_key_auth, monkeypatch, caplog, request_route @@ -2036,6 +2037,53 @@ async def test_streaming_iterator_hook_delivers_legacy_hook_rewrite_on_every_cho assert _warnings(caplog) == [] +def _text_completion_chunk(text: str, finish_reason: str | None = None) -> TextCompletionResponse: + return TextCompletionResponse( + id="cmpl-1", model="m", choices=[TextChoices(index=0, text=text, finish_reason=finish_reason)] + ) + + +def _rewritten_text_completion(response: Any) -> TextCompletionResponse: + return TextCompletionResponse( + id=response.id, + model=response.model, + choices=[TextChoices(index=0, text="[REWRITTEN] " + response.choices[0].text, finish_reason="stop")], + ) + + +@pytest.mark.asyncio +async def test_streaming_iterator_hook_delivers_legacy_hook_rewrite_on_a_text_completion_stream( + proxy_logging, make_user_api_key_auth, monkeypatch, caplog +): + seen: Dict[str, Any] = {} + monkeypatch.setattr(litellm, "callbacks", [_legacy_hook_stream_guardrail(seen, rewrite=_rewritten_text_completion)]) + monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None, raising=False) + data = {**_post_call_pipeline_data(stream=True), "prompt": "hi"} + chunks = [_text_completion_chunk("hello "), _text_completion_chunk("world"), _text_completion_chunk("", "stop")] + auth = make_user_api_key_auth(request_route="/v1/completions") + + with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"): + await proxy_logging.pre_call_hook( + user_api_key_dict=auth, data=data, call_type="text_completion", guardrails_only=True + ) + delivered = [ + item + async for item in proxy_logging.async_post_call_streaming_iterator_hook( + user_api_key_dict=auth, response=_async_chunk_iter(chunks), request_data=data + ) + ] + + assert isinstance(seen["response"], TextCompletionResponse) + assert seen["response"].choices[0].text == "hello world" + assert [(item.choices[0].text, item.choices[0].finish_reason) for item in delivered] == [ + ("[REWRITTEN] hello world", None), + ("", None), + ("", "stop"), + ] + assert data["metadata"]["applied_guardrails"] == ["gr-post"] + assert _warnings(caplog) == [] + + @pytest.mark.asyncio async def test_streaming_iterator_hook_releases_stream_untouched_when_legacy_hook_returns_none( proxy_logging, make_user_api_key_auth, monkeypatch