docs(proxy): cover duplicate guard review cases

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,

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
+    )