From 67798ccd916ea9d064379d5c1bedbfe6476e4b7c Mon Sep 17 00:00:00 2001 From: foxjl85-dev <272123326+foxjl85-dev@users.noreply.github.com> Date: Mon, 29 Jun 2026 15:58:49 -0500 Subject: [PATCH] docs(proxy): cover duplicate guard review cases --- .../duplicate_burst_guard.py | 22 +++- .../test_duplicate_burst_guard_example.py | 123 ++++++++++++++++++ 2 files changed, 139 insertions(+), 6 deletions(-) diff --git a/cookbook/litellm_proxy_server/duplicate_burst_guard.py b/cookbook/litellm_proxy_server/duplicate_burst_guard.py index 33060687048..5170e7a3553 100644 --- a/cookbook/litellm_proxy_server/duplicate_burst_guard.py +++ b/cookbook/litellm_proxy_server/duplicate_burst_guard.py @@ -11,8 +11,8 @@ This is an in-memory, single-process example. Use a shared store if your proxy runs multiple workers or needs duplicate detection across instances. The request fingerprint uses a whitelist of model-affecting fields and scopes -duplicates by request user/session before API-key owner. Adjust those fields if -your deployment needs different duplicate semantics. +duplicates by authenticated API-key owner plus request user/session. Adjust +those fields if your deployment needs different duplicate semantics. """ import asyncio @@ -32,9 +32,14 @@ from litellm.types.utils import CallTypesLiteral FINGERPRINT_REQUEST_KEYS = ( "model", + "system", "messages", "prompt", "input", + "query", + "documents", + "instructions", + "previous_response_id", "temperature", "max_tokens", "max_completion_tokens", @@ -56,7 +61,9 @@ FINGERPRINT_REQUEST_KEYS = ( "reasoning_effort", "thinking", "extra_body", + "vector_store_id", ) +SKIP_CALL_TYPES: frozenset[str] = frozenset({"aimage_edit"}) class DuplicateBurstGuard(CustomLogger): @@ -76,6 +83,9 @@ class DuplicateBurstGuard(CustomLogger): data: dict[str, object], call_type: CallTypesLiteral, ) -> dict[str, object]: + if call_type in SKIP_CALL_TYPES: + return data + fingerprint = self._fingerprint( data=data, user_api_key_dict=user_api_key_dict, call_type=call_type ) @@ -127,19 +137,19 @@ class DuplicateBurstGuard(CustomLogger): key_user_id: object = None if user_api_key_dict is not None: key_user_id = user_api_key_dict.user_id - user_id: object = ( + request_user_id: object = ( data.get("user") or metadata.get("user_id") or metadata.get("session_id") - or key_user_id - or "anonymous" + or "anonymous_request" ) request = {key: data[key] for key in FINGERPRINT_REQUEST_KEYS if key in data} raw = json.dumps( { + "api_key_identity": key_user_id or "anonymous_key", "call_type": call_type, "request": request, - "request_identity": user_id, + "request_identity": request_user_id, }, default=str, ensure_ascii=False, diff --git a/tests/test_litellm/test_duplicate_burst_guard_example.py b/tests/test_litellm/test_duplicate_burst_guard_example.py index fa753f6005b..75d5672ca77 100644 --- a/tests/test_litellm/test_duplicate_burst_guard_example.py +++ b/tests/test_litellm/test_duplicate_burst_guard_example.py @@ -10,6 +10,7 @@ from litellm.proxy._types import UserAPIKeyAuth from litellm.types.utils import CallTypesLiteral USER_API_KEY = UserAPIKeyAuth(user_id="user-1") +OTHER_USER_API_KEY = UserAPIKeyAuth(user_id="user-2") CACHE = DualCache() CALL_TYPE: CallTypesLiteral = "completion" @@ -238,3 +239,125 @@ async def test_duplicate_burst_guard_ignores_ephemeral_call_ids() -> None: await guard.async_pre_call_hook(USER_API_KEY, CACHE, request_b, CALL_TYPE) assert exc_info.value.status_code == 429 + + +@pytest.mark.asyncio +async def test_duplicate_burst_guard_includes_authenticated_key_identity() -> None: + guard = DuplicateBurstGuard(max_calls=1, window_seconds=60) + request = _chat_request("Summarize invoice A") + request["metadata"] = {"session_id": "shared-session"} + + await guard.async_pre_call_hook(USER_API_KEY, CACHE, request, CALL_TYPE) + accepted = await guard.async_pre_call_hook( + OTHER_USER_API_KEY, CACHE, request, CALL_TYPE + ) + + assert accepted == request + + +@pytest.mark.asyncio +async def test_duplicate_burst_guard_includes_rerank_fields() -> None: + guard = DuplicateBurstGuard(max_calls=1, window_seconds=60) + request_a: dict[str, object] = { + "model": "rerank-english-v3.0", + "query": "invoice A", + "documents": ["invoice A is overdue"], + } + request_b: dict[str, object] = { + "model": "rerank-english-v3.0", + "query": "invoice B", + "documents": ["invoice B has a credit memo"], + } + + await guard.async_pre_call_hook(USER_API_KEY, CACHE, request_a, "rerank") + accepted = await guard.async_pre_call_hook(USER_API_KEY, CACHE, request_b, "rerank") + + assert accepted == request_b + + +@pytest.mark.asyncio +async def test_duplicate_burst_guard_includes_responses_context() -> None: + guard = DuplicateBurstGuard(max_calls=1, window_seconds=60) + request_a: dict[str, object] = { + "model": "gpt-4o-mini", + "input": "continue", + "previous_response_id": "resp-a", + "instructions": "Use policy A", + } + request_b: dict[str, object] = { + "model": "gpt-4o-mini", + "input": "continue", + "previous_response_id": "resp-b", + "instructions": "Use policy B", + } + + await guard.async_pre_call_hook(USER_API_KEY, CACHE, request_a, "aresponses") + accepted = await guard.async_pre_call_hook( + USER_API_KEY, CACHE, request_b, "aresponses" + ) + + assert accepted == request_b + + +@pytest.mark.asyncio +async def test_duplicate_burst_guard_includes_top_level_system_prompt() -> None: + guard = DuplicateBurstGuard(max_calls=1, window_seconds=60) + request_a: dict[str, object] = { + "model": "claude-sonnet-4-5", + "system": "Use policy A", + "messages": [{"role": "user", "content": "Summarize invoice A"}], + } + request_b: dict[str, object] = { + "model": "claude-sonnet-4-5", + "system": "Use policy B", + "messages": [{"role": "user", "content": "Summarize invoice A"}], + } + + await guard.async_pre_call_hook(USER_API_KEY, CACHE, request_a, CALL_TYPE) + accepted = await guard.async_pre_call_hook( + USER_API_KEY, CACHE, request_b, CALL_TYPE + ) + + assert accepted == request_b + + +@pytest.mark.asyncio +async def test_duplicate_burst_guard_includes_search_fields() -> None: + guard = DuplicateBurstGuard(max_calls=1, window_seconds=60) + request_a: dict[str, object] = { + "model": "search-model", + "query": "invoice A", + "vector_store_id": "store-a", + } + request_b: dict[str, object] = { + "model": "search-model", + "query": "invoice B", + "vector_store_id": "store-b", + } + + await guard.async_pre_call_hook(USER_API_KEY, CACHE, request_a, "asearch") + accepted = await guard.async_pre_call_hook( + USER_API_KEY, CACHE, request_b, "asearch" + ) + + assert accepted == request_b + + +@pytest.mark.asyncio +async def test_duplicate_burst_guard_skips_image_edits() -> None: + guard = DuplicateBurstGuard(max_calls=1, window_seconds=60) + request: dict[str, object] = { + "model": "gpt-image-1", + "prompt": "Add the logo", + "image": object(), + "mask": object(), + } + + assert ( + await guard.async_pre_call_hook(USER_API_KEY, CACHE, request, "aimage_edit") + == request + ) + assert ( + await guard.async_pre_call_hook(USER_API_KEY, CACHE, request, "aimage_edit") + == request + )