fix(batches): price a retrieved batch from its deployment's model and rates

litellm/batches/batch_utils.py

@@ -48,6 +48,7 @@ async def _handle_completed_batch(
     custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"],
     model_name: str | None = None,
     litellm_params: dict | None = None,
+    model_info: ModelInfo | None = None,
 ) -> tuple[float, Usage, list[str]]:
     """Fetch a completed batch's output file and aggregate its cost, usage, and
     models in a single pass over the JSONL lines, so the parsed file content is
@@ -58,6 +59,9 @@ async def _handle_completed_batch(
         custom_llm_provider: The LLM provider
         model_name: Optional model name
         litellm_params: Optional litellm parameters containing credentials (api_key, api_base, etc.)
+        model_info: Optional deployment-level model info with custom pricing,
+            threaded through so a deployment's configured rates win over the
+            global cost map.
     """
     file_content = await _fetch_batch_output_file_content(batch, custom_llm_provider, litellm_params=litellm_params)
 
@@ -75,6 +79,7 @@ async def _handle_completed_batch(
         entries=_iter_batch_input_entries(file_content),
         custom_llm_provider=custom_llm_provider,
         model_name=model_name,
+        model_info=model_info,
     )
 
 

litellm/cost_calculator.py

@@ -2160,7 +2160,7 @@ def batch_cost_calculator(
     output_cost_per_token: Final = model_info.get("output_cost_per_token")
     total_prompt_cost = 0.0
     total_completion_cost = 0.0
-    if input_cost_per_token_batches:
+    if input_cost_per_token_batches is not None:
         total_prompt_cost = usage.prompt_tokens * input_cost_per_token_batches
     elif input_cost_per_token:
         details: Final = parse_prompt_tokens_details(usage)
@@ -2180,7 +2180,7 @@ def batch_cost_calculator(
 
         cache_creation_cost: Final = model_info.get("cache_creation_input_token_cost") or input_cost_per_token
         total_prompt_cost += cache_creation_tokens * cache_creation_cost / 2
-    if output_cost_per_token_batches:
+    if output_cost_per_token_batches is not None:
         total_completion_cost = usage.completion_tokens * output_cost_per_token_batches
     elif output_cost_per_token:
         total_completion_cost = (

litellm/litellm_core_utils/litellm_logging.py

@@ -107,6 +107,7 @@
     LiteLLMBatch,
     LiteLLMLoggingBaseClass,
     LiteLLMRealtimeStreamLoggingObject,
+    ModelInfo,
     ModelResponse,
     ModelResponseStream,
     RawRequestTypedDict,
@@ -306,6 +307,14 @@ def _get_cached_prometheus_logger():
     return _PrometheusLogger
 
 
+_DEPLOYMENT_PRICING_KEYS: Final = (
+    "input_cost_per_token",
+    "output_cost_per_token",
+    "input_cost_per_token_batches",
+    "output_cost_per_token_batches",
+)
+
+
 class Logging(LiteLLMLoggingBaseClass):
     global \
         supabaseClient, \
@@ -578,6 +587,67 @@ def get_router_model_id(self) -> str | None:
                 return model_id
         return None
 
+    def get_deployment_model_for_cost(self) -> str | None:
+        """The provider-qualified model to price against.
+
+        On a batch retrieve both self.model and litellm_params["model"] can be
+        unset, and self.model can otherwise carry the router's model_group alias,
+        which no cost map resolves. model_call_details holds the deployment's own
+        provider-qualified model, so it is preferred.
+        """
+        candidates: Final = (
+            (self.model_call_details or {}).get("model") if hasattr(self, "model_call_details") else None,
+            self.litellm_params.get("model") if hasattr(self, "litellm_params") else None,
+            self.model,
+        )
+        return next((candidate for candidate in candidates if isinstance(candidate, str) and candidate), None)
+
+    def get_router_deployment_model_info(self) -> ModelInfo | None:
+        """Pricing the router registered under this deployment's model_info.id.
+
+        Returns None when the deployment declares no pricing of its own, so the
+        caller falls back to the global cost map. The raw registration is what
+        decides that: the router registers an entry for every deployment, and
+        get_model_info fills absent costs with 0, so asking it directly cannot
+        tell "configured as free" apart from "no pricing configured". A deployment
+        may declare only one side of its pricing, so a rate it leaves unset keeps
+        the model's published value instead of billing as zero.
+        """
+        model_id: Final = self.get_router_model_id()
+        if model_id is None:
+            return None
+        registered: Final = litellm.model_cost.get(model_id)
+        if not isinstance(registered, dict) or not any(
+            registered.get(key) is not None for key in _DEPLOYMENT_PRICING_KEYS
+        ):
+            return None
+        try:
+            merged: Final = litellm.get_model_info(model=model_id)
+        except Exception:  # noqa: BLE001  # get_model_info raises for ids it cannot resolve a provider for
+            return None
+        published: Final = self._published_model_info()
+        if published is None:
+            return merged
+        if registered.get("input_cost_per_token") is None:
+            merged["input_cost_per_token"] = published.get("input_cost_per_token")
+        if registered.get("output_cost_per_token") is None:
+            merged["output_cost_per_token"] = published.get("output_cost_per_token")
+        if registered.get("input_cost_per_token_batches") is None:
+            merged["input_cost_per_token_batches"] = published.get("input_cost_per_token_batches")
+        if registered.get("output_cost_per_token_batches") is None:
+            merged["output_cost_per_token_batches"] = published.get("output_cost_per_token_batches")
+        return merged
+
+    def _published_model_info(self) -> ModelInfo | None:
+        """The cost map's own entry for this deployment's model, when it resolves."""
+        deployment_model: Final = self.get_deployment_model_for_cost()
+        if deployment_model is None:
+            return None
+        try:
+            return litellm.get_model_info(model=deployment_model)
+        except Exception:  # noqa: BLE001  # no published entry to layer the declared rates over
+            return None
+
     def update_environment_variables(
         self,
         litellm_params: dict,
@@ -2592,7 +2662,9 @@ async def _async_success_handler_body(
                 ) = await _handle_completed_batch(
                     batch=result,
                     custom_llm_provider=self.custom_llm_provider,
+                    model_name=self.get_deployment_model_for_cost(),
                     litellm_params=self.litellm_params,
+                    model_info=self.get_router_deployment_model_info(),
                 )
 
                 result._hidden_params["response_cost"] = response_cost

tests/test_litellm/batches/test_batch_utils.py

@@ -1299,3 +1299,91 @@ async def fake_afile_content(**kw):
     assert captured["aws_region_name"] == "us-west-2"
     assert captured["_litellm_internal_model_credentials"] is snapshot
     assert "model" not in captured
+
+
+# =========================================================================== #
+# _handle_completed_batch threads the deployment's model identity + pricing
+# =========================================================================== #
+
+
+def _bedrock_row(model: str, input_tokens: int, output_tokens: int) -> dict[str, object]:
+    return {
+        "modelInput": {"messages": [{"role": "user", "content": [{"type": "text", "text": "hi"}]}]},
+        "modelOutput": {
+            "model": model,
+            "id": "msg_1",
+            "type": "message",
+            "role": "assistant",
+            "content": [{"type": "text", "text": "ok"}],
+            "stop_reason": "end_turn",
+            "usage": {
+                "input_tokens": input_tokens,
+                "output_tokens": output_tokens,
+                "cache_creation_input_tokens": 0,
+                "cache_read_input_tokens": 0,
+            },
+        },
+        "recordId": "r",
+    }
+
+
+@pytest.mark.asyncio
+async def test_handle_completed_bedrock_batch_prices_from_deployment_model(monkeypatch) -> None:
+    """A bedrock batch must price from the deployment model, not the response model."""
+    rows = [_bedrock_row("claude-sonnet-4-6", 18, 10)] * 100
+
+    async def fake_fetch(batch: object, custom_llm_provider: str, litellm_params: dict | None = None) -> bytes:
+        return _vertex_jsonl(rows)
+
+    monkeypatch.setattr(bu, "_fetch_batch_output_file_content", fake_fetch)
+
+    cost, usage, _ = await bu._handle_completed_batch(
+        _batch("of"),
+        custom_llm_provider="bedrock",
+        model_name="bedrock/global.anthropic.claude-sonnet-4-6",
+    )
+
+    assert (usage.prompt_tokens, usage.completion_tokens, usage.total_tokens) == (1800, 1000, 2800)
+    # 3e-06 / 1.5e-05 on-demand, halved for batch.
+    assert cost == pytest.approx(1800 * 3e-06 / 2 + 1000 * 1.5e-05 / 2)
+
+    # The response model alone cannot price a bedrock batch: this is the $0 bug.
+    zero_cost, zero_usage, _ = await bu._handle_completed_batch(
+        _batch("of"),
+        custom_llm_provider="bedrock",
+        model_name=None,
+    )
+    assert zero_cost == 0.0
+    assert zero_usage.total_tokens == 2800
+
+
+@pytest.mark.asyncio
+async def test_handle_completed_batch_honors_deployment_pricing(monkeypatch) -> None:
+    """A deployment's configured rates must win over the global cost map."""
+    rows = [_success_row(model="gemini-2.5-flash", usage=_usage(60, 75))]
+
+    async def fake_fetch(batch: object, custom_llm_provider: str, litellm_params: dict | None = None) -> bytes:
+        return _vertex_jsonl(rows)
+
+    monkeypatch.setattr(bu, "_fetch_batch_output_file_content", fake_fetch)
+
+    free_cost, _, _ = await bu._handle_completed_batch(
+        _batch("of"),
+        custom_llm_provider="vertex_ai",
+        model_name="vertex_ai/gemini-2.5-flash",
+        model_info={
+            "input_cost_per_token": 0.0,
+            "output_cost_per_token": 0.0,
+            "input_cost_per_token_batches": 0.0,
+            "output_cost_per_token_batches": 0.0,
+        },
+    )
+    assert free_cost == 0.0
+
+    billed_cost, _, _ = await bu._handle_completed_batch(
+        _batch("of"),
+        custom_llm_provider="vertex_ai",
+        model_name="vertex_ai/gemini-2.5-flash",
+        model_info=None,
+    )
+    assert billed_cost > 0.0