Merge remote-tracking branch 'origin/main' into litellm_layer_d1_passthrough_logging_obj

This commit is contained in:
nate-berri 2026-10-10 21:06:20 +00:00
commit 0d99b11d3e
No known key found for this signature in database
488 changed files with 35497 additions and 18057 deletions

View file

@ -243,7 +243,7 @@ commands:
mkdir -p "$HOME/.local/bin"
mv /tmp/codecov "$HOME/.local/bin/codecov"
start_openai_record_replay_proxy:
description: "Start the record/replay proxy (tests/_openai_record_replay_proxy.py) on host port 8090 and wait until healthy. Models whose api_base points here replay recorded provider responses, so the E2E run neither pays for nor depends on the live provider. The default upstream is OpenAI; a non-OpenAI model must point its api_base at /__recorder_upstream/<host>/ so the recorder forwards there instead of defaulting to OpenAI. Run after uv deps are synced."
description: "Start the record/replay proxy (tests/_openai_record_replay_proxy.py) on host port 8090 and wait until healthy. Models whose api_base points here replay Redis cache hits. Cache misses call the live provider and store its response, so these legacy E2E jobs still require provider availability and may incur provider charges. The default upstream is OpenAI; a non-OpenAI model must point its api_base at /__recorder_upstream/<host>/ so the recorder forwards there instead of defaulting to OpenAI. Run after uv deps are synced."
steps:
- run:
name: Start record/replay proxy
@ -379,50 +379,6 @@ commands:
- save_cargo_target
jobs:
# Add Windows testing job
using_litellm_on_windows:
executor:
name: win/default
shell: powershell.exe
working_directory: ~/project
environment:
UV_PYTHON: "3.11"
CARGO_HTTP_MULTIPLEXING: "false"
CARGO_NET_RETRY: "5"
steps:
- checkout
- run:
name: Install Python
command: |
choco install python --version=3.11.0 -y --no-progress --force
refreshenv
python --version
environment:
CHOCOLATEY_CONFIRM_ALL: "true"
- install_windows_toolchain
- run:
name: Install Dependencies
no_output_timeout: 30m
environment:
UV_HTTP_TIMEOUT: "300"
command: |
$env:Path = "$HOME\.cargo\bin;$HOME\.local\bin;$env:Path"
for ($attempt = 1; $attempt -le 5; $attempt++) {
Write-Host "uv sync attempt $attempt/5"
uv sync --frozen --group dev --python 3.11
if ($LASTEXITCODE -eq 0) {
break
}
if ($attempt -eq 5) {
exit $LASTEXITCODE
}
Start-Sleep -Seconds 15
}
- run:
name: Run Windows-specific test
command: |
uv run --no-sync python -m pytest --tb=short tests/windows_tests/ -v
windows_release_wheel:
executor:
name: win/default
@ -808,7 +764,7 @@ jobs:
- *python312_image
working_directory: ~/project
resource_class: large
parallelism: 4
parallelism: 1
environment:
FAKE_OPENAI_API_BASE: http://127.0.0.1:8190
steps:
@ -1384,89 +1340,6 @@ jobs:
paths:
- audio_coverage.xml
- audio_coverage
installing_litellm_on_python:
docker:
- *python312_image
- *postgres16_sidecar
working_directory: ~/project
environment: *sidecar_database_url
steps:
- checkout
- skip_if_unrelated_changes
- setup_google_dns
- install_uv
- install_rust
- run:
name: Install Dependencies
command: |
uv sync --frozen --all-groups --all-extras --python 3.12
- setup_litellm_enterprise_pip
- wait_for_service:
url: tcp://localhost:5432
timeout: "60"
- run:
name: Run tests
command: |
uv run --no-sync python -m pytest --tb=short -vv tests/local_testing/test_basic_python_version.py -k "not legacy_resolver"
installing_litellm_on_python_3_13:
docker:
- image: cimg/python:3.13.1@sha256:87b243ae80d154db75ce5e58af16c72c5dd4b1e23e5c7264a816e85e0c440c13
auth:
username: ${DOCKERHUB_USERNAME}
password: ${DOCKERHUB_PASSWORD}
- *postgres16_sidecar
working_directory: ~/project
environment: *sidecar_database_url
resource_class: medium
steps:
- checkout
- skip_if_unrelated_changes
- setup_google_dns
- install_uv
- install_rust
- run:
name: Install Dependencies
command: |
uv sync --frozen --all-groups --all-extras --python 3.13
- wait_for_service:
url: tcp://localhost:5432
timeout: "60"
- run:
name: Run tests
command: |
uv run --no-sync python -m pytest --tb=short -v tests/local_testing/test_basic_python_version.py -k "not legacy_resolver"
installing_litellm_on_python_v2_migration_resolver:
docker:
- *python312_image
- *postgres16_sidecar
working_directory: ~/project
environment: *sidecar_database_url
steps:
- checkout
- skip_if_unrelated_changes
- setup_google_dns
- install_uv
- install_rust
- run:
name: Install Dependencies
command: |
uv sync --frozen --all-groups --all-extras --python 3.12
- setup_litellm_enterprise_pip
- wait_for_service:
url: tcp://localhost:5432
timeout: "60"
- run:
name: Run both migration resolvers against Postgres
command: |
uv run --no-sync python -m pytest --tb=short -vv \
tests/local_testing/test_basic_python_version.py::test_litellm_proxy_server_config_no_general_settings \
tests/local_testing/test_basic_python_version.py::test_litellm_proxy_server_config_no_general_settings_legacy_resolver
helm_chart_testing:
machine:
image: ubuntu-2204:2024.04.1 # Use machine executor instead of docker
@ -2159,95 +2032,6 @@ jobs:
- store_test_results:
path: test-results
proxy_build_from_pip_tests:
# Change from docker to machine executor
machine:
image: ubuntu-2204:2024.04.1
resource_class: large
working_directory: ~/project
steps:
- checkout
- run:
name: Generate LiteLLM master key
command: |
key="$(openssl rand -hex 16)"
printf 'export LITELLM_MASTER_KEY=sk-%s\n' "$key" >> "$BASH_ENV"
- skip_if_unrelated_changes
- setup_google_dns
# Remove Docker CLI installation since it's already available in machine executor
- install_uv
- install_rust
- run:
name: Install Dependencies
command: |
uv sync --frozen --all-groups --all-extras --python 3.12
- run:
name: Build Docker image
command: |
docker build -t my-app:latest -f docker/build_from_pip/Dockerfile.build_from_pip .
- start_postgres
- start_fake_openai_endpoint
- run:
name: Run Docker container
# intentionally give bad redis credentials here
# the OTEL test - should get this as a trace
command: |
docker run -d \
-p 4000:4000 \
-e DATABASE_URL=postgresql://postgres:postgres@host.docker.internal:5432/circle_test \
-e REDIS_HOST=$REDIS_HOST \
-e REDIS_PASSWORD=$REDIS_PASSWORD \
-e REDIS_PORT=$REDIS_PORT \
-e LITELLM_MASTER_KEY="$LITELLM_MASTER_KEY" \
-e OPENAI_API_KEY=$OPENAI_API_KEY \
-e FAKE_OPENAI_API_BASE=http://host.docker.internal:8190 \
-e LITELLM_LICENSE=$LITELLM_LICENSE \
-e OTEL_EXPORTER="in_memory" \
-e AWS_ACCESS_KEY_ID=$AWS_ACCESS_KEY_ID \
-e AWS_SECRET_ACCESS_KEY=$AWS_SECRET_ACCESS_KEY \
-e AWS_REGION_NAME=$AWS_REGION_NAME \
-e COHERE_API_KEY=$COHERE_API_KEY \
-e USE_DDTRACE=True \
-e DD_API_KEY=$DD_API_KEY \
-e DD_SITE=$DD_SITE \
-e GCS_FLUSH_INTERVAL="1" \
-e LITELLM_LOG=ERROR \
--add-host host.docker.internal:host-gateway \
--name my-app \
-v $(pwd)/docker/build_from_pip/litellm_config.yaml:/app/config.yaml \
my-app:latest \
--config /app/config.yaml \
--port 4000
- run:
name: Start outputting logs
command: docker logs -f my-app
background: true
- wait_for_service:
url: http://localhost:4000
timeout: "300"
- run:
name: Run tests
command: |
mkdir -p test-results
TEST_FILES=$(circleci tests glob "tests/basic_proxy_startup_tests/**/test_*.py")
echo "$TEST_FILES" | circleci tests run \
--verbose \
--command="tr ' ' '\\n' | awk '/\\.py/ {print; next} {sub(/\\.[A-Z][^.]*$/, \"\"); gsub(/\\./, \"/\"); print \$0 \".py\"}' | xargs uv run --no-sync python -m pytest --tb=short \
-vv \
--junitxml=test-results/junit-2.xml \
--durations=5"
no_output_timeout: 15m
# Clean up first container
- store_test_results:
path: test-results
- run:
name: Stop and remove first container
command: |
docker stop my-app || true
docker rm my-app || true
docker stop postgres-db || true
docker rm postgres-db || true
when: always
proxy_pass_through_endpoint_tests:
machine:
image: ubuntu-2204:2024.04.1
@ -3028,11 +2812,72 @@ jobs:
docker logs redis-cache > test-results/services-<< parameters.suite >>-<< parameters.mode >>/redis.log 2>&1 || true
docker rm -f postgres-db redis-cache
test -z "$(docker ps -aq --filter name=postgres-db --filter name=redis-cache)"
- when:
condition:
not:
equal: [1, << parameters.parallelism >>]
steps:
- run:
name: Prepare this shard's execution record
command: |
results="test-results/integration-<< parameters.suite >>"
if [ "<< parameters.mode >>" = replica ]; then
results="$results-replica"
fi
mkdir -p "/tmp/integration-shards/<< parameters.suite >>-<< parameters.mode >>/$CIRCLE_NODE_INDEX"
cp "$results/execution.json" "/tmp/integration-shards/<< parameters.suite >>-<< parameters.mode >>/$CIRCLE_NODE_INDEX/execution.json"
if [ "<< parameters.suite >>" = mcp ]; then
mkdir -p "/tmp/integration-shards/mcp-standard/$CIRCLE_NODE_INDEX/coverage"
cp test-results/integration-mcp/coverage/data "/tmp/integration-shards/mcp-standard/$CIRCLE_NODE_INDEX/coverage/data.$CIRCLE_NODE_INDEX"
fi
- persist_to_workspace:
root: /tmp/integration-shards
paths:
- << parameters.suite >>-<< parameters.mode >>
- store_test_results:
path: test-results
- store_artifacts:
path: test-results
integration_shard_coverage:
parameters:
suite:
type: enum
enum: [management, accounting, database, extensions, providers, mcp]
mode:
type: enum
enum: [standard, replica]
default: standard
shards:
type: integer
machine:
image: ubuntu-2204:2024.04.1
resource_class: medium
working_directory: ~/project
steps:
- setup_litellm_test_deps
- attach_workspace:
at: /tmp/integration-shards
- run:
name: Verify every << parameters.suite >> case ran exactly once
command: |
.venv/bin/python -m tests.integration._support.sharding /tmp/integration-shards/<< parameters.suite >>-<< parameters.mode >> \
--shards << parameters.shards >>
- when:
condition:
equal: [mcp, << parameters.suite >>]
steps:
- run:
name: Combine coverage from every MCP shard
command: |
mkdir -p test-results/integration-mcp/coverage
export COVERAGE_FILE="$PWD/test-results/integration-mcp/coverage/data"
.venv/bin/python -m coverage combine --rcfile=tests/integration/mcp_coverage.toml /tmp/integration-shards/mcp-standard/*/coverage
.venv/bin/python -m coverage report --rcfile=tests/integration/mcp_coverage.toml > test-results/integration-mcp/coverage/coverage.txt
.venv/bin/python -m coverage html --rcfile=tests/integration/mcp_coverage.toml -d test-results/integration-mcp/coverage/html
- store_artifacts:
path: test-results/integration-mcp/coverage
routing_parity:
parameters:
suite:
@ -3114,6 +2959,9 @@ jobs:
default: ""
timeout_minutes:
type: integer
workers:
type: integer
default: 1
machine:
image: ubuntu-2204:2024.04.1
resource_class: large
@ -3141,12 +2989,16 @@ jobs:
command: |
mkdir -p test-results
coverage_args=()
worker_args=()
if [ "<< parameters.workers >>" -gt 1 ]; then
worker_args=(-n << parameters.workers >> --dist=worksteal)
fi
if [ -n "<< parameters.coverage_flag >>" ]; then
coverage_args=(--cov=./litellm --cov-report=xml:coverage.xml)
fi
timeout --signal=TERM << parameters.timeout_minutes >>m uv run --no-sync pytest \
<< parameters.test_path >> -vv --tb=short --durations=10 -o junit_family=xunit1 \
--junitxml=test-results/junit.xml "${coverage_args[@]}"
--junitxml=test-results/junit.xml "${coverage_args[@]}" "${worker_args[@]}"
- when:
condition: << parameters.coverage_flag >>
steps:
@ -3244,8 +3096,8 @@ jobs:
timeout --signal=TERM 15m uv run --no-sync pytest \
tests/unit/test_redis.py tests/unit/caching/test_redis_connection_pool.py \
tests/unit/caching/test_redis_cluster_cache.py tests/unit/caching/test_evicted_client_closer.py \
tests/local_testing/test_caching.py::test_sync_cluster_authenticates_with_azure_credentials \
tests/local_testing/test_caching.py::test_sync_cluster_authenticates_with_gcp_credentials \
tests/integration/sdk/test_redis_cluster_iam_auth.py::test_sync_cluster_authenticates_with_azure_credentials \
tests/integration/sdk/test_redis_cluster_iam_auth.py::test_sync_cluster_authenticates_with_gcp_credentials \
--tb=short -vv --reruns 2 --reruns-delay 1 --durations=20 --cov=./litellm \
--cov-report=xml:coverage.xml -o junit_family=xunit1 --junitxml=test-results/junit.xml
- when:
@ -3323,21 +3175,93 @@ workflows:
name: integration-<< matrix.suite >>
matrix:
parameters:
suite: [management, accounting, database, providers, sdk, cost, security, browser]
suite: [sdk, cost, browser]
parallelism: 1
- integration_contracts:
name: integration-security
suite: security
parallelism: 1
resource_class: xlarge
- integration_contracts:
name: integration-management
suite: management
parallelism: 4
- integration_shard_coverage:
name: integration-management-coverage
suite: management
shards: 4
requires: [integration-management]
- integration_contracts:
name: integration-accounting
suite: accounting
parallelism: 4
- integration_shard_coverage:
name: integration-accounting-coverage
suite: accounting
shards: 4
requires: [integration-accounting]
- integration_contracts:
name: integration-providers
suite: providers
parallelism: 8
- integration_shard_coverage:
name: integration-providers-coverage
suite: providers
shards: 8
requires:
- integration-providers
- integration_contracts:
name: integration-database
suite: database
parallelism: 2
- integration_shard_coverage:
name: integration-database-coverage
suite: database
shards: 2
requires: [integration-database]
- integration_contracts:
name: integration-mcp
suite: mcp
resource_class: xlarge
parallelism: 2
- integration_shard_coverage:
name: integration-mcp-coverage
suite: mcp
shards: 2
requires:
- integration-mcp
- integration_contracts:
name: integration-extensions
suite: extensions
parallelism: 4
parallelism: 8
- integration_shard_coverage:
name: integration-extensions-coverage
suite: extensions
shards: 8
requires:
- integration-extensions
- integration_contracts:
name: integration-<< matrix.suite >>-replica
matrix:
parameters:
suite: [management, database]
mode: [replica]
name: integration-management-replica
suite: management
mode: replica
parallelism: 4
- integration_shard_coverage:
name: integration-management-replica-coverage
suite: management
mode: replica
shards: 4
requires: [integration-management-replica]
- integration_contracts:
name: integration-database-replica
suite: database
mode: replica
parallelism: 2
- integration_shard_coverage:
name: integration-database-replica-coverage
suite: database
mode: replica
shards: 2
requires: [integration-database-replica]
- postgres_suite:
name: proxy-behavior
test_path: tests/proxy_behavior
@ -3354,6 +3278,7 @@ workflows:
test_path: tests/proxy_migration_tests
seed: false
timeout_minutes: 20
workers: 2
- postgres_suite:
name: roi-database
test_path: tests/integration/database/test_roi_observed.py
@ -3371,6 +3296,8 @@ workflows:
- "6.4.0"
- "7.4.1"
- "8.0.1"
- windows_release_wheel
- base_sdk_install
build_and_test:
unless:
or:
@ -3378,10 +3305,7 @@ workflows:
- not:
equal: ["", << pipeline.parameters.routing_parity_base >>]
jobs:
- using_litellm_on_windows
- windows_release_wheel
- provider_replay_harness
- base_sdk_install
- local_testing_part1
- local_testing_part2
- langfuse_logging_unit_tests
@ -3405,7 +3329,6 @@ workflows:
- proxy_store_model_in_db_tests:
requires:
- build_docker_database_image
- proxy_build_from_pip_tests
- proxy_pass_through_endpoint_tests:
requires:
- build_docker_database_image
@ -3442,9 +3365,6 @@ workflows:
- db_migration_disable_update_check:
requires:
- build_docker_database_image
- installing_litellm_on_python
- installing_litellm_on_python_3_13
- installing_litellm_on_python_v2_migration_resolver
- helm_chart_testing:
requires:
- build_docker_database_image

View file

@ -15,7 +15,7 @@ outside_cost_map_set=false
while IFS= read -r file || [ -n "$file" ]; do
[ -n "$file" ] || continue
case "$file" in
litellm/_redis.py | litellm/_redis_credential_provider.py | litellm/caching/redis_cache.py | litellm/caching/evicted_client_closer.py | tests/unit/test_redis.py | tests/local_testing/test_caching.py | tests/unit/caching/test_redis_connection_pool.py | tests/unit/caching/test_redis_cluster_cache.py | tests/unit/caching/test_evicted_client_closer.py | .circleci/config.yml | .circleci/scripts/classify_changes.sh | .circleci/scripts/path_filter.sh | pyproject.toml | uv.lock)
litellm/_redis.py | litellm/_redis_credential_provider.py | litellm/caching/redis_cache.py | litellm/caching/evicted_client_closer.py | tests/unit/test_redis.py | tests/integration/sdk/test_redis_cluster_iam_auth.py | tests/unit/caching/test_redis_connection_pool.py | tests/unit/caching/test_redis_cluster_cache.py | tests/unit/caching/test_evicted_client_closer.py | .circleci/config.yml | .circleci/scripts/classify_changes.sh | .circleci/scripts/path_filter.sh | pyproject.toml | uv.lock)
has_redis_compat=true ;;
esac
case "$file" in

View file

@ -134,6 +134,9 @@ upstream_pid=$!
if [ "$suite" = cost ]; then
export INTEGRATION_WORKERS=8
fi
if [ "$suite" = security ]; then
export INTEGRATION_WORKERS=4
fi
if [ "$suite" = mcp ]; then
export INTEGRATION_WORKERS=4 INTEGRATION_COVERAGE=1
fi
@ -213,13 +216,11 @@ if [ "$suite" = browser ]; then
exit 0
fi
node_files=()
shard_arguments=()
if [ "${CIRCLE_NODE_TOTAL:-1}" -gt 1 ]; then
split="$(.venv/bin/python tests/integration/run.py "$suite" --list \
| circleci tests split --split-by=timings --timings-type=filename)"
read -r -a node_files <<< "$(printf '%s' "$split" | tr '\n' ' ')"
test "${#node_files[@]}" -gt 0
printf '%s\n' "${node_files[@]}" > "$results/node-files.txt"
.venv/bin/python tests/integration/run.py "$suite" --list \
| circleci tests split --split-by=timings --timings-type=filename > "$results/node-files.txt"
shard_arguments=(--shard-files "$results/node-files.txt")
fi
env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" \
@ -232,11 +233,13 @@ env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" \
LITELLM_LICENSE="${LITELLM_LICENSE:-}" \
INTEGRATION_SEED="$INTEGRATION_SEED" \
INTEGRATION_ORDER_SEED="$INTEGRATION_ORDER_SEED" \
INTEGRATION_CASE_SHARDS="${CIRCLE_NODE_TOTAL:-1}" INTEGRATION_CASE_SHARD_INDEX="${CIRCLE_NODE_INDEX:-0}" \
LITELLM_LOCAL_MODEL_COST_MAP=True AWS_EC2_METADATA_DISABLED=true DO_NOT_TRACK=1 \
INTEGRATION_PROXY_DATABASE_URL="$INTEGRATION_PROXY_DATABASE_URL" \
INTEGRATION_PROXY_READ_REPLICA_URL="$INTEGRATION_PROXY_READ_REPLICA_URL" \
INTEGRATION_ROUTING="$INTEGRATION_ROUTING" \
.venv/bin/python tests/integration/run.py "$suite" --results "$results" "${node_files[@]}"
.venv/bin/python tests/integration/run.py "$suite" --results "$results" \
"${shard_arguments[@]}"
if [ "${INTEGRATION_COVERAGE:-0}" = 1 ]; then
for covered_pid in "$proxy_pid" "$peer_pid"; do

Binary file not shown.

Before

Width:  |  Height:  |  Size: 248 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 128 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 101 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 132 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 184 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 259 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 94 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 94 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 77 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 79 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 61 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 86 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 75 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 40 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 88 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 95 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 96 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 94 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 70 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 82 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 82 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 64 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 90 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 80 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 58 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 63 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 70 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 47 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 76 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 75 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 39 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 70 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 93 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 73 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 39 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 81 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 72 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 76 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 50 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 59 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 57 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 56 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 65 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 55 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 108 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 109 KiB

View file

@ -21,17 +21,13 @@ test_paths:
Live-provider caching cases in tests/local_testing that remain outside CI. Jobs that
glob that directory either deselect them (local_testing_part1 and part2 carry `-k "... and
not caching and not cache"`) or keep only another keyword (langfuse, router, assistants).
Separately, the CircleCI redis-compat jobs select two IAM cluster authentication tests in
test_caching.py by node ID. It does not run that file's other tests.
The gap was eight files and 118 tests when measured 2026-08-20; the five keyless files now
run in the caching-local shard, leaving live cases in these three. Measured 2026-08-21 with no provider
credentials and no Redis: test_caching.py needs both (37 of 65 fail without them),
test_disk_cache_unit_tests.py needs OPENAI_API_KEY for 2 of its 4, and
test_gcs_cache_unit_tests.py needs GCS credentials for all 4. They want the keyless/live
split that porting tests/local_testing off CircleCI will force, not a job that is red by
construction
The two Redis Cluster IAM authentication tests moved to tests/integration/sdk and remain
selected by node ID in redis_compat, where redis-server is installed.
The five keyless files now run in the caching-local shard, leaving live cases in these two.
Measured 2026-08-21 with no provider credentials: test_disk_cache_unit_tests.py needs
OPENAI_API_KEY for 2 of its 4 tests, and test_gcs_cache_unit_tests.py needs GCS credentials
for all 4. Keep these live cases outside PR CI rather than making a job red by construction
paths:
- tests/local_testing/test_caching.py
- tests/local_testing/test_disk_cache_unit_tests.py
- tests/local_testing/test_gcs_cache_unit_tests.py
- reason: >-

View file

@ -9,6 +9,8 @@ import sys
import warnings
from collections.abc import Callable, Iterable, Mapping, Sequence
from dataclasses import dataclass
from functools import lru_cache
from types import MappingProxyType
from typing import Final
import yaml
@ -316,6 +318,7 @@ def _keyword_terms(
def _slices() -> tuple[Slice, ...]:
if not CIRCLECI_CONFIG.exists():
return ()
test_files: Final = frozenset(_test_files())
jobs: Final = yaml.safe_load(CIRCLECI_CONFIG.read_text()).get("jobs", {})
return tuple(
Slice(job=job, globs=globs, named=named, required=required, excluded=excluded, understood=understood)
@ -323,7 +326,7 @@ def _slices() -> tuple[Slice, ...]:
for text in ("\n".join(_strings(body)),)
if "pytest" in text
for globs in (tuple(GLOB_CALL_RE.findall(text)),)
for named in (frozenset(TEST_TOKEN_RE.findall(text)) & frozenset(_test_files()),)
for named in (frozenset(TEST_TOKEN_RE.findall(text)) & test_files,)
for required, excluded, understood in (
_keyword_terms(tuple(KEYWORD_RE.findall(text)), attributable=len(globs) < 2),
)
@ -363,6 +366,7 @@ def _workflow_named_tokens() -> frozenset[str]:
def _deselected_everywhere(allowlist: Allowlist) -> tuple[Finding, ...]:
matchable_names: Final = lru_cache(maxsize=None)(_matchable_names)
slices: Final = _slices()
named_by_workflow: Final = _workflow_named_tokens()
globbed: Final = tuple(
@ -376,7 +380,7 @@ def _deselected_everywhere(allowlist: Allowlist) -> tuple[Finding, ...]:
for path in globbed
if not allowlist.covers_test(path)
and not any(_token_covers(token, path) for token in named_by_workflow)
and not any(slice_.claims(path, _matchable_names(path)) for slice_ in slices)
and not any(slice_.claims(path, matchable_names(path)) for slice_ in slices)
)

View file

@ -91,9 +91,158 @@ jobs:
if: always()
run: sudo ip netns delete smoke
all-extras:
name: all-extras-py${{ matrix.python-version }}
runs-on: ubuntu-24.04
timeout-minutes: 10
strategy:
fail-fast: false
matrix:
python-version: ["3.12", "3.13"]
env:
LITELLM_LOCAL_MODEL_COST_MAP: "True"
steps:
- name: Checkout
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
persist-credentials: false
- name: Set up Python
uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: ${{ matrix.python-version }}
- name: Set up uv
uses: ./.github/actions/setup-uv-with-retries
with:
version: "0.10.9"
- name: Install every extra and dependency group
run: .github/scripts/uv_sync_with_retries.sh --frozen --all-groups --all-extras --python ${{ matrix.python-version }}
- name: Import the SDK, the proxy server and the enterprise package
run: .venv/bin/python -c "import litellm, litellm.proxy.proxy_server, litellm_enterprise"
windows-sdk:
name: windows-sdk
runs-on: windows-2025
timeout-minutes: 10
env:
LITELLM_LOCAL_MODEL_COST_MAP: "True"
defaults:
run:
shell: bash
steps:
- name: Checkout
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
persist-credentials: false
- name: Set up Python
uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: "3.11"
- name: Set up uv
uses: ./.github/actions/setup-uv-with-retries
with:
version: "0.10.9"
- name: Install dependencies
run: .github/scripts/uv_sync_with_retries.sh --frozen --no-install-workspace --group dev --python 3.11
- name: Run the Windows tests
run: uv run --no-sync python -m pytest --tb=short -v tests/windows_tests
proxy-build-from-pip:
name: proxy-build-from-pip
runs-on: ubuntu-24.04
timeout-minutes: 10
services:
postgres:
image: postgres:14@sha256:6a70deda415ec296f977890e11aba04a0db9f632a362e3fce45e845e3db74f26
env:
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
POSTGRES_DB: circle_test
ports:
- 5432:5432
options: >-
--health-cmd "pg_isready -U postgres"
--health-interval 2s
--health-timeout 5s
--health-retries 30
steps:
- name: Checkout
uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
with:
persist-credentials: false
- name: Set up Python
uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
with:
python-version: "3.12"
- name: Set up uv
uses: ./.github/actions/setup-uv-with-retries
with:
version: "0.10.9"
- name: Install the test client dependencies
run: .github/scripts/uv_sync_with_retries.sh --frozen --no-install-workspace --extra proxy --group dev --python 3.12
- name: Build the image from the published package
run: docker build -t litellm-build-from-pip:smoke -f docker/build_from_pip/Dockerfile.build_from_pip .
- name: Start the fake OpenAI endpoint
run: |
nohup .venv/bin/python tests/_fake_openai_endpoint_server.py --host 0.0.0.0 --port 8190 > "${RUNNER_TEMP}/fake-openai.log" 2>&1 &
for _ in $(seq 1 30); do
curl -sf http://localhost:8190/health > /dev/null && exit 0
sleep 1
done
cat "${RUNNER_TEMP}/fake-openai.log"
exit 1
- name: Generate a master key
run: |
key="sk-$(openssl rand -hex 16)"
echo "::add-mask::${key}"
echo "LITELLM_MASTER_KEY=${key}" >> "${GITHUB_ENV}"
- name: Start the proxy container
run: |
docker run -d \
-p 4000:4000 \
-e DATABASE_URL=postgresql://postgres:postgres@host.docker.internal:5432/circle_test \
-e LITELLM_MASTER_KEY \
-e FAKE_OPENAI_API_BASE=http://host.docker.internal:8190 \
-e OTEL_EXPORTER=in_memory \
-e USE_DDTRACE=True \
-e GCS_FLUSH_INTERVAL=1 \
-e LITELLM_LOG=ERROR \
--add-host host.docker.internal:host-gateway \
--name litellm-build-from-pip \
-v "${GITHUB_WORKSPACE}/docker/build_from_pip/litellm_config.yaml:/app/config.yaml" \
litellm-build-from-pip:smoke \
--config /app/config.yaml \
--port 4000
for _ in $(seq 1 150); do
curl -sf --max-time 5 -o /dev/null http://localhost:4000 && exit 0
sleep 2
done
exit 1
- name: Run the startup tests
run: .venv/bin/python -m pytest --tb=short -vv tests/basic_proxy_startup_tests
- name: Show the proxy logs
if: always()
run: docker logs litellm-build-from-pip || true
smoke-passed:
name: smoke passed
needs: [dashboard-build, core-checks]
needs: [dashboard-build, core-checks, all-extras, windows-sdk, proxy-build-from-pip]
if: always()
runs-on: ubuntu-latest
timeout-minutes: 2

View file

@ -327,6 +327,7 @@ jobs:
tests/unit/llms/sagemaker
tests/unit/llms/sail
tests/unit/llms/sambanova
tests/unit/llms/scaledown
tests/unit/llms/sap
tests/unit/llms/scaleway
tests/unit/llms/searchapi
@ -927,6 +928,7 @@ jobs:
timeout-minutes: 3
with:
persist-credentials: false
fetch-depth: ${{ matrix.shard == 'db-and-spend' && '0' || '1' }}
- name: Detect relevant changes
id: changes
@ -959,11 +961,13 @@ jobs:
with:
artifact: ${{ needs.rust-bridge.outputs.artifact }}
- name: Install dependencies
- name: Install dependencies and check schema
if: steps.changes.outputs.decision != 'skip'
timeout-minutes: 8
env:
RUST_BRIDGE_ARTIFACT: ${{ needs.rust-bridge.outputs.artifact }}
UNIT_SHARD: ${{ matrix.shard }}
SCHEMA_BASE_REF: origin/${{ github.event.repository.default_branch }}
run: |
diff -u model_prices_and_context_window.json litellm/model_prices_and_context_window_backup.json
if [ -z "$RUST_BRIDGE_ARTIFACT" ]; then
@ -975,6 +979,9 @@ jobs:
uv run --no-sync python -c "import importlib.metadata; import litellm.rust_bridge._native; print(importlib.metadata.version('litellm'))"
fi
uv run --no-sync python -c 'import os, sys; print(sys.version); assert f"{sys.version_info.major}.{sys.version_info.minor}" == os.environ["UV_PYTHON"]'
if [ "$UNIT_SHARD" = "db-and-spend" ]; then
uv run --no-sync python scripts/check_schema_compatibility.py --base-ref "$SCHEMA_BASE_REF"
fi
- name: Cache Prisma binaries
if: steps.changes.outputs.decision != 'skip'

View file

@ -407,6 +407,7 @@ Set `LITELLM_PROXY_API_BASE` and `LITELLM_PROXY_API_KEY` and every model call th
| [Sagemaker Chat (`sagemaker_chat`)](https://docs.litellm.ai/docs/providers/aws_sagemaker) | ✅ | ✅ | ✅ | | | | | | | |
| [Sail (`sail`)](https://docs.litellm.ai/docs/providers/sail) | ✅ | ✅ | ✅ | | | | | | | |
| [Sambanova (`sambanova`)](https://docs.litellm.ai/docs/providers/sambanova) | ✅ | ✅ | ✅ | | | | | | | |
| [ScaleDown (`scaledown`)](https://docs.litellm.ai/docs/providers/scaledown) | ✅ | | | | | | | | | |
| [Snowflake (`snowflake`)](https://docs.litellm.ai/docs/providers/snowflake) | ✅ | ✅ | ✅ | | | | | | | |
| [Strands Decider (`strands_decider`)](https://docs.litellm.ai/docs/providers) | | | | | | | | | | |
| [Text Completion Codestral (`text-completion-codestral`)](https://docs.litellm.ai/docs/providers/codestral) | ✅ | ✅ | ✅ | | | | | | | |

View file

@ -1,4 +1,5 @@
import random
import re
import time
from collections.abc import Generator, Mapping
from contextlib import contextmanager
@ -7,25 +8,50 @@ from typing import TYPE_CHECKING, Final
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
from litellm_proxy_extras._logging import logger
from litellm_proxy_extras.prisma_toolchain import MIGRATION_LOCK_TIMEOUT_ENV_VAR, migration_lock_timeout
from litellm_proxy_extras.prisma_toolchain import (
MIGRATION_LOCK_TIMEOUT_ENV_VAR,
migration_ddl_lock_timeout,
migration_lock_timeout,
)
MIGRATION_LOCK_KEY: Final = int.from_bytes(b"llm_mig2", "big")
_LOCK_TIMEOUT_PINNED_RE: Final = re.compile(r"(?:-c\s*|--)lock_timeout=")
if TYPE_CHECKING:
import psycopg
def migration_environment(environment: Mapping[str, str]) -> Mapping[str, str]:
database_url: Final = environment.get("DATABASE_URL")
direct_url: Final = environment.get("DIRECT_URL")
if not database_url or not direct_url:
return environment
def _migration_url(database_url: str, direct_url: "str | None") -> str:
if not direct_url:
return database_url
schema: Final = next((value for key, value in parse_qsl(urlsplit(database_url).query) if key == "schema"), "public")
direct: Final = urlsplit(direct_url)
parameters: Final = tuple((key, value) for key, value in parse_qsl(direct.query) if key != "schema")
return urlunsplit(direct._replace(query=urlencode((*parameters, ("schema", schema)))))
def _with_ddl_lock_timeout(url: str) -> str:
split: Final = urlsplit(url)
pairs: Final = tuple(parse_qsl(split.query))
if any(key == "pgbouncer" and value == "true" for key, value in pairs):
return url
existing: Final = next((value for key, value in pairs if key == "options"), "")
if _LOCK_TIMEOUT_PINNED_RE.search(existing):
return url
timeout_ms: Final = max(1, int(migration_ddl_lock_timeout() * 1000))
options: Final = f"{existing} -c lock_timeout={timeout_ms}".strip()
rewritten: Final = tuple((key, value) for key, value in pairs if key != "options") + (("options", options),)
return urlunsplit(split._replace(query=urlencode(rewritten)))
def migration_environment(environment: Mapping[str, str]) -> Mapping[str, str]:
database_url: Final = environment.get("DATABASE_URL")
if not database_url:
return environment
migration_url: Final = _migration_url(database_url, environment.get("DIRECT_URL"))
return {
**environment,
"DATABASE_URL": urlunsplit(direct._replace(query=urlencode((*parameters, ("schema", schema))))),
"DATABASE_URL": _with_ddl_lock_timeout(migration_url),
}

View file

@ -60,11 +60,13 @@ PRISMA_COMMAND_TIMEOUT_ENV_VAR = "LITELLM_PRISMA_COMMAND_TIMEOUT"
PRISMA_BOOTSTRAP_TIMEOUT_ENV_VAR = "LITELLM_PRISMA_BOOTSTRAP_TIMEOUT"
PRISMA_MIGRATE_DEPLOY_TIMEOUT_ENV_VAR = "LITELLM_PRISMA_MIGRATE_DEPLOY_TIMEOUT"
MIGRATION_LOCK_TIMEOUT_ENV_VAR = "LITELLM_MIGRATION_LOCK_TIMEOUT"
MIGRATION_DDL_LOCK_TIMEOUT_ENV_VAR = "LITELLM_MIGRATION_DDL_LOCK_TIMEOUT"
NODEENV_CACHE_DIR_ENV_VAR = "PRISMA_NODEENV_CACHE_DIR"
DEFAULT_PRISMA_COMMAND_TIMEOUT = 60.0
DEFAULT_PRISMA_BOOTSTRAP_TIMEOUT = 600.0
DEFAULT_PRISMA_MIGRATE_DEPLOY_TIMEOUT = 600.0
DEFAULT_MIGRATION_DDL_LOCK_TIMEOUT = 10.0
BOOTSTRAP_ARG = "--version"
PRISMA_CONSOLE_SCRIPT = "prisma"
@ -111,6 +113,11 @@ def migration_lock_timeout() -> float:
return _timeout_from_env(MIGRATION_LOCK_TIMEOUT_ENV_VAR, 600.0)
def migration_ddl_lock_timeout() -> float:
"""Seconds one migration statement may wait on a table lock before Postgres cancels it."""
return _timeout_from_env(MIGRATION_DDL_LOCK_TIMEOUT_ENV_VAR, DEFAULT_MIGRATION_DDL_LOCK_TIMEOUT)
def prisma_bootstrap_timeout() -> float:
"""Seconds the one-time Node toolchain install may run for."""
return _timeout_from_env(

View file

@ -18,6 +18,7 @@ from litellm_proxy_extras import prisma_toolchain
from litellm_proxy_extras._logging import logger
from litellm_proxy_extras.migration_lock import held_migration_lock
from litellm_proxy_extras.prisma_toolchain import (
MIGRATION_DDL_LOCK_TIMEOUT_ENV_VAR,
PRISMA_COMMAND_TIMEOUT_ENV_VAR,
PRISMA_MIGRATE_DEPLOY_TIMEOUT_ENV_VAR,
ensure_prisma_toolchain,
@ -62,6 +63,7 @@ def _get_prisma_env() -> dict:
_MIGRATION_TS_RE = re.compile(r"^(\d{14})_")
_MIGRATION_DEADLOCK_MARKER = "deadlock detected"
_MIGRATION_LOCK_TIMEOUT_MARKER = "canceling statement due to lock timeout"
INDEX_REPAIR_ADVISORY_LOCK_KEY: Final = int.from_bytes(b"litellm", "big")
_TRANSIENT_INDEX_SUFFIX_RE: Final = re.compile(r"_cc(?:new|old)\d*$")
_INVALID_LITELLM_INDEXES_SQL: Final = (
@ -1162,9 +1164,11 @@ class ProxyExtrasDBManager:
raise RuntimeError(
f"Database migration failed after {MAX_MIGRATE_DEPLOY_ATTEMPTS} "
"attempts that made no progress (timeouts or deadlock retries). Check database connectivity, "
"load, and _prisma_migrations ledger state, and raise "
f"{PRISMA_MIGRATE_DEPLOY_TIMEOUT_ENV_VAR} if the attempts timed out."
"attempts that made no progress (timeouts, lock timeouts or deadlock retries). "
"Check database connectivity, load, long-running transactions on the migrated tables, "
"and _prisma_migrations ledger state. Raise "
f"{PRISMA_MIGRATE_DEPLOY_TIMEOUT_ENV_VAR} if the attempts timed out, or "
f"{MIGRATION_DDL_LOCK_TIMEOUT_ENV_VAR} if they timed out waiting for a table lock."
)
finally:
os.chdir(original_dir)
@ -1225,6 +1229,13 @@ class ProxyExtrasDBManager:
)
ProxyExtrasDBManager._v2_roll_back_migration_best_effort(migration_name)
return budget.spend()
if ledger_logs and _MIGRATION_LOCK_TIMEOUT_MARKER in ledger_logs:
logger.info(
"Migration %s timed out waiting for a table lock, rolling its ledger row back and retrying",
migration_name,
)
ProxyExtrasDBManager._v2_roll_back_migration_best_effort(migration_name)
return budget.spend()
if ProxyExtrasDBManager._failed_migration_recovered(migration_name, started_at):
logger.info(
"Migration %s started at %s was already rolled back or completed by a concurrent "
@ -1270,6 +1281,16 @@ class ProxyExtrasDBManager:
ProxyExtrasDBManager._v2_roll_back_migration_best_effort(migration_name)
return budget.spend()
if migration_name and _MIGRATION_LOCK_TIMEOUT_MARKER in stderr:
ProxyExtrasDBManager._log_migration_lock_holders(migration_name)
logger.warning(
"Migration %s timed out waiting for a table lock, rolling its ledger row back and retrying. "
f"Raise {MIGRATION_DDL_LOCK_TIMEOUT_ENV_VAR} if the database needs longer.",
migration_name,
)
ProxyExtrasDBManager._v2_roll_back_migration_best_effort(migration_name)
return budget.spend()
raise RuntimeError(
"Database migration failed and cannot be auto-recovered. "
f"Manual intervention required.\n\nPrisma error:\n{stderr}"
@ -1282,6 +1303,13 @@ class ProxyExtrasDBManager:
)
return budget.spend()
if _MIGRATION_LOCK_TIMEOUT_MARKER in stderr:
logger.info(
"Waiting for the advisory lock held by another Prisma migration; "
"contention does not spend a migration failure attempt"
)
return budget.after_contention(attempt_seconds)
if "P1002" in stderr and "advisory lock" in stderr:
logger.info(
"Waiting for the advisory lock held by another Prisma migration; "
@ -1294,6 +1322,68 @@ class ProxyExtrasDBManager:
f"Manual intervention required.\n\nPrisma error:\n{stderr}"
) from error
@staticmethod
def _log_migration_lock_holders(migration_name: str) -> None:
"""Best-effort log of the sessions holding locks on the tables a timed-out migration touches."""
try:
migration_sql: Final = (
Path(ProxyExtrasDBManager._get_prisma_dir()) / "migrations" / migration_name / "migration.sql"
).read_text()
except OSError:
return
relations: Final = sorted(set(re.findall(r'"(LiteLLM_\w+)"', migration_sql)))
database_url: Final = os.getenv("DATABASE_URL")
if not relations or not database_url:
return
try:
import psycopg
except ImportError:
return
try:
with psycopg.connect(
ProxyExtrasDBManager._strip_prisma_query_params(database_url),
connect_timeout=10,
autocommit=True,
options="-c statement_timeout=5000",
) as conn:
rows: Final = conn.execute(
"SELECT l.pid, c.relname, l.mode, a.state, a.application_name, "
"date_trunc('second', now() - a.xact_start) "
"FROM pg_locks l "
"JOIN pg_class c ON c.oid = l.relation "
"JOIN pg_namespace n ON n.oid = c.relnamespace "
"JOIN pg_stat_activity a ON a.pid = l.pid "
"WHERE l.granted AND n.nspname = %s AND c.relname = ANY(%s) AND l.pid <> pg_backend_pid() "
"ORDER BY a.xact_start NULLS LAST "
"LIMIT 10",
(
ProxyExtrasDBManager._prisma_schema_param(database_url) or "public",
relations,
),
).fetchall()
except psycopg.Error:
return
if not rows:
logger.warning(
"Migration %s timed out waiting for a lock, but no current lock holder "
"was found on %s; the holder has likely since committed or rolled back",
migration_name,
", ".join(relations),
)
return
for pid, relname, mode, state, application_name, age in rows:
logger.warning(
"Migration %s timed out waiting for a lock on %s held by pid %s "
"(mode %s, state %s, application %s, transaction open for %s)",
migration_name,
relname,
pid,
mode,
state,
application_name,
age,
)
@staticmethod
def _mark_migration_applied(name: str) -> None:
"""Roll a failed ledger row back if it is still there, then mark it applied."""

View file

@ -674,6 +674,7 @@ nvidia_nim_models: Set = set()
nvidia_riva_models: Set = set()
soniox_models: Set = set()
sambanova_models: Set = set()
scaledown_models: Final[Set[str]] = set() # Price reloads update this registry through existing references.
sambanova_embedding_models: Set = set()
novita_models: Set = set()
assemblyai_models: Set = set()
@ -756,7 +757,7 @@ def is_openai_finetune_model(key: str) -> bool:
return key.startswith("ft:") and not key.count(":") > 1
def _populate_provider_model_sets(model_cost_map: Dict) -> None:
def _populate_provider_model_sets(model_cost_map: Mapping[str, Mapping[str, object]]) -> None:
for key, value in model_cost_map.items():
if value.get("litellm_provider") == "openai" and not is_openai_finetune_model(key):
open_ai_chat_completion_models.add(key)
@ -918,6 +919,8 @@ def _populate_provider_model_sets(model_cost_map: Dict) -> None:
soniox_models.add(key)
elif value.get("litellm_provider") == "sambanova":
sambanova_models.add(key)
elif value.get("litellm_provider") == "scaledown":
scaledown_models.add(key)
elif value.get("litellm_provider") == "sambanova-embedding-models":
sambanova_embedding_models.add(key)
elif value.get("litellm_provider") == "novita":
@ -1106,6 +1109,7 @@ model_list = list(
| nvidia_riva_models
| soniox_models
| sambanova_models
| scaledown_models
| azure_text_models
| novita_models
| assemblyai_models
@ -1215,6 +1219,7 @@ def _build_models_by_provider() -> dict:
"nvidia_riva": nvidia_riva_models,
"soniox": soniox_models,
"sambanova": sambanova_models | sambanova_embedding_models,
"scaledown": scaledown_models,
"novita": novita_models,
"nebius": nebius_models | nebius_embedding_models,
"aiml": aiml_models,
@ -2054,6 +2059,9 @@ if TYPE_CHECKING:
from .llms.featherless_ai.chat.transformation import (
FeatherlessAIConfig as FeatherlessAIConfig,
)
from .llms.scaledown.chat.transformation import (
ScaleDownChatConfig as ScaleDownChatConfig,
)
from .llms.cerebras.chat import CerebrasConfig as CerebrasConfig
from .llms.nadir.chat.transformation import NadirConfig as NadirConfig
from .llms.baseten.chat import BasetenConfig as BasetenConfig

View file

@ -277,6 +277,7 @@ LLM_CONFIG_NAMES: Final = (
"NvidiaNimConfig",
"NvidiaNimEmbeddingConfig",
"FeatherlessAIConfig",
"ScaleDownChatConfig",
"CerebrasConfig",
"NadirConfig",
"BasetenConfig",
@ -1103,6 +1104,10 @@ _LLM_CONFIGS_IMPORT_MAP: Final = {
".llms.featherless_ai.chat.transformation",
"FeatherlessAIConfig",
),
"ScaleDownChatConfig": (
".llms.scaledown.chat.transformation",
"ScaleDownChatConfig",
),
"CerebrasConfig": (".llms.cerebras.chat", "CerebrasConfig"),
"NadirConfig": (".llms.nadir.chat.transformation", "NadirConfig"),
"BasetenConfig": (".llms.baseten.chat", "BasetenConfig"),

View file

@ -797,6 +797,8 @@ class RedisCache(BaseCache):
def init_async_client(
self,
) -> async_redis_client | async_redis_cluster_client:
from redis.asyncio import RedisCluster
from litellm import in_memory_llm_clients_cache
from .._redis import get_redis_async_client, get_redis_connection_pool
@ -809,7 +811,11 @@ class RedisCache(BaseCache):
# Create new connection pool and client for current event loop
self.async_redis_conn_pool = get_redis_connection_pool(**self.redis_kwargs)
redis_async_client = get_redis_async_client(connection_pool=self.async_redis_conn_pool, **self.redis_kwargs)
in_memory_llm_clients_cache.set_cache(key=cache_key, value=redis_async_client)
in_memory_llm_clients_cache.set_cache(
key=cache_key,
value=redis_async_client,
litellm_owned_client=isinstance(redis_async_client, RedisCluster),
)
self.redis_async_client = redis_async_client
return redis_async_client

View file

@ -11,7 +11,7 @@ from litellm.caching.redis_cache import RedisCache
if TYPE_CHECKING:
from opentelemetry.trace import Span as _Span
from redis.asyncio import Redis, RedisCluster
from redis.asyncio import Redis
from redis.asyncio.client import Pipeline
pipeline = Pipeline
@ -26,22 +26,6 @@ else:
class RedisClusterCache(RedisCache):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.redis_async_redis_cluster_client: RedisCluster | None = None
self.redis_sync_redis_cluster_client: RedisCluster | None = None
def init_async_client(self):
from redis.asyncio import RedisCluster
from .._redis import get_redis_async_client
if self.redis_async_redis_cluster_client:
return self.redis_async_redis_cluster_client
_redis_client: Final = get_redis_async_client(connection_pool=self.async_redis_conn_pool, **self.redis_kwargs)
if isinstance(_redis_client, RedisCluster):
self.redis_async_redis_cluster_client = _redis_client
return _redis_client
def _run_redis_mget_operation(self, keys: list[str]) -> list[Any]:
"""

View file

@ -387,6 +387,8 @@ BEDROCK_REALTIME_SDK_DISTRIBUTION: Final = "aws-sdk-bedrock-runtime"
BEDROCK_REALTIME_SDK_SUPPORTED_RANGE: Final = ">=0.10.0,<0.12.0"
CLIENT_REQUESTED_MODEL_SCOPE_KEY: Final = "litellm.client_requested_model"
MODEL_GROUP_ALIAS_RESOLVED_SCOPE_KEY: Final = "litellm.model_group_alias_resolved"
MCP_PEEKED_BODY_SCOPE_KEY: Final = "litellm_mcp_peeked_body"
MCP_ADMISSION_BODY_PEEK_TIMEOUT_SECONDS: Final = 5.0
REALTIME_SESSION_SUCCESS_LOGGED_KEY: Final = "realtime_session_success_logged"
REALTIME_SESSION_FAILURE_LOGGED_KEY: Final = "realtime_session_failure_logged"
@ -823,6 +825,7 @@ LITELLM_CHAT_PROVIDERS: Final = [
"lemonade",
"docker_model_runner",
"amazon_nova",
"scaledown",
]
# Resolving these providers runs an OAuth device flow (their provider info IS the login), so any
@ -1967,6 +1970,7 @@ LITELLM_SETTINGS_SAFE_DB_OVERRIDES: Final = [
]
SPECIAL_LITELLM_AUTH_TOKEN: Final = ["ui-token"]
DEFAULT_MANAGEMENT_OBJECT_IN_MEMORY_CACHE_TTL = int(os.getenv("DEFAULT_MANAGEMENT_OBJECT_IN_MEMORY_CACHE_TTL", 60))
AUTH_OBJECTS_TARGET: Final = "auth_objects"
DEFAULT_ACCESS_GROUP_CACHE_TTL: Final = int(os.getenv("DEFAULT_ACCESS_GROUP_CACHE_TTL", 600))
SPEND_LOG_KEY_METADATA_CACHE_TTL: Final = 600
SPEND_LOG_KEY_METADATA_MISS_CACHE_TTL: Final = 30
@ -2215,6 +2219,8 @@ USAGE_TOP_API_KEYS_DEFAULT: Final[int] = 100
USAGE_TOP_API_KEYS_MAX: Final[int] = 1000
USAGE_KEY_PAGE_DEFAULT: Final[int] = 50
USAGE_KEY_PAGE_MAX: Final[int] = 100
USAGE_USER_PAGE_DEFAULT: Final[int] = 50
USAGE_USER_PAGE_MAX: Final[int] = 100
USAGE_KEY_SEARCH_DEFAULT: Final[int] = 100
USAGE_KEY_SEARCH_MAX: Final[int] = 100
USAGE_MODEL_TOP_KEYS_DEFAULT: Final[int] = 5

View file

@ -1,7 +1,7 @@
from abc import ABC, abstractmethod
from typing import Final, Literal
from litellm.proxy._types import CallInfo, Litellm_EntityType
from litellm.types.integrations.slack_alerting import CallInfo, Litellm_EntityType
class BaseBudgetAlertType(ABC):

View file

@ -38,20 +38,20 @@ from litellm.llms.custom_httpx.http_handler import (
get_async_httpx_client,
httpxSpecialProvider,
)
from litellm.proxy._types import (
CallInfo,
InvitationModel,
InvitationNew,
Litellm_EntityType,
UserAPIKeyAuth,
VirtualKeyEvent,
WebhookEvent,
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.repositories.table_repositories import InvitationLinkRepository
from litellm.repositories.team_repository import TeamRepository
from litellm.repositories.user_repository import UserRepository
from litellm.types.integrations.slack_alerting import *
from litellm.types.integrations.slack_alerting import AlertType
from litellm.types.integrations.slack_alerting import (
AlertType,
CallInfo,
InvitationModel,
InvitationNew,
Litellm_EntityType,
VirtualKeyEvent,
WebhookEvent,
)
from litellm.types.proxy.model_deprecation import (
DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS,
DEPRECATION_IDLE_POLL_SECONDS,

View file

@ -223,6 +223,7 @@ class CustomGuardrail(CustomLogger):
records_own_guardrail_information: ClassVar[bool] = False
logging_only_scope: LoggingOnlyScope | None
logging_only_continue_on_input_failure: bool
stream_scope_default: GuardrailStreamScope = DEFAULT_GUARDRAIL_STREAM_SCOPE
stream_scope_by_hook: tuple[tuple[str, GuardrailStreamScope], ...] = ()
@ -305,6 +306,7 @@ class CustomGuardrail(CustomLogger):
stream_scope_arg: Final[object] = cast(object, kwargs.pop("stream_scope", None)) # cast-ok: config
self.apply_stream_scope(stream_scope_arg)
self.logging_only_scope = None
self.logging_only_continue_on_input_failure = False
if timeout is not None:
self.timeout = timeout
@ -1064,7 +1066,7 @@ class CustomGuardrail(CustomLogger):
except Exception as e:
if self.logging_only_scope == "output":
return None
if self.logging_only_scope == "both":
if self.logging_only_continue_on_input_failure:
verbose_logger.warning(
"Guardrail %s: logging_only request copy failed, skipping request scan: %s",
self.guardrail_name,
@ -1092,7 +1094,7 @@ class CustomGuardrail(CustomLogger):
"metadata": scratch_metadata,
}
if self.logging_only_scope != "output" and scratch_fields is not None:
if self.logging_only_scope == "both":
if self.logging_only_continue_on_input_failure:
try:
await translation.process_input_messages(data=scratch_request, guardrail_to_apply=self)
except Exception as e: # noqa: BLE001 # one direction's scan failure must not drop the other direction's verdict

View file

@ -33,6 +33,10 @@ from litellm.integrations.datadog.datadog_mock_client import (
should_use_datadog_mock,
)
from litellm.litellm_core_utils.dd_tracing import tracer
from litellm.litellm_core_utils.llm_cost_calc.cache_tokens import (
extract_cache_creation_tokens,
extract_cache_read_tokens,
)
from litellm.litellm_core_utils.prompt_templates.common_utils import (
convert_content_list_to_str,
handle_any_messages_to_chat_completion_str_messages_conversion,
@ -44,7 +48,6 @@ from litellm.llms.custom_httpx.http_handler import (
get_async_httpx_client,
httpxSpecialProvider,
)
from litellm.proxy.spend_tracking.savings import extract_cache_creation_tokens, extract_cache_read_tokens
from litellm.types.integrations.datadog_llm_obs import *
from litellm.types.utils import (
AUDIT_GUARDRAIL_FIELDS,

View file

@ -6,8 +6,8 @@ import os
from typing import Final
from litellm._logging import verbose_logger, verbose_proxy_logger
from litellm.proxy._types import WebhookEvent
from litellm.repositories.team_repository import TeamRepository
from litellm.types.integrations.slack_alerting import WebhookEvent
# we use this for the email header, please send a test email if you change this. verify it looks good on email
LITELLM_LOGO_URL: Final = "https://litellm-listing.s3.amazonaws.com/litellm_logo.png"

View file

@ -26,8 +26,8 @@ else:
class GCSBucketLogger(GCSBucketBase, AdditionalLoggingUtils):
def __init__(self, bucket_name: str | None = None) -> None:
from litellm.proxy._types import CommonProxyErrors
from litellm.proxy.proxy_server import premium_user
from litellm.types.proxy.common_proxy_errors import CommonProxyErrors
self.batch_size = int(os.getenv("GCS_BATCH_SIZE", GCS_DEFAULT_BATCH_SIZE))
self.flush_interval = int(os.getenv("GCS_FLUSH_INTERVAL", GCS_DEFAULT_FLUSH_INTERVAL_SECONDS))
@ -52,8 +52,8 @@ class GCSBucketLogger(GCSBucketBase, AdditionalLoggingUtils):
#### ASYNC ####
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
from litellm.proxy._types import CommonProxyErrors
from litellm.proxy.proxy_server import premium_user
from litellm.types.proxy.common_proxy_errors import CommonProxyErrors
if premium_user is not True:
raise ValueError(

View file

@ -10,7 +10,9 @@ import asyncio
import json
import os
import re
import time
import traceback
from collections.abc import Mapping
from typing import Final, Literal
import httpx
@ -21,6 +23,7 @@ from litellm._uuid import uuid
from litellm.integrations.custom_batch_logger import CustomBatchLogger
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.llms.custom_httpx.http_handler import (
AsyncHTTPHandler,
get_async_httpx_client,
httpxSpecialProvider,
)
@ -104,6 +107,7 @@ class GenericAPILogger(CustomBatchLogger):
max_retries: int = 0,
retry_delay: float = 1.0,
timeout: float | httpx.Timeout | None = None,
async_httpx_client: AsyncHTTPHandler | None = None,
**kwargs,
):
"""
@ -151,7 +155,11 @@ class GenericAPILogger(CustomBatchLogger):
#########################################################
# Init httpx client
#########################################################
self.async_httpx_client = get_async_httpx_client(llm_provider=httpxSpecialProvider.LoggingCallback)
self.async_httpx_client = (
async_httpx_client
if async_httpx_client is not None
else get_async_httpx_client(llm_provider=httpxSpecialProvider.LoggingCallback)
)
endpoint = endpoint or os.getenv("GENERIC_LOGGER_ENDPOINT")
if endpoint is None:
raise ValueError(
@ -191,7 +199,7 @@ class GenericAPILogger(CustomBatchLogger):
self.flush_lock = asyncio.Lock()
super().__init__(**kwargs, flush_lock=self.flush_lock)
asyncio.create_task(self.periodic_flush())
self.log_queue: list[dict | StandardLoggingPayload] = []
self.log_queue: list[Mapping[str, object] | StandardLoggingPayload] = []
def _get_headers(self, headers: dict | None = None):
"""
@ -343,7 +351,14 @@ class GenericAPILogger(CustomBatchLogger):
except Exception as e:
verbose_logger.exception("Generic API Logger Error - %s\n%s", e, traceback.format_exc())
async def async_send_batch(self):
async def flush_queue(self) -> None:
await self.async_send_batch()
async def async_send_batch(self) -> None:
async with self.flush_lock:
await self._send_batch()
async def _send_batch(self) -> None:
"""
Sends the batch of messages to Generic API Endpoint
@ -352,18 +367,18 @@ class GenericAPILogger(CustomBatchLogger):
- ndjson: Sends logs as newline-delimited JSON
- single: Sends each log as individual HTTP request in parallel
"""
batch: Final = tuple(self.log_queue)
if not batch:
return
try:
if not self.log_queue:
return
verbose_logger.debug(
"Generic API Logger - about to flush %s events in '%s' format", len(self.log_queue), self.log_format
"Generic API Logger - about to flush %s events in '%s' format", len(batch), self.log_format
)
if self.log_format == "single":
# Send each log as individual HTTP request in parallel
tasks: Final = []
for log_entry in self.log_queue:
for log_entry in batch:
task = self._post_with_retries(data=safe_dumps(log_entry))
tasks.append(task)
@ -384,9 +399,9 @@ class GenericAPILogger(CustomBatchLogger):
else:
# Format the payload based on log_format
if self.log_format == "json_array":
data = safe_dumps(self.log_queue)
data = safe_dumps(batch)
elif self.log_format == "ndjson":
data = "\n".join(safe_dumps(log) for log in self.log_queue)
data = "\n".join(safe_dumps(log) for log in batch)
else:
raise ValueError(f"Unknown log_format: {self.log_format}")
@ -403,7 +418,8 @@ class GenericAPILogger(CustomBatchLogger):
except Exception as e:
verbose_logger.exception("Generic API Logger Error sending batch - %s\n%s", e, traceback.format_exc())
finally:
self.log_queue.clear()
del self.log_queue[: len(batch)]
self.last_flush_time = time.time()
def _get_v1_logging_payload(self, kwargs, response_obj, start_time, end_time) -> dict:
"""

View file

@ -295,7 +295,9 @@ nothing here imports outside it:
- [`config.py`](./model/config.py) — `OpenTelemetryV2Config`, a pydantic-settings
model that reads `OTEL_*` / `LITELLM_OTEL_*` env vars, plus the feature gate.
`capture_span_content` gates whether prompt/response bodies may be written as
span attributes; it defaults **off** (`no_content`). The Baggage allowlists are
span attributes; it defaults **off** (`no_content`). A team or key can override
the global mode per OTel v2 destination with `capture_message_content`; omitted
values follow the proxy's own callback for that backend, else the global mode. The Baggage allowlists are
configurable, not hard-coded: set `LITELLM_OTEL_BAGGAGE_PROMOTED_KEYS` /
`LITELLM_OTEL_BAGGAGE_METADATA_KEYS` /
`LITELLM_OTEL_BAGGAGE_TEAM_METADATA_KEYS` (comma-separated) as env vars, or

View file

@ -51,6 +51,10 @@ class LangfuseContentOpenTelemetryV2(LangfuseOpenTelemetryV2):
response: "AsyncIterator[ModelResponseStream]",
request_data: Mapping[str, object],
) -> "AsyncGenerator[ModelResponseStream, None]":
if not self._capture_span_content():
async for chunk in response:
yield chunk
return
relayed: Final[list[ModelResponseStream]] = [] # mutable-ok: relayed as they arrive, assembled at end of stream
async for chunk in response:
relayed.append(chunk)
@ -58,6 +62,8 @@ class LangfuseContentOpenTelemetryV2(LangfuseOpenTelemetryV2):
self._stamp_root_io(request_data, lambda: stream_output(tuple(relayed), request_data))
def _stamp_root_io(self, data: Mapping[str, object], render_output: Callable[[], str | None]) -> None:
if not self._capture_span_content():
return
root: Final = request_root_span()
if root is None or not root.is_recording():
return

View file

@ -58,6 +58,7 @@ from litellm.integrations.otel.plumbing.context import (
is_recordable_span,
mcp_message_transport_span,
post_response_root,
request_destinations,
request_root_http_route,
request_root_span,
resolve_internal_call_span_context,
@ -82,6 +83,7 @@ from litellm.integrations.otel.plumbing.providers import (
resolve_meter_provider,
)
from litellm.integrations.otel.plumbing.routing import TenantTracerCache
from litellm.types.utils import captures_span_content
if TYPE_CHECKING:
from opentelemetry.metrics import MeterProvider
@ -254,6 +256,11 @@ class OpenTelemetryV2(CustomLogger):
return None
return GenAIEventRecorder(get_event_logger(provider, LITELLM_TRACER_NAME), provider.resource)
def _capture_span_content(self) -> bool:
return self.config.capture_span_content or any(
captures_span_content(destination.capture_message_content) for destination in request_destinations()
)
# ====================================================================== #
# Proxy global registration
# ====================================================================== #
@ -433,7 +440,7 @@ class OpenTelemetryV2(CustomLogger):
return False
payload: Final = cast("StandardLoggingPayload", raw_payload)
data: Final = MCPToolCallSpanData.from_standard_logging_payload(
payload, capture_content=self.config.capture_span_content
payload, capture_content=self._capture_span_content()
)
# A stray LLM carrier from a ``pre_call`` that mis-fired for this id would
# otherwise linger until evicted; drop it so it's neither leaked nor closed
@ -477,7 +484,7 @@ class OpenTelemetryV2(CustomLogger):
return False
payload: Final = cast("StandardLoggingPayload", raw_payload)
data: Final = MCPListToolsSpanData.from_standard_logging_payload(
payload, capture_content=self.config.capture_span_content
payload, capture_content=self._capture_span_content()
)
if data.identity.call_id:
self._release_carrier(self._open_llm_calls.pop(data.identity.call_id, None))
@ -569,7 +576,7 @@ class OpenTelemetryV2(CustomLogger):
return None
data: Final = LLMCallSpanData.from_standard_logging_payload(
payload,
capture_content=self.config.capture_span_content,
capture_content=self._capture_span_content(),
time_to_first_chunk_seconds=call.time_to_first_chunk_seconds,
request_route=request_root_http_route(),
request_purpose=call.purpose,
@ -1076,6 +1083,6 @@ def build_otel_v2_logger(
def _logger_class(config: OpenTelemetryV2Config) -> type[OpenTelemetryV2]:
if "langfuse" not in config.mapper_names:
return OpenTelemetryV2
from litellm.integrations.otel.langfuse_logger import LangfuseContentOpenTelemetryV2, LangfuseOpenTelemetryV2
from litellm.integrations.otel.langfuse_logger import LangfuseContentOpenTelemetryV2
return LangfuseContentOpenTelemetryV2 if config.capture_span_content else LangfuseOpenTelemetryV2
return LangfuseContentOpenTelemetryV2

View file

@ -16,19 +16,17 @@ from litellm.integrations.otel.model.baggage import (
)
from litellm.integrations.otel.model.spans import POSTGRESQL, db_system
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.utils import OtelSpanScope
from litellm.types.utils import (
CAPTURE_MESSAGE_CONTENT_VALUES,
CaptureMessageContent,
OtelSpanScope,
captures_span_content,
)
#: Master feature-flag env var. The logger is inert until this is truthy.
OTEL_V2_ENV: Final = "LITELLM_OTEL_V2"
class CaptureMessageContent(str):
NO_CONTENT = "no_content"
SPAN_ONLY = "span_only"
EVENT_ONLY = "event_only"
SPAN_AND_EVENT = "span_and_event"
class ExporterOwner(str, Enum):
"""The preset that contributed an exporter. Values match the callback names
in ``presets.PRESET_BY_CALLBACK`` so per-request dynamic-credential routing
@ -297,9 +295,7 @@ class OpenTelemetryV2Config(BaseSettings):
here are lower_snake_case; normalizing at the boundary keeps both
spellings working and lets every downstream comparison stay exact.
"""
if isinstance(value, str):
return value.lower()
return value
return _normalized_capture_message_content(value)
@field_validator("langfuse_span_scope", mode="before")
@classmethod
@ -371,22 +367,39 @@ class OpenTelemetryV2Config(BaseSettings):
@property
def capture_span_content(self) -> bool:
"""Whether prompt/response content may be stamped as span attributes.
Defaults off (``no_content``): an operator must opt in before message
bodies leave the process, so a user request can never force its prompt
or completion into the configured backend while capture is disabled.
"""
return self.capture_message_content in (
CaptureMessageContent.SPAN_ONLY,
CaptureMessageContent.SPAN_AND_EVENT,
)
"""Whether the global setting permits prompt/response span attributes."""
return captures_span_content(self.capture_message_content)
@classmethod
def from_env(cls) -> "OpenTelemetryV2Config":
return cls()
def _normalized_capture_message_content(value: object) -> object:
if isinstance(value, str):
return value.lower()
return value
class _CaptureMessageContentSettings(BaseSettings):
model_config = SettingsConfigDict(extra="ignore", frozen=True)
capture_message_content: str = Field(
default=CaptureMessageContent.NO_CONTENT,
validation_alias=AliasChoices("OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"),
)
@field_validator("capture_message_content", mode="before")
@classmethod
def _normalize_capture_message_content(cls, value: object) -> object:
return _normalized_capture_message_content(value)
def capture_message_content_from_env() -> str:
"""The global capture mode, read without building a settings model that rereads every other env var"""
return _CaptureMessageContentSettings().capture_message_content
_EXCLUDED_SERVICES_INPUT: Final[TypeAdapter[str | tuple[object, ...]]] = TypeAdapter(str | tuple[object, ...])
@ -395,6 +408,10 @@ def excluded_db_systems_from(value: object) -> frozenset[str]:
return _normalize_excluded_services(excluded_service_names(value))
def parse_capture_message_content(value: object) -> str | None:
return value if isinstance(value, str) and value in CAPTURE_MESSAGE_CONTENT_VALUES else None
def excluded_service_names(value: object) -> frozenset[str]:
"""Read a YAML list or comma-separated string of service names, logging and dropping unusable input
so a malformed value cannot stop the OTel logger from being built"""

View file

@ -11,7 +11,7 @@ from urllib.parse import quote
from pydantic import ConfigDict, Field
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.utils import OtelSpanScope
from litellm.types.utils import OtelSpanScope, captures_span_content
class OtelDestination(LiteLLMBaseModel):
@ -50,6 +50,17 @@ class OtelDestination(LiteLLMBaseModel):
default=None,
description="Same, for the requests with a failed span in their tree; ``None`` forwards every one.",
)
capture_message_content: str | None = Field(
default=None,
description=(
"An explicit mode overrides the global capture policy for this destination. "
"Omitted follows the global setting."
),
)
def captures_content(self, default: str) -> bool:
setting: Final = self.capture_message_content
return captures_span_content(default if setting is None else setting)
def header_string(self) -> str:
"""Render headers as the ``k=v,k2=v2`` form an ``ExporterSpec`` expects.
@ -64,9 +75,9 @@ class OtelDestination(LiteLLMBaseModel):
def cache_key(self) -> tuple[str, tuple[tuple[str, str], ...], tuple[tuple[str, str], ...], str | None]:
"""Identity for processor reuse, so one destination means one exporter.
``span_scope`` and the sampling rates are left out on purpose: they decide which
spans reach the processor, not how the processor exports them, so a full and an
``llm_only`` view of the same account share one exporter.
``span_scope``, the sampling rates and ``capture_message_content`` are left out on
purpose: they decide which spans reach the processor and what they carry, not how
the processor exports them, so two views of the same account share one exporter.
"""
return (
self.endpoint,

View file

@ -6,7 +6,7 @@ from typing_extensions import ReadOnly, TypedDict
import litellm
from litellm.integrations.otel.mappers.utils import json_or_none
from litellm.proxy.guardrails.anthropic_sse import assemble_anthropic_sse_stream, is_raw_sse_stream
from litellm.llms.anthropic.pass_through.stream_assembly import assemble_anthropic_sse_stream, is_raw_sse_stream
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.llms.openai import ResponseCompletedEvent, ResponsesAPIResponse
from litellm.types.utils import ModelResponse, ModelResponseStream

View file

@ -5,7 +5,7 @@ import random
import threading
import time
from collections import OrderedDict
from collections.abc import Callable, Iterable, Mapping, Sequence
from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, Literal
@ -40,8 +40,19 @@ from opentelemetry.util.types import Attributes, AttributeValue
from litellm._logging import verbose_logger
from litellm._version import version as litellm_version
from litellm.integrations.otel.mappers.langfuse import LANGFUSE_TRACE_NAME
from litellm.integrations.otel.model.config import ExporterOwner, ExporterSpec, OpenTelemetryV2Config
from litellm.integrations.otel.mappers.langfuse import (
LANGFUSE_OBSERVATION_INPUT,
LANGFUSE_OBSERVATION_OUTPUT,
LANGFUSE_TRACE_NAME,
)
from litellm.integrations.otel.model.config import (
CaptureMessageContent,
ExporterOwner,
ExporterSpec,
OpenTelemetryV2Config,
capture_message_content_from_env,
captures_span_content,
)
from litellm.integrations.otel.model.semconv import (
DB,
MCP,
@ -385,6 +396,26 @@ _CAPTURED_HEADER_PREFIXES: Final = ("http.request.header.", "http.response.heade
# ``?key=`` query parameter.
_URL_KEYS: Final = frozenset({"http.url", "http.target", "url.full"})
_URL_QUERY_KEY: Final = "url.query"
# Prompt and response content, in every vocabulary the mappers write it in once the
# global capture policy collects it: GenAI semconv (MCP tool-call arguments and results
# included), Langfuse, OpenInference's blobs and per-index messages, Langtrace, Weave.
_MESSAGE_CONTENT_KEYS: Final = frozenset(
{
GenAI.INPUT_MESSAGES,
GenAI.OUTPUT_MESSAGES,
GenAI.SYSTEM_INSTRUCTIONS,
GenAI.TOOL_CALL_ARGUMENTS,
GenAI.TOOL_CALL_RESULT,
LANGFUSE_OBSERVATION_INPUT,
LANGFUSE_OBSERVATION_OUTPUT,
"input.value",
"output.value",
"llm.prompts",
"llm.completions",
"weave.output",
}
)
_INDEXED_MESSAGE_PREFIXES: Final = ("llm.input_messages.", "llm.output_messages.")
class _SpanView(ReadableSpan):
@ -415,6 +446,19 @@ class _SpanView(ReadableSpan):
)
def _is_message_content(key: str) -> bool:
return key in _MESSAGE_CONTENT_KEYS or key.startswith(_INDEXED_MESSAGE_PREFIXES)
def without_message_content(span: ReadableSpan) -> ReadableSpan:
"""``span`` without its prompt and response content, everything else left as it was."""
attributes: Final = span.attributes or _NO_ATTRIBUTES
if not any(_is_message_content(key) for key in attributes):
return span
kept: Final = MappingProxyType({key: value for key, value in attributes.items() if not _is_message_content(key)})
return _SpanView(span, span.resource, kept, span.events, span.status, parent=span.parent)
def _is_database_span(attributes: Mapping[str, AttributeValue]) -> bool:
return any(key in attributes for key in _DB_SYSTEM_KEYS)
@ -553,7 +597,11 @@ def _tenant_resource(resource: Resource, defaults: Mapping[str, str], extra: Map
return filled.merge(Resource(extra)) if extra else filled
def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> ReadableSpan:
def _for_destination(
span: ReadableSpan,
destination: "OtelDestination",
default_capture: str,
) -> ReadableSpan:
"""The view of ``span`` a tenant destination receives.
A span the tenant's own call produced keeps its error text. Every other span is
@ -564,8 +612,10 @@ def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> Read
exception it raised and names the operator's guardrail endpoint. Stack traces walk
the operator's install and come off every span, as do the headers the operator
captures on the server span, whose request side holds the caller's bearer token,
and the query string of the request URL, which can hold the same key. The span
itself stays, so the tenant still gets the whole trace tree.
and the query string of the request URL, which can hold the same key. A
destination whose effective setting disables span content loses the prompt and
response content too. The span itself stays, so the tenant still gets the whole
trace tree.
"""
extra: Final = destination.resource_attributes
defaults: Final = destination.resource_defaults
@ -573,11 +623,12 @@ def _for_destination(span: ReadableSpan, destination: "OtelDestination") -> Read
database: Final = _is_database_span(attributes)
owned: Final = _is_tenant_owned_span(attributes)
unreachable: Final = _guardrail_unreachable(attributes)
redacted: Final = not destination.captures_content(default_capture)
kept: Final = MappingProxyType(
{
key: _without_query(key, value)
for key, value in attributes.items()
if _tenant_visible(key, database, owned, unreachable)
if _tenant_visible(key, database, owned, unreachable) and not (redacted and _is_message_content(key))
}
)
recorded: Final = span.events
@ -612,6 +663,8 @@ class TenantFanOutSpanProcessor(SpanProcessor):
shutdown_drain_seconds: float = _SHUTDOWN_DRAIN_SECONDS,
operator_sinks: 'Mapping[_SinkKey, "OtelSpanScope"]' = MappingProxyType({}),
excluded_db_systems: frozenset[str] = frozenset(),
default_capture: str = CaptureMessageContent.NO_CONTENT,
default_capture_by_backend: Mapping[str, str] = MappingProxyType({}),
pending_drains: int = _MAX_PENDING_DRAINS,
drain_pool: _DrainPool | None = None,
sampling_draw: Callable[[], float] | None = None,
@ -619,6 +672,8 @@ class TenantFanOutSpanProcessor(SpanProcessor):
self._operator_sinks: Final = operator_sinks
self._draw: Final = sampling_draw if sampling_draw is not None else random.random
self._excluded_db_systems: Final = excluded_db_systems
self._default_capture: Final = default_capture
self._default_capture_by_backend: Final = default_capture_by_backend
self._drain_seconds: Final = shutdown_drain_seconds
self._lock: Final = threading.Condition()
self._closed = False # guarded by ``_lock``: an unlocked read races the teardown it gates
@ -661,12 +716,22 @@ class TenantFanOutSpanProcessor(SpanProcessor):
for held_span, held_destination in self._settle(trace_id, failed):
self._forward(held_span, held_destination)
def _default_capture_for(self, destination: "OtelDestination") -> str:
backend: Final = destination.callback_name
return (
self._default_capture_by_backend.get(backend, self._default_capture) if backend else self._default_capture
)
def _forward(self, span: ReadableSpan, destination: "OtelDestination") -> None:
processor: Final = self._acquire(destination)
if processor is None:
return
try:
processor.on_end(_scoped(_for_destination(span, destination), destination.span_scope))
processor.on_end(
_scoped(
_for_destination(span, destination, self._default_capture_for(destination)), destination.span_scope
)
)
except Exception as exc: # noqa: BLE001 # one destination's failure must not cost the others their span
verbose_logger.debug("OTel V2 fan-out: forwarding to %s failed: %s", destination.endpoint, exc)
finally:
@ -963,14 +1028,15 @@ class _OverriddenBackendFilter(SpanProcessor):
Wrapping is the only place this works: ``SynchronousMultiSpanProcessor.on_end``
ignores return values, so a sibling processor can never veto the export.
Under ``additive`` mode nothing is suppressed, so the wrapper passes every span
straight through and the operator keeps its copy.
Under ``additive`` mode spans are not suppressed, though global content policy
still strips message content when disabled.
``scope`` narrows what the exporter receives independently of that: under
``llm_only`` the model-call spans go through as trace roots and the rest of the
tree is held back, unless a destination of the request names ``sink``, the account
this exporter writes to, with a wider scope: the fan-out then delivers the rest of
the tree there and the model call keeps its place in it.
the tree there and the model call keeps its place in it. A destination naming
``sink`` sets this exporter's content policy; otherwise it follows global capture.
"""
def __init__(
@ -979,11 +1045,15 @@ class _OverriddenBackendFilter(SpanProcessor):
owner: str | None,
scope: "OtelSpanScope" = "full",
sink: _SinkKey | None = None,
global_captures: bool = True,
filter_content: bool = True,
) -> None:
self._inner: Final = inner
self._owner: Final = owner
self._scope: Final = scope
self._sink: Final = sink
self._global_captures: Final = global_captures
self._filter_content: Final = filter_content
def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None:
self._inner.on_start(span, parent_context)
@ -991,7 +1061,28 @@ class _OverriddenBackendFilter(SpanProcessor):
def on_end(self, span: ReadableSpan) -> None:
if self._owner in suppressed_backends() or not _in_scope(span, self._scope):
return
self._inner.on_end(_scoped(span, self._account_scope()))
scoped: Final = _scoped(span, self._account_scope())
self._inner.on_end(
without_message_content(scoped) if self._filter_content and not self._account_captures_content() else scoped
)
def _account_captures_content(self) -> bool:
"""Whether a destination of this request that names ``sink`` captures content.
The fan-out skips a destination this exporter already writes to, so this copy is
the one that account gets and has to honor the destination's content mode.
"""
explicit: Final = next(
(
destination.capture_message_content
for destination in request_destinations()
if self._sink is not None
and destination.capture_message_content is not None
and _sink_key(destination.endpoint, destination.headers) == self._sink
),
None,
)
return captures_span_content(explicit) if explicit is not None else self._global_captures
def _account_scope(self) -> "OtelSpanScope":
if self._scope == "full" or self._sink is None:
@ -1010,6 +1101,25 @@ class _OverriddenBackendFilter(SpanProcessor):
return self._inner.force_flush(timeout_millis)
class _MessageContentFilter(SpanProcessor):
"""Strip prompt and response content from every span before ``inner`` exports it."""
def __init__(self, inner: SpanProcessor) -> None:
self._inner: Final = inner
def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None:
self._inner.on_start(span, parent_context)
def on_end(self, span: ReadableSpan) -> None:
self._inner.on_end(without_message_content(span))
def shutdown(self) -> None:
self._inner.shutdown()
def force_flush(self, timeout_millis: int = 30000) -> bool:
return self._inner.force_flush(timeout_millis)
def build_span_exporter(config: OpenTelemetryV2Config) -> SpanExporter:
"""Build a single exporter from the top-level config fields.
@ -1261,12 +1371,23 @@ def build_resource(config: OpenTelemetryV2Config) -> Resource:
return Resource.create(attributes)
def _spec_processor(
spec: ExporterSpec, use_simple_processor: bool | None, strip_message_content: bool
) -> SpanProcessor:
exporting: Final = _processor_for(
_exporter_from_spec(spec),
(spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor),
)
return _MessageContentFilter(exporting) if strip_message_content else exporting
def build_tracer_provider(
config: OpenTelemetryV2Config,
exporter: SpanExporter | None = None,
baggage_processor: SpanProcessor | None = None,
use_simple_processor: bool | None = None,
tenant_overrides: bool = False,
strip_message_content: bool = False,
) -> TracerProvider:
"""Build the shared :class:`TracerProvider`.
@ -1285,6 +1406,9 @@ def build_tracer_provider(
``config.langfuse_span_scope`` narrows the exporter owned by ``langfuse_otel``
alone; a collector or any other backend in the same config keeps the full tree.
``strip_message_content`` removes prompt and response content before every
exporter, for a routed provider whose spans can carry content another
destination asked for.
"""
provider: Final = TracerProvider(resource=build_resource(config))
if baggage_processor is None:
@ -1300,17 +1424,23 @@ def build_tracer_provider(
for spec in config.exporters:
if spec.requires_headers and not spec.headers:
continue
exp = _exporter_from_spec(spec)
processor = _processor_for(
exp,
(spec.use_simple_processor if spec.use_simple_processor is not None else use_simple_processor),
)
processor = _spec_processor(spec, use_simple_processor, strip_message_content)
owner = spec.owner.value if tenant_overrides and spec.owner is not None else None
scope = _operator_scope(config, spec)
sink = _sink_key(spec.endpoint, parse_headers(spec.headers)) if _exports_to_the_wire(spec) else None
provider.add_span_processor(
_OverriddenBackendFilter(processor, owner, scope, sink)
if owner is not None or scope != "full"
_OverriddenBackendFilter(
processor,
owner,
scope,
sink,
config.capture_span_content,
tenant_overrides,
)
if owner is not None
or scope != "full"
or sink is not None
or (tenant_overrides and not config.capture_span_content)
else processor
)
return provider
@ -1340,10 +1470,29 @@ def attach_tenant_fan_out(
TenantFanOutSpanProcessor(
operator_sinks=operator_sink_scopes(*configs),
excluded_db_systems=excluded_db_systems,
default_capture=_operator_capture_mode(configs),
default_capture_by_backend=MappingProxyType(dict(_owned_capture_modes(configs))),
)
)
def _operator_capture_mode(configs: Sequence[OpenTelemetryV2Config]) -> str:
"""The mode of the operator's generic ``otel`` callback, so ``callback_settings.otel`` counts as the global default"""
return next(
(config.capture_message_content for config in configs if any(spec.owner is None for spec in config.exporters)),
capture_message_content_from_env(),
)
def _owned_capture_modes(
configs: Sequence[OpenTelemetryV2Config],
) -> Iterator[tuple[str, str]]:
for config in configs:
for spec in config.exporters:
if spec.owner is not None:
yield spec.owner.value, config.capture_message_content
def deliverable_destinations(
destinations: Iterable["OtelDestination"],
provider: trace.TracerProvider | None = None,

View file

@ -282,7 +282,8 @@ class TenantTracerCache:
self._providers.move_to_end(cache_key)
return cached
built: Final = build_tracer_provider(
self._routed_config(credential_headers, project_headers, endpoint, service_name)
self._routed_config(credential_headers, project_headers, endpoint, service_name),
strip_message_content=not self._config.capture_span_content,
)
self._providers[cache_key] = built
return built

View file

@ -187,6 +187,7 @@ def destination_for(
callback_name: str,
params: StandardCallbackDynamicParams,
service_name: str | None = None,
capture_message_content: str | None = None,
) -> OtelDestination | None:
"""The destination ``params`` names for ``callback_name``, or ``None``.
@ -217,6 +218,7 @@ def destination_for(
callback_name=callback_name,
protocol=protocol,
span_scope=_span_scope(callback_name, params),
capture_message_content=capture_message_content,
success_sampling_rate=success_sampling_rate,
error_sampling_rate=error_sampling_rate,
)

View file

@ -9,6 +9,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
from litellm._logging import verbose_logger
from litellm.integrations.otel.model.config import (
CaptureMessageContent,
ExporterOwner,
ExporterSpec,
OpenTelemetryV2Config,
@ -66,7 +67,11 @@ def newrelic_preset(
# New Relic ingests the OTLP GenAI semantic conventions natively.
"mapper_names": ensure_mappers(base.mapper_names, "genai"),
**(
{"capture_message_content": ("span_only" if settings.record_content else "no_content")}
{
"capture_message_content": (
CaptureMessageContent.SPAN_ONLY if settings.record_content else CaptureMessageContent.NO_CONTENT
)
}
if settings.record_content is not None
else {}
),

View file

@ -21,6 +21,7 @@ import litellm
from litellm._internal_context import with_service_target
from litellm._logging import print_verbose, verbose_logger
from litellm.constants import (
AUTH_OBJECTS_TARGET,
PROMETHEUS_OVERFLOW_SERIES_LABEL_VALUE,
PROXY_LLM_PROVIDER_FALLBACK,
PROXY_REJECTED_BEFORE_ROUTING_KEY,
@ -49,13 +50,10 @@ from litellm.litellm_core_utils.service_tier_utils import (
get_service_tier_from_standard_logging_payload,
)
from litellm.models.end_user import LiteLLM_EndUserTable
from litellm.proxy._types import (
LiteLLM_DeletedVerificationToken,
LiteLLM_TeamTable,
LiteLLM_UserTable,
UserAPIKeyAuth,
)
from litellm.proxy.common_utils.user_api_key_cache import AUTH_OBJECTS_TARGET
from litellm.models.team import LiteLLM_TeamTable
from litellm.models.user import LiteLLM_UserTable
from litellm.models.verification_token import LiteLLM_DeletedVerificationToken
from litellm.proxy._types import UserAPIKeyAuth
from litellm.repositories.base_repository import BaseRepository
from litellm.repositories.budget_repository import BudgetRepository
from litellm.repositories.organization_repository import OrganizationRepository

View file

@ -0,0 +1,61 @@
"""
This module is responsible for handling Getting/Setting the proxy server request from cold storage.
It allows fetching a dict of the proxy server request from s3 or GCS bucket.
"""
from typing import Final
import litellm
from litellm import _custom_logger_compatible_callbacks_literal
from litellm.integrations.custom_logger import CustomLogger
class ColdStorageHandler:
"""
This class is responsible for handling Getting/Setting the proxy server request from cold storage.
It allows fetching a dict of the proxy server request from s3 or GCS bucket.
The cold storage logger can be injected for testing; when omitted it is
resolved from the configured ``litellm.cold_storage_custom_logger``.
"""
def __init__(self, cold_storage_logger: CustomLogger | None = None):
self._injected_cold_storage_logger = cold_storage_logger
async def get_proxy_server_request_from_cold_storage_with_object_key(
self,
object_key: str,
) -> dict | None:
"""
Get the proxy server request from cold storage using the object key directly.
Args:
object_key: The S3/GCS object key to retrieve
Returns:
Optional[dict]: The proxy server request dict or None if not found
"""
custom_logger: Final = self._injected_cold_storage_logger or self._resolve_cold_storage_logger()
if custom_logger is None:
return None
return await custom_logger.get_proxy_server_request_from_cold_storage_with_object_key(
object_key=object_key,
)
def _resolve_cold_storage_logger(self) -> CustomLogger | None:
custom_logger_name: Final = self._select_custom_logger_for_cold_storage()
if custom_logger_name is None:
return None
return litellm.logging_callback_manager.get_active_custom_logger_for_callback_name(custom_logger_name)
def _select_custom_logger_for_cold_storage(
self,
) -> _custom_logger_compatible_callbacks_literal | None:
cold_storage_custom_logger: Final[_custom_logger_compatible_callbacks_literal | None] = (
litellm.cold_storage_custom_logger
)
return cold_storage_custom_logger

View file

@ -7,6 +7,7 @@ from types import MappingProxyType
from typing import Final, Protocol, cast
import httpx
from pydantic import TypeAdapter, ValidationError
import litellm
from litellm._logging import _ENABLE_SECRET_REDACTION, redact_string, verbose_logger
@ -256,6 +257,23 @@ class _ProviderHTTPException(Protocol):
llm_provider: str
_ERROR_BODY_ADAPTER: Final = TypeAdapter(dict[str, object])
def _openai_cyber_policy_body(original_exception: _ProviderHTTPException) -> Mapping[str, object] | None:
raw_body: Final[object] = getattr(original_exception, "body", None)
try:
body: Final = (
_ERROR_BODY_ADAPTER.validate_python(raw_body)
if raw_body is not None
else _ERROR_BODY_ADAPTER.validate_json(getattr(original_exception, "message", ""))
)
error: Final = _ERROR_BODY_ADAPTER.validate_python(body.get("error", body))
except ValidationError:
return None
return error if error.get("code") == "cyber_policy" else None
def _litellm_proxy_response(
original_exception: _ProviderHTTPException, custom_llm_provider: str
) -> httpx.Response | None:
@ -284,6 +302,15 @@ def _map_openai_exception(
extra_information: str,
) -> None:
response: Final = _litellm_proxy_response(original_exception, custom_llm_provider)
if (cyber_policy_body := _openai_cyber_policy_body(original_exception)) is not None:
raise ContentPolicyViolationError(
message=f"ContentPolicyViolationError: {exception_provider} - {original_exception}",
llm_provider=custom_llm_provider,
model=model,
response=response,
litellm_debug_info=extra_information,
body=dict(cyber_policy_body),
)
# custom_llm_provider is openai, make it OpenAI
message = get_error_message(error_obj=original_exception)
if message is None:

View file

@ -6,6 +6,7 @@ from typing import Any, Final
from litellm.types.utils import (
ARIZE_OTLP_PROTOCOLS,
CAPTURE_MESSAGE_CONTENT_VALUES,
OTEL_SPAN_SCOPES,
TRUSTED_CALLBACK_VARS_FIELD,
StandardCallbackDynamicParams,
@ -91,6 +92,13 @@ def validate_arize_otlp_protocol_value(value: str) -> None:
raise ValueError(f"Invalid arize_otlp_protocol {value!r}: must be one of {sorted(ARIZE_OTLP_PROTOCOLS)}")
def validate_capture_message_content_value(value: str) -> None:
if value not in CAPTURE_MESSAGE_CONTENT_VALUES:
raise ValueError(
f"Invalid capture_message_content {value!r}: must be one of {sorted(CAPTURE_MESSAGE_CONTENT_VALUES)}"
)
# Hardcoded list of supported callback params to avoid runtime inspection issues with TypedDict
_supported_callback_params: Final[tuple[str, ...]] = (
"langfuse_public_key",

View file

@ -0,0 +1,49 @@
from collections.abc import Mapping
from typing import Final
def extract_cache_read_tokens(usage_object: Mapping[str, object] | None) -> int:
"""Cache-read tokens from a logged usage object, whatever shape recorded them.
Anthropic writes a top-level ``cache_read_input_tokens``; OpenAI-compatible
providers (moonshotai, openai, deepseek, etc.) write
``prompt_tokens_details.cached_tokens``. This is the one owner of that
normalization: callers hand over the usage object rather than threading a
count that could disagree with it.
"""
if not usage_object:
return 0
explicit: Final = usage_object.get("cache_read_input_tokens")
if isinstance(explicit, (int, float)) and explicit:
return int(explicit)
details: Final = usage_object.get("prompt_tokens_details")
if not isinstance(details, Mapping):
return 0
cached: Final = details.get("cached_tokens")
return int(cached) if isinstance(cached, (int, float)) else 0
def extract_cache_creation_tokens(usage_object: Mapping[str, object] | None) -> int:
"""Cache-write tokens from a logged usage object, whatever shape recorded them.
Anthropic writes a top-level ``cache_creation_input_tokens``; OpenAI-compatible
providers (kimi-k2 etc.) write ``prompt_tokens_details.cache_write_tokens`` or
``prompt_tokens_details.cache_creation_tokens``.
"""
if not usage_object:
return 0
explicit: Final = usage_object.get("cache_creation_input_tokens")
if isinstance(explicit, (int, float)) and explicit:
return int(explicit)
details: Final = usage_object.get("prompt_tokens_details")
if not isinstance(details, Mapping):
return 0
written: Final = next(
(
value
for value in (details.get("cache_write_tokens"), details.get("cache_creation_tokens"))
if isinstance(value, (int, float)) and value
),
0,
)
return int(written)

View file

@ -0,0 +1,96 @@
"""
Safe filesystem path construction for user-controlled inputs.
Use safe_join() instead of os.path.join() whenever a path component
comes from user input (request parameters, uploaded filenames, etc.)
to prevent directory traversal attacks.
"""
import os
from typing import Final
def safe_join(base_dir: str, *parts: str) -> str:
"""
Join path components and verify the result stays within base_dir.
Resolves symlinks and '..' sequences, then checks the final path
is a descendant of base_dir. Raises ValueError if traversal is
detected.
Args:
base_dir: The trusted base directory.
*parts: User-controlled path components to append.
Returns:
The resolved absolute path as a string.
Raises:
ValueError: If the resolved path escapes base_dir.
"""
for part in parts:
if "\x00" in part:
raise ValueError("Path contains null byte")
base: Final = os.path.realpath(base_dir)
resolved: Final = os.path.realpath(os.path.join(base, *parts))
if not (resolved.startswith(base + os.sep) or resolved == base):
raise ValueError(f"Path {resolved!r} escapes base directory {base!r}")
return resolved
def try_safe_join(base_dir: str, *parts: str) -> str | None:
"""safe_join, with None instead of ValueError when the path escapes base_dir."""
try:
return safe_join(base_dir, *parts)
except ValueError:
return None
def is_within(path: str, base_dir: str) -> bool:
"""True when path, with symlinks resolved, is base_dir or sits inside it."""
base: Final = os.path.realpath(base_dir)
resolved: Final = os.path.realpath(path)
return resolved.startswith(base + os.sep) or resolved == base
def join_within(base_dir: str, *parts: str) -> str | None:
"""Join without following symlinks; None when the joined path leaves base_dir.
Only the supplied components are checked (``..`` and absolute parts are
rejected), so a symlink stored inside base_dir that points elsewhere is
still returned. Use safe_join when the target itself must stay inside.
"""
for part in parts:
if "\x00" in part:
return None
base: Final = os.path.normpath(os.path.abspath(base_dir))
joined: Final = os.path.normpath(os.path.join(base, *parts))
if not joined.startswith(base + os.sep):
return None
return joined
def safe_filename(filename: str) -> str:
"""
Extract a safe filename from a user-supplied path.
Strips all directory components (both Unix and Windows separators),
returning only the final name. Use this for uploaded file names
before writing to disk.
Args:
filename: User-supplied filename (may contain path separators).
Returns:
The basename only, with no directory components.
Raises:
ValueError: If the resulting filename is empty or contains null bytes.
"""
if "\x00" in filename:
raise ValueError("Filename contains null byte")
# Normalize backslash separators for cross-platform safety
name: Final = filename.replace("\\", "/").rsplit("/", 1)[-1]
if not name or name in (".", ".."):
raise ValueError("Empty or unsafe filename")
return name

View file

@ -0,0 +1,93 @@
from typing import Final
from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth
from litellm.types.proxy.auth.user_roles import LitellmUserRoles
def is_proxy_admin(user_api_key_dict: UserAPIKeyAuth | None) -> bool:
if user_api_key_dict is None:
return False
return (
user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
or user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value
)
def get_resource_owner_scopes(
user_api_key_dict: UserAPIKeyAuth | None,
) -> list[str]:
"""
Return ownership scopes that may access a user-created proxy resource.
Raw user_id is included for rows created before scope prefixes existed.
Prefixes avoid collisions when falling back to team/org/key ownership
for keys that do not have a user_id.
Identity-less callers (no user_id, team_id, org_id, api_key, or token)
return ``[]`` — they share no scope with any other caller, so access
checks against an existing owner always fail and creates that depend
on a primary scope must reject up front. Returning a shared sentinel
here would let any two identity-less callers see each other's data.
"""
if user_api_key_dict is None:
return []
scopes: Final[list[str]] = []
def _add(scope: str | None) -> None:
if scope and scope not in scopes:
scopes.append(scope)
if user_api_key_dict.user_id:
_add(user_api_key_dict.user_id)
_add(f"user:{user_api_key_dict.user_id}")
if user_api_key_dict.team_id:
_add(f"team:{user_api_key_dict.team_id}")
if user_api_key_dict.org_id:
_add(f"org:{user_api_key_dict.org_id}")
if user_api_key_dict.api_key:
_add(f"key:{user_api_key_dict.api_key}")
if user_api_key_dict.token:
_add(f"key:{user_api_key_dict.token}")
return scopes
def get_primary_resource_owner_scope(
user_api_key_dict: UserAPIKeyAuth | None,
) -> str | None:
"""Return the canonical owner scope to stamp on newly-created rows.
``None`` for identity-less callers — callers that depend on a primary
scope to record ownership must surface that as a hard error rather
than fall back to a shared sentinel (which would collapse every
identity-less caller into the same logical owner).
"""
if user_api_key_dict is None:
return None
if user_api_key_dict.user_id:
return user_api_key_dict.user_id
if user_api_key_dict.team_id:
return f"team:{user_api_key_dict.team_id}"
if user_api_key_dict.org_id:
return f"org:{user_api_key_dict.org_id}"
if user_api_key_dict.api_key:
return f"key:{user_api_key_dict.api_key}"
if user_api_key_dict.token:
return f"key:{user_api_key_dict.token}"
return None
def user_can_access_resource_owner(
owner: str | None,
user_api_key_dict: UserAPIKeyAuth | None,
) -> bool:
if user_api_key_dict is None:
return True
if is_proxy_admin(user_api_key_dict):
return True
if owner is None:
return False
return owner in get_resource_owner_scopes(user_api_key_dict)

View file

@ -110,7 +110,10 @@ def served_output_texts(response: object) -> ServedTexts | None:
def served_stream_output_texts(chunks: Sequence[object]) -> ServedTexts | None:
if chunks and all(isinstance(chunk, ModelResponseStream) for chunk in chunks):
return _chat_stream_texts(tuple(chunk for chunk in chunks if isinstance(chunk, ModelResponseStream)))
from litellm.proxy.guardrails.anthropic_sse import assemble_anthropic_sse_stream, is_anthropic_sse_stream
from litellm.llms.anthropic.pass_through.stream_assembly import (
assemble_anthropic_sse_stream,
is_anthropic_sse_stream,
)
if not is_anthropic_sse_stream(chunks):
return None

View file

@ -3,12 +3,17 @@ import time
from collections.abc import Callable, Iterator, Mapping, Sequence
from itertools import groupby
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, TypeAlias, TypedDict, Union, cast
from typing import TYPE_CHECKING, Any, Final, Optional, TypeAlias, TypedDict, Union, cast
from pydantic import BaseModel
from typing_extensions import ReadOnly, Required
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR, set_hidden_params
from litellm.litellm_core_utils.prompt_templates.common_utils import (
get_content_from_model_response, # pyright: ignore[reportUnknownVariableType] # response param accepts an untyped dict
)
from litellm.types.llms.openai import (
ChatCompletionAssistantContentValue,
ChatCompletionAudioDelta,
@ -28,11 +33,13 @@ from litellm.types.utils import (
Delta,
Function,
FunctionCall,
LlmProviders,
ModelResponse,
ModelResponseStream,
PromptTokensDetailsWrapper,
ServerToolUse,
StreamingChoices,
TextCompletionResponse,
Usage,
)
from litellm.utils import print_verbose, token_counter
@ -1194,3 +1201,421 @@ def concatenate_base64_list(base64_strings: list[str]) -> str:
# Encode the concatenated bytes back to base64
return base64.b64encode(combined_bytes).decode("utf-8")
def stream_chunk_builder_text_completion(chunks: list, messages: Sequence | None = None) -> TextCompletionResponse:
id: Final = chunks[0]["id"]
object: Final = chunks[0]["object"]
created: Final = chunks[0]["created"]
model: Final = chunks[0]["model"]
system_fingerprint: Final = chunks[0].get("system_fingerprint", None)
finish_reason: Final = chunks[-1]["choices"][0]["finish_reason"]
logprobs: Final = chunks[-1]["choices"][0]["logprobs"]
content_list: Final = []
for chunk in chunks:
choices = chunk["choices"]
for choice in choices:
if choice is not None and hasattr(choice, "text") and choice.get("text") is not None:
_choice = choice.get("text")
content_list.append(_choice)
# Combine the "content" strings into a single string || combine the 'function' strings into a single string
combined_content: Final = "".join(content_list)
try:
prompt_tokens = token_counter(model=model, messages=messages)
except Exception: # don't allow this failing to block a complete streaming response from being returned
print_verbose("token_counter failed, assuming prompt tokens is 0")
prompt_tokens = 0
completion_tokens: Final = token_counter(
model=model,
text=combined_content,
count_response_tokens=True, # count_response_tokens is a Flag to tell token counter this is a response, No need to add extra tokens we do for input messages
)
response: Final = {
"id": id,
"object": object,
"created": created,
"model": model,
"system_fingerprint": system_fingerprint,
"choices": [
{
"text": combined_content,
"index": 0,
"logprobs": logprobs,
"finish_reason": finish_reason,
}
],
"usage": {
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": prompt_tokens + completion_tokens,
},
}
return TextCompletionResponse(**response)
_CALCULATOR_PRICED_REPORTED_COST_PROVIDERS: Final = frozenset({LlmProviders.XAI.value})
def _reported_cost_is_priced_by_calculator(logging_obj: Optional["Logging"]) -> bool:
if logging_obj is None:
return False
provider: Final[object] = logging_obj.model_call_details.get("custom_llm_provider")
return provider in _CALCULATOR_PRICED_REPORTED_COST_PROVIDERS
def _stream_builder_response_cost(response: ModelResponse, logging_obj: Optional["Logging"]) -> float | None:
usage_cost: Final = getattr(getattr(response, "usage", None), "cost", None)
if isinstance(usage_cost, (int, float)) and not _reported_cost_is_priced_by_calculator(logging_obj):
return float(usage_cost)
if logging_obj is not None:
return None
provider_hint: Final = response._hidden_params.get( # pyright: ignore[reportPrivateUsage] # no public accessor
"custom_llm_provider"
)
try:
return litellm.completion_cost(completion_response=response, custom_llm_provider=provider_hint)
except Exception:
return _stream_builder_model_map_cost(response)
def _joined_streamed_citations(streamed_citations: "tuple[object, ...]") -> "list[object]":
if all(isinstance(citation, list) for citation in streamed_citations):
return list(streamed_citations)
return [list(streamed_citations)]
def _stream_builder_model_map_cost(response: ModelResponse) -> float | None:
model_name: Final = response.model
usage: Final = getattr(response, "usage", None)
if not model_name or not isinstance(usage, Usage):
return None
try:
prompt_cost, completion_tokens_cost = litellm.cost_per_token(model=model_name, usage_object=usage)
return prompt_cost + completion_tokens_cost
except Exception: # noqa: BLE001 # cost_per_token raises bare Exception for unpriceable models
return None
def _set_stream_builder_response_cost(response: ModelResponse, logging_obj: Optional["Logging"]) -> None:
response_cost: Final = _stream_builder_response_cost(response, logging_obj)
if response_cost is None:
return
hidden_params: Final = response._hidden_params # pyright: ignore[reportPrivateUsage] # no public accessor
hidden_params["response_cost"] = response_cost
def _stamp_streaming_usage_cost(usage: Usage, response: ModelResponse, logging_obj: Optional["Logging"]) -> None:
if logging_obj is None:
return
if isinstance(getattr(usage, "cost", None), (int, float)):
return
computed_cost: Final = logging_obj.response_cost_calculator(result=response)
if isinstance(computed_cost, (int, float)) and computed_cost > 0:
setattr(usage, "cost", computed_cost)
_NON_TEXT_DELTA_FIELDS: Final = (
"tool_calls",
"function_call",
"reasoning_content",
"thinking_blocks",
"annotations",
"audio",
"images",
"provider_specific_fields",
)
def _stream_choice_delta(choice: object) -> Mapping[str, object]:
delta: Final = choice.get("delta", {}) if isinstance(choice, dict) else getattr(choice, "delta", {})
if isinstance(delta, Mapping):
return delta
if isinstance(delta, BaseModel):
return delta.model_dump()
return {}
def _delta_carries_more_than_text(delta: Mapping[str, object]) -> bool:
return any(delta.get(field) is not None for field in _NON_TEXT_DELTA_FIELDS)
def _simple_text_part(choices: Sequence[object]) -> str | None:
deltas: Final = tuple(_stream_choice_delta(choice) for choice in choices)
if any(_delta_carries_more_than_text(delta) for delta in deltas):
return None
content: Final = deltas[0].get("content")
return content if isinstance(content, str) else ""
def stream_chunk_builder(
chunks: list,
messages: Sequence | None = None,
start_time=None,
end_time=None,
logging_obj: Optional["Logging"] = None,
count_prompt_tokens: Callable[[], int] | None = None,
) -> ModelResponse | TextCompletionResponse | None:
try:
if chunks is None:
raise litellm.APIError(
status_code=500,
message="Error building chunks for logging/streaming usage calculation",
llm_provider="",
model="",
)
if not chunks:
return None
processor: Final = ChunkProcessor(chunks, messages)
chunks = processor.chunks
### BASE-CASE ###
if len(chunks) == 0:
return None
## Route to the text completion logic
first_chunk_with_choices: Final = next((c for c in chunks if c.get("choices")), None)
if first_chunk_with_choices is not None and isinstance(
first_chunk_with_choices["choices"][0], litellm.utils.TextChoices
): # route to the text completion logic
return stream_chunk_builder_text_completion(chunks=chunks, messages=messages)
model: Final = chunks[0]["model"]
# Initialize the response dictionary
response: Final = processor.build_base_response(chunks)
# Fast path for the common text-only streaming case:
# avoid repeated multi-pass list scans over chunks.
simple_content_parts: Final[list[str]] = []
is_simple_text_stream = True
for chunk in chunks:
if not chunk.get("choices"):
continue
if (part := _simple_text_part(chunk["choices"])) is None:
is_simple_text_stream = False
break
if part:
simple_content_parts.append(part)
if is_simple_text_stream:
if simple_content_parts:
response["choices"][0]["message"]["content"] = "".join(simple_content_parts)
completion_output = get_content_from_model_response(response)
usage = processor.calculate_usage(
chunks=chunks,
model=model,
completion_output=completion_output,
messages=messages,
reasoning_tokens=0,
count_prompt_tokens=count_prompt_tokens,
)
setattr(response, "usage", usage)
# Propagate provider_specific_fields from chunk hidden params when present.
for chunk in reversed(chunks):
if isinstance(chunk, dict):
hidden = chunk.get("_hidden_params")
else:
hidden = getattr(chunk, HIDDEN_PARAMS_ATTR, None)
if isinstance(hidden, dict) and "provider_specific_fields" in hidden:
response.hidden_params.setdefault("provider_specific_fields", {}).update(
hidden["provider_specific_fields"]
)
break
_stamp_streaming_usage_cost(usage, response, logging_obj)
_set_stream_builder_response_cost(response, logging_obj)
processor.apply_provider_assembled_streaming_metadata(response, chunks, logging_obj)
return response
tool_call_chunks: Final = [
chunk
for chunk in chunks
if any(
"tool_calls" in choice["delta"] and choice["delta"]["tool_calls"] is not None
for choice in chunk.get("choices") or ()
)
]
if len(tool_call_chunks) > 0:
tool_calls_list: Final = processor.get_combined_tool_content(tool_call_chunks)
_choice = response.choices[0]
_choice.message.content = None
_choice.message.tool_calls = tool_calls_list
function_call_chunks: Final = [
chunk
for chunk in chunks
if chunk.get("choices")
and "function_call" in chunk["choices"][0]["delta"]
and chunk["choices"][0]["delta"]["function_call"] is not None
]
if len(function_call_chunks) > 0:
_choice = response.choices[0]
_choice.message.content = None
_choice.message.function_call = processor.get_combined_function_call_content(function_call_chunks)
content_chunks: Final = [
chunk
for chunk in chunks
if chunk.get("choices")
and "content" in chunk["choices"][0]["delta"]
and chunk["choices"][0]["delta"]["content"] is not None
]
if len(content_chunks) > 0:
response["choices"][0]["message"]["content"] = processor.get_combined_content(content_chunks)
thinking_blocks: Final = [
chunk
for chunk in chunks
if chunk.get("choices")
and "thinking_blocks" in chunk["choices"][0]["delta"]
and chunk["choices"][0]["delta"]["thinking_blocks"] is not None
]
if len(thinking_blocks) > 0:
response["choices"][0]["message"]["thinking_blocks"] = processor.get_combined_thinking_content(
thinking_blocks
)
reasoning_chunks: Final = [
chunk
for chunk in chunks
if chunk.get("choices")
and "reasoning_content" in chunk["choices"][0]["delta"]
and chunk["choices"][0]["delta"]["reasoning_content"] is not None
]
if len(reasoning_chunks) > 0:
response["choices"][0]["message"]["reasoning_content"] = processor.get_combined_reasoning_content(
reasoning_chunks
)
annotation_chunks: Final = [
chunk
for chunk in chunks
if chunk.get("choices")
and "annotations" in chunk["choices"][0]["delta"]
and chunk["choices"][0]["delta"]["annotations"] is not None
]
if len(annotation_chunks) > 0:
# Merge annotations from ALL chunks — providers may spread
# them across multiple streaming chunks or send them only in
# the final chunk.
all_annotations: Final[list] = []
for ac in annotation_chunks:
all_annotations.extend(ac["choices"][0]["delta"]["annotations"])
response["choices"][0]["message"]["annotations"] = all_annotations
audio_chunks: Final = [
chunk
for chunk in chunks
if chunk.get("choices")
and "audio" in chunk["choices"][0]["delta"]
and chunk["choices"][0]["delta"]["audio"] is not None
]
if len(audio_chunks) > 0:
_choice = response.choices[0]
_choice.message.audio = processor.get_combined_audio_content(audio_chunks)
# Handle image chunks from models like gemini-2.5-flash-image
# See: https://github.com/BerriAI/litellm/issues/19478
image_chunks: Final = [
chunk
for chunk in chunks
if chunk.get("choices")
and "images" in chunk["choices"][0]["delta"]
and chunk["choices"][0]["delta"]["images"] is not None
]
if len(image_chunks) > 0:
# Images come complete in a single chunk, collect all images from all chunks
all_images: Final = []
for chunk in image_chunks:
all_images.extend(chunk["choices"][0]["delta"]["images"])
response["choices"][0]["message"]["images"] = all_images
# Combine provider_specific_fields from streaming chunks (e.g., web_search_results, citations)
# See: https://github.com/BerriAI/litellm/issues/17737
provider_specific_chunks: Final = [
chunk
for chunk in chunks
if chunk.get("choices")
and "provider_specific_fields" in chunk["choices"][0]["delta"]
and chunk["choices"][0]["delta"]["provider_specific_fields"] is not None
]
if len(provider_specific_chunks) > 0:
provider_field_dicts: Final = tuple(
fields
for chunk in provider_specific_chunks
for fields in (chunk["choices"][0]["delta"]["provider_specific_fields"],)
if isinstance(fields, dict)
)
streamed_citations: Final = tuple(
fields["citation"] for fields in provider_field_dicts if fields.get("citation") is not None
)
citation_fields: Final = (
{"citations": _joined_streamed_citations(streamed_citations)} if streamed_citations else {}
)
combined_provider_fields: Final = {
key: value
for fields in (citation_fields, *provider_field_dicts)
for key, value in fields.items()
if key != "citation"
}
if combined_provider_fields:
_choice = response.choices[0]
_choice.message.provider_specific_fields = combined_provider_fields
completion_output = get_content_from_model_response(response)
reasoning_tokens: Final = processor.count_reasoning_tokens(response)
usage = processor.calculate_usage(
chunks=chunks,
model=model,
completion_output=completion_output,
messages=messages,
reasoning_tokens=reasoning_tokens,
count_prompt_tokens=count_prompt_tokens,
)
setattr(response, "usage", usage)
# Propagate provider_specific_fields from the last chunk (contains provider
# metadata like traffic_type set during streaming)
for chunk in reversed(chunks):
if isinstance(chunk, dict):
hidden = chunk.get("_hidden_params")
else:
hidden = getattr(chunk, HIDDEN_PARAMS_ATTR, None)
if isinstance(hidden, dict) and "provider_specific_fields" in hidden:
response.hidden_params.setdefault("provider_specific_fields", {}).update(
hidden["provider_specific_fields"]
)
break
_stamp_streaming_usage_cost(usage, response, logging_obj)
_set_stream_builder_response_cost(response, logging_obj)
processor.apply_provider_assembled_streaming_metadata(response, chunks, logging_obj)
return response
except Exception as e:
verbose_logger.exception("litellm.main.py::stream_chunk_builder() - Exception occurred - %s", e)
raise litellm.APIError(
status_code=500,
message="Error building chunks for logging/streaming usage calculation",
llm_provider="",
model="",
)

View file

@ -19,12 +19,12 @@ from litellm.llms.anthropic.prompt_cache_prediction import (
UnsupportedPredictionTarget,
cache_scope,
count_prompt_tokens,
lookup,
parse_prompt,
resolve_prediction_target,
supported_prediction_headers,
)
from litellm.proxy.common_utils.prompt_cache_pricing import price_cache_tokens
from litellm.proxy.hooks.prompt_cache_prediction import lookup
from litellm.llms.anthropic.prompt_cache_pricing import price_cache_tokens
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.management_endpoints.prompt_cache_prediction import (
CacheCostScenario,

View file

@ -29,6 +29,10 @@ from litellm.llms.anthropic.pass_through.adapters.transformation import (
LiteLLMAnthropicMessagesAdapter,
is_provider_native_tool_dict,
)
from litellm.llms.anthropic.pass_through.stream_assembly import (
build_complete_streaming_response,
build_usage_only_response_from_chunks,
)
from litellm.llms.base_llm.guardrail_translation.base_translation import (
BaseTranslation,
StreamingScanKey,
@ -47,9 +51,6 @@ from litellm.llms.base_llm.guardrail_translation.utils import (
stream_item_fingerprint,
unappliable_request_rewrite,
)
from litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler import (
AnthropicPassthroughLoggingHandler,
)
from litellm.types.llms.anthropic import (
AllAnthropicToolsValues,
AnthropicMessagesRequest,
@ -322,7 +323,7 @@ class AnthropicMessagesHandler(BaseTranslation):
if not chunks:
return None
try:
return AnthropicPassthroughLoggingHandler.build_usage_only_response_from_chunks(
return build_usage_only_response_from_chunks(
all_chunks=chunks,
model=str((request_data or {}).get("model") or ""),
)
@ -1226,7 +1227,7 @@ class AnthropicMessagesHandler(BaseTranslation):
has_ended: Final = self._check_streaming_has_ended(responses_so_far)
if has_ended:
# build the model response from the responses_so_far
built_response: Final = AnthropicPassthroughLoggingHandler.build_complete_streaming_response(
built_response: Final = build_complete_streaming_response(
all_chunks=responses_so_far,
litellm_logging_obj=cast("LiteLLMLoggingObj", litellm_logging_obj),
model="",

View file

@ -412,7 +412,7 @@ class LiteLLMMessagesToCompletionTransformationHandler:
return None
if resolved_provider == "litellm_proxy":
return None
from litellm.main import responses_api_bridge_check
from litellm.responses.bridge_check import responses_api_bridge_check
web_search_options: Final = completion_kwargs.get("web_search_options")
tools: Final = completion_kwargs.get("tools")

View file

@ -84,6 +84,7 @@ _PROPAGATED_METADATA_KEYS: Final = (
"user_api_key_model_max_budget",
"user_api_key_team_model_max_budget",
"user_api_key_user_model_max_budget",
"user_api_key_team_member_model_max_budget",
"user_api_key_end_user_model_max_budget",
"litellm_call_id",
"litellm_parent_otel_span",
@ -125,6 +126,23 @@ class _SummaryAcompletion(Protocol):
) -> "Awaitable[ModelResponse | CustomStreamWrapper]": ...
class _SummaryModelBudgetLimiter(Protocol):
async def is_team_member_within_model_budget(
self,
user_id: str,
team_id: str,
team_member_model_max_budget: Mapping[str, object],
model: str,
) -> bool: ...
async def is_end_user_within_model_budget(
self,
end_user_id: str,
end_user_model_max_budget: Mapping[str, object],
model: str,
) -> bool: ...
class _CreateRateLimitDescriptors(Protocol):
def __call__(
self,
@ -392,6 +410,42 @@ async def _check_summary_model_access(
return True
async def _check_summary_team_member_model_budget(
user_api_key_auth: "UserAPIKeyAuth",
model_max_budget_limiter: _SummaryModelBudgetLimiter,
summary_model: str,
) -> bool:
team_member_model_max_budget: Final = user_api_key_auth.team_member_model_max_budget
user_id: Final = user_api_key_auth.user_id
team_id: Final = user_api_key_auth.team_id
if (
not isinstance(team_member_model_max_budget, Mapping)
or not team_member_model_max_budget
or not isinstance(user_id, str)
or not isinstance(team_id, str)
):
return True
try:
await model_max_budget_limiter.is_team_member_within_model_budget(
user_id=user_id,
team_id=team_id,
team_member_model_max_budget=team_member_model_max_budget,
model=summary_model,
)
except litellm.BudgetExceededError:
return False
except Exception as e: # noqa: BLE001 # unexpected budget-check errors must deny compaction
verbose_logger.warning(
"compact_20260112: unexpected error during team member model-budget "
"check for summary_model=%s; denying: %s",
summary_model,
e,
)
return False
return True
async def _check_summary_model_budget(
user_api_key_auth: Optional["UserAPIKeyAuth"],
summary_model: str,
@ -407,9 +461,9 @@ async def _check_summary_model_budget(
per-model budget is configured.
Every scope is checked because the summary's spend is charged to every
scope: this file propagates the key, team, user and end-user budgets into the
subrequest's metadata, so skipping one of them would let compaction
increment a counter it can never be refused by.
scope: this file propagates the key, team, user, team-member and end-user
budgets into the subrequest's metadata, so skipping one of them would let
compaction increment a counter it can never be refused by.
"""
if user_api_key_auth is None:
return True
@ -497,7 +551,11 @@ async def _check_summary_model_budget(
)
return False
return True
return await _check_summary_team_member_model_budget(
user_api_key_auth=user_api_key_auth,
model_max_budget_limiter=model_max_budget_limiter,
summary_model=summary_model,
)
def _without_parallel_request_gauges(

View file

@ -45,7 +45,7 @@ _RESPONSES_API_PROVIDERS: Final = frozenset({"openai"})
def _bridges_to_responses_api(model: str, custom_llm_provider: str) -> bool:
from litellm.main import responses_api_bridge_check
from litellm.responses.bridge_check import responses_api_bridge_check
model_info, _ = responses_api_bridge_check(model=model, custom_llm_provider=custom_llm_provider)
return model_info.get("mode") == "responses"

View file

@ -0,0 +1,519 @@
import codecs
import json
from collections.abc import Mapping, Sequence
from typing import Final
import litellm
from litellm._logging import verbose_proxy_logger
from litellm.litellm_core_utils.core_helpers import map_finish_reason
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.anthropic.chat.handler import (
ModelResponseIterator as AnthropicModelResponseIterator,
)
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.types.utils import (
Choices,
Message,
ModelResponse,
TextCompletionResponse,
)
def split_sse_chunk_into_events(chunk: str | bytes) -> list[str]:
"""
Split a chunk that may contain multiple SSE events into individual events.
SSE format: "event: type\ndata: {...}\n\n"
Multiple events in a single chunk are separated by double newlines.
Args:
chunk: Raw chunk string that may contain multiple SSE events
Returns:
List of individual SSE event strings (each containing "event: X\ndata: {...}")
"""
# Handle bytes input
if isinstance(chunk, bytes):
chunk = chunk.decode("utf-8")
# Split on double newlines to separate SSE events
# Filter out empty strings
events: Final = [event.strip() for event in chunk.split("\n\n") if event.strip()]
return events
def build_complete_streaming_response(
all_chunks: Sequence[str | bytes],
litellm_logging_obj: LiteLLMLoggingObj,
model: str,
speed: str | None = None,
) -> ModelResponse | TextCompletionResponse | None:
"""
Builds complete response from raw Anthropic chunks.
Fast path: for the dominant case of a pure-text streaming response
(no tool_use / thinking / non-text content blocks), the long run of
``content_block_delta`` text deltas is collapsed into a single
equivalent SSE event before conversion. ``chunk_parser`` and
``stream_chunk_builder`` remain the single source of truth for chunk
shape, usage math and finish-reason mapping, so the rebuilt response
(and therefore the logged/billed payload) is identical -- this is
asserted by a parity test. Anything non-trivial falls back to the
unchanged legacy reconstruction.
Per-event Pydantic ``ModelResponseStream`` construction dominated
event-loop CPU under concurrent streaming; collapsing the homogeneous
text run removes O(num_output_tokens) of it.
"""
collapsed: Final = collapse_pure_text_chunks(all_chunks)
if collapsed is not None:
return build_complete_streaming_response_legacy(
all_chunks=collapsed,
litellm_logging_obj=litellm_logging_obj,
model=model,
speed=speed,
)
return build_complete_streaming_response_legacy(
all_chunks=all_chunks,
litellm_logging_obj=litellm_logging_obj,
model=model,
speed=speed,
)
# Anthropic SSE block/delta types that the fast path is NOT allowed to
# collapse -- their presence forces the unchanged legacy path so tool
# calls, thinking, citations, etc. keep byte-identical reconstruction.
_FAST_PATH_DISALLOWED_DELTA_TYPES: Final = frozenset(
{
"input_json_delta",
"thinking_delta",
"signature_delta",
"citations_delta",
}
)
def collapse_pure_text_chunks(
all_chunks: Sequence[str | bytes],
) -> list[str] | None:
"""
Return a new chunk list with the contiguous run of text-only
``content_block_delta`` events replaced by a single equivalent event,
or ``None`` if the stream is not a pure single-text-block response
(in which case the caller uses the legacy path unchanged).
Only ``message_start`` / ``content_block_start(text)`` /
``content_block_delta(text_delta)`` / ``content_block_stop`` /
``message_delta`` / ``message_stop`` / ``ping`` events are accepted.
Any other content-block type or delta type returns ``None``.
"""
normalized: Final[list[str]] = []
for raw in all_chunks:
line = raw.decode("utf-8") if isinstance(raw, bytes) else raw
for ev in line.split("\n\n"):
ev = ev.strip()
if ev:
normalized.append(ev)
text_block_indexes: Final[set] = set()
out: Final[list[str]] = []
pending_text: list[str] = []
pending_index: int | None = None
saw_any_text_delta = False
def flush() -> None:
nonlocal pending_text, pending_index
if pending_text:
merged: Final = {
"type": "content_block_delta",
"index": pending_index if pending_index is not None else 0,
"delta": {"type": "text_delta", "text": "".join(pending_text)},
}
out.append("data: " + json.dumps(merged))
pending_text = []
pending_index = None
for ev in normalized:
idx = ev.find("data:")
if idx == -1:
# Bare "event: <name>" line. The legacy converter turns this
# into an empty ModelResponseStream that contributes nothing
# to stream_chunk_builder. Drop the high-frequency interior
# markers (content_block_delta / ping); keep every other
# bare event line verbatim so chunk ordering and the
# load-bearing chunks[0] (event: message_start) are retained.
name = ev[len("event:") :].strip() if ev.startswith("event:") else ""
if name in ("content_block_delta", "ping"):
continue
flush()
out.append(ev)
continue
json_str = ev[idx + len("data:") :].strip()
try:
data = json.loads(json_str)
except (json.JSONDecodeError, ValueError):
return None
etype = data.get("type")
if etype == "content_block_start":
block = data.get("content_block") or {}
if block.get("type") != "text":
return None
text_block_indexes.add(data.get("index"))
flush()
out.append(ev)
elif etype == "content_block_delta":
delta = data.get("delta") or {}
dtype = delta.get("type")
if dtype in _FAST_PATH_DISALLOWED_DELTA_TYPES:
return None
if dtype != "text_delta":
return None
cur_index = data.get("index")
if cur_index not in text_block_indexes:
return None
# Defensive: Anthropic sends blocks strictly sequentially
# (start/deltas/stop, then next block), so pending_text from
# block N must be flushed by content_block_stop before block
# N+1's deltas arrive. If we ever see a delta whose index
# disagrees with the current pending buffer, the stream is
# interleaved -- fall back to legacy rather than risk merging
# text from different blocks under a single index.
if pending_text and pending_index is not None and cur_index != pending_index:
return None
saw_any_text_delta = True
pending_index = cur_index
pending_text.append(delta.get("text") or "")
elif etype == "ping":
# Interior no-op; legacy maps it to an empty chunk.
continue
else:
# message_start / content_block_stop / message_delta /
# message_stop / error: pass through unchanged.
flush()
out.append(ev)
flush()
if not saw_any_text_delta:
return None
return out
def build_complete_streaming_response_legacy(
all_chunks: Sequence[str | bytes],
litellm_logging_obj: LiteLLMLoggingObj,
model: str,
speed: str | None = None,
) -> ModelResponse | TextCompletionResponse | None:
"""
Original reconstruction: convert every SSE event to a generic chunk
and assemble via stream_chunk_builder. Kept verbatim as the fallback
/ source of truth for the fast path's parity test.
- Splits multi-event chunks into individual SSE events
- Converts str chunks to generic chunks
- Converts generic chunks to litellm chunks (OpenAI format)
- Builds complete response from litellm chunks
"""
verbose_proxy_logger.debug("Building complete streaming response from %d chunks", len(all_chunks))
anthropic_model_response_iterator: Final = AnthropicModelResponseIterator(
streaming_response=None,
sync_stream=False,
speed=speed,
)
all_openai_chunks: Final = []
# Process each chunk - a chunk may contain multiple SSE events
for _chunk_str in all_chunks:
# Split chunk into individual SSE events
individual_events = split_sse_chunk_into_events(_chunk_str)
# Process each individual event
for event_str in individual_events:
try:
# Skip OpenAI-style [DONE] sentinels some Anthropic-compatible
# providers emit. Match the whole SSE line so a valid chunk whose
# text payload happens to contain "[DONE]" is not dropped.
if any(line.strip() == "data: [DONE]" for line in event_str.split("\n")):
continue
transformed_openai_chunk = anthropic_model_response_iterator.convert_str_chunk_to_generic_chunk(
chunk=event_str
)
if transformed_openai_chunk is not None:
all_openai_chunks.append(transformed_openai_chunk)
except (StopIteration, StopAsyncIteration):
break
except json.JSONDecodeError:
# Some upstreams emit non-JSON SSE lines; skip them so the
# logging pipeline is not broken by a single bad frame.
verbose_proxy_logger.debug(
"Skipping non-JSON SSE event: %s",
event_str[:200],
)
continue
complete_streaming_response: Final = litellm.stream_chunk_builder(
chunks=all_openai_chunks,
logging_obj=litellm_logging_obj,
)
verbose_proxy_logger.debug("Complete streaming response built: %s", complete_streaming_response)
return complete_streaming_response
def extract_sse_data(event_str: str) -> dict | None:
"""Parse the JSON object from the ``data:`` line of an Anthropic SSE event."""
for line in event_str.splitlines():
stripped = line.strip()
if stripped.startswith("data:"):
payload = stripped[len("data:") :].strip()
if not payload or payload == "[DONE]":
return None
try:
return json.loads(payload)
except (ValueError, TypeError):
return None
return None
def build_usage_only_response_from_chunks(
all_chunks: Sequence[str | bytes],
model: str,
speed: str | None = None,
) -> ModelResponse | None:
"""
Build a usage-bearing ModelResponse from Anthropic SSE token-usage events, for
cost tracking when stream_chunk_builder cannot reassemble the stream.
Anthropic emits usage in ``message_start`` (uncached input + cache tokens, and an
initial output_tokens) and the final ``message_delta`` (cumulative output_tokens)
regardless of the content/tool shape, so cost is recoverable even when full
content assembly fails. Returns ``None`` if no usage event is found.
"""
input_tokens = 0
cache_read = 0
cache_creation = 0
cache_creation_5m: int | None = None
cache_creation_1h: int | None = None
output_tokens = 0
web_search_requests: int | None = None
tool_search_requests: int | None = None
inference_geo: str | None = None
speed_from_stream: str | None = None
stop_reason: str | None = None
found_usage = False
resolved_model = model
for _chunk_str in all_chunks:
for event_str in split_sse_chunk_into_events(_chunk_str):
data = extract_sse_data(event_str)
if not data:
continue
event_type = data.get("type")
if event_type == "message_start":
message = data.get("message") or {}
if not resolved_model or resolved_model == "unknown":
resolved_model = message.get("model") or resolved_model
usage = message.get("usage") or {}
input_tokens = usage.get("input_tokens") or input_tokens
cache_read = usage.get("cache_read_input_tokens") or cache_read
cache_creation = usage.get("cache_creation_input_tokens") or cache_creation
_cc = usage.get("cache_creation")
if isinstance(_cc, dict):
cache_creation_5m = _cc.get("ephemeral_5m_input_tokens")
cache_creation_1h = _cc.get("ephemeral_1h_input_tokens")
if usage.get("inference_geo") is not None:
inference_geo = usage.get("inference_geo")
if isinstance(usage.get("speed"), str):
speed_from_stream = usage.get("speed")
if usage.get("output_tokens") is not None:
output_tokens = usage.get("output_tokens")
found_usage = True
elif event_type == "message_delta":
_delta_stop = (data.get("delta") or {}).get("stop_reason")
if _delta_stop:
stop_reason = _delta_stop
usage = data.get("usage") or {}
if usage.get("output_tokens") is not None:
output_tokens = usage.get("output_tokens")
_stu = usage.get("server_tool_use")
if isinstance(_stu, dict):
if _stu.get("web_search_requests") is not None:
web_search_requests = _stu.get("web_search_requests")
if _stu.get("tool_search_requests") is not None:
tool_search_requests = _stu.get("tool_search_requests")
if usage.get("cache_read_input_tokens") is not None:
cache_read = usage.get("cache_read_input_tokens")
if usage.get("inference_geo") is not None:
inference_geo = usage.get("inference_geo")
if isinstance(usage.get("speed"), str):
speed_from_stream = usage.get("speed")
found_usage = True
if not found_usage:
return None
# If only the 5m/1h split was provided, derive the cache_creation total from it.
if not cache_creation and (cache_creation_5m or cache_creation_1h):
cache_creation = (cache_creation_5m or 0) + (cache_creation_1h or 0)
# build usage via the same AnthropicConfig.calculate_usage path the success
# cases use, so prompt_tokens are cache-inclusive and cache / server_tool_use /
# inference_geo tokens are priced instead of left at $0
usage_object: Final[dict] = {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
}
if cache_read:
usage_object["cache_read_input_tokens"] = cache_read
if cache_creation:
usage_object["cache_creation_input_tokens"] = cache_creation
if cache_creation_5m is not None or cache_creation_1h is not None:
usage_object["cache_creation"] = {
"ephemeral_5m_input_tokens": cache_creation_5m or 0,
"ephemeral_1h_input_tokens": cache_creation_1h or 0,
}
if web_search_requests is not None or tool_search_requests is not None:
_server_tool_use: Final[dict] = {}
if web_search_requests is not None:
_server_tool_use["web_search_requests"] = web_search_requests
if tool_search_requests is not None:
_server_tool_use["tool_search_requests"] = tool_search_requests
usage_object["server_tool_use"] = _server_tool_use
if inference_geo is not None:
usage_object["inference_geo"] = inference_geo
if speed_from_stream is not None:
usage_object["speed"] = speed_from_stream
usage_obj: Final = AnthropicConfig().calculate_usage(usage_object=usage_object, reasoning_content=None, speed=speed)
return ModelResponse(
model=resolved_model,
choices=[
Choices(
finish_reason=(map_finish_reason(stop_reason) if stop_reason else "stop"),
index=0,
message=Message(role="assistant", content=""),
)
],
usage=usage_obj,
)
_ANTHROPIC_EVENT_TYPES: Final = frozenset(
{
"message_start",
"message_delta",
"message_stop",
"content_block_start",
"content_block_delta",
"content_block_stop",
"ping",
"error",
}
)
def is_raw_sse_stream(all_chunks: Sequence[object]) -> bool:
return any(isinstance(chunk, (str, bytes)) for chunk in all_chunks)
def _joined_sse_stream(all_chunks: Sequence[object]) -> str | None:
raw: Final = b"".join(
chunk if isinstance(chunk, bytes) else chunk.encode("utf-8")
for chunk in all_chunks
if isinstance(chunk, (str, bytes))
)
try:
return codecs.getincrementaldecoder("utf-8")().decode(raw, final=False)
except UnicodeDecodeError:
return None
def _parsed_sse_events(sse_stream: str) -> tuple[Mapping[str, object], ...]:
return tuple(
event_data
for event in split_sse_chunk_into_events(sse_stream)
if (event_data := extract_sse_data(event)) is not None
)
def _anthropic_message_start(sse_stream: str) -> Mapping[str, object] | None:
return next(
(
message
for event_data in _parsed_sse_events(sse_stream)
if event_data.get("type") == "message_start" and isinstance(message := event_data.get("message"), dict)
),
None,
)
def is_anthropic_sse_stream(all_chunks: Sequence[object]) -> bool:
"""Whether raw SSE frames are Anthropic Messages events.
``is_raw_sse_stream`` only says the chunks are unparsed bytes, and ``/v1/messages`` is not the
only endpoint that streams those: the Google ``:streamGenerateContent`` route marks its own
stream raw too. Reading its frames as Anthropic ones would refuse the response in a wire format
its client cannot parse, so the surface is decided on the event types actually present.
"""
sse_stream: Final = _joined_sse_stream(all_chunks)
if sse_stream is None:
return False
return any(event.get("type") in _ANTHROPIC_EVENT_TYPES for event in _parsed_sse_events(sse_stream))
def assemble_anthropic_sse_stream(
all_chunks: Sequence[object], *, restore_identity: bool = False
) -> ModelResponse | None:
"""Assemble raw Anthropic SSE frames into a ModelResponse.
``restore_identity`` stamps the upstream message id and model onto the result, which the
assembler does not carry through. It is off by default so callers that re-emit the assembled
response keep the wire shape they had before this helper was shared. The writes land on a
freshly built object that is unreachable from caller state until returned.
"""
sse_stream: Final = _joined_sse_stream(all_chunks)
if sse_stream is None:
return None
message_start: Final = _anthropic_message_start(sse_stream)
if message_start is None:
return None
model: Final = message_start.get("model") if restore_identity else None
try:
assembled: Final = build_complete_streaming_response(
all_chunks=(sse_stream,),
litellm_logging_obj=None, # pyright: ignore[reportArgumentType] # only forwarded to stream_chunk_builder, which accepts None
model=model if isinstance(model, str) else "",
)
except Exception: # noqa: BLE001 # stream_chunk_builder re-raises every assembly failure as litellm.APIError
return None
if not isinstance(assembled, ModelResponse):
return None
if not restore_identity:
return assembled
message_id: Final = message_start.get("id")
if isinstance(message_id, str):
assembled.id = message_id
if isinstance(model, str) and model:
assembled.model = model
return assembled
def is_sse_error_stream(all_chunks: Sequence[object]) -> bool:
"""Whether the buffered stream carries nothing but error frames.
post_call guardrails run in a chain, so a hook can be handed the terminal error frames an
earlier guardrail emitted when it blocked. Those carry no message to assemble, and replacing
them would hide the refusal the client is owed. Covers both wire forms a guardrail emits: the
Anthropic ``error`` event and the chat-completions ``{"error": ...}`` payload.
"""
if not all(isinstance(chunk, (str, bytes)) for chunk in all_chunks):
# A stream mixing typed chunks with an error frame still carries content to scan, and the
# frames-only join below would drop exactly the part that has to be scanned
return False
sse_stream: Final = _joined_sse_stream(all_chunks)
if sse_stream is None:
return False
events: Final = _parsed_sse_events(sse_stream)
return len(events) > 0 and all(
event.get("type") == "error" or isinstance(event.get("error"), Mapping) for event in events
)

View file

@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio
import hashlib
import json
import time
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from functools import reduce
@ -14,6 +15,7 @@ import httpx
from pydantic import ConfigDict, Field, JsonValue, StrictInt, TypeAdapter, ValidationError
import litellm
from litellm.caching.dual_cache import DualCache
from litellm.litellm_core_utils.dot_notation_indexing import delete_nested_value
from litellm.llms.anthropic.common_utils import (
AnthropicModelInfo,
@ -771,3 +773,50 @@ def parse_observed_cache(
cached_tokens=cache_tokens,
cache_creation_tokens=usage.cache_creation_tokens,
)
_RETENTION_SECONDS: Final = 86_400
class CacheObservation(LiteLLMBaseModel):
model_config = ConfigDict(extra="forbid", frozen=True, strict=True)
fingerprint: str = Field(pattern=r"^[0-9a-f]{64}$")
cached_tokens: int = Field(gt=0)
observed_at: float = Field(ge=0, allow_inf_nan=False)
expires_at: float = Field(ge=0, allow_inf_nan=False)
_CACHE_ENTRY: Final[TypeAdapter[CacheObservation | str | None]] = TypeAdapter(CacheObservation | str | None)
def _cache_key(scope: str, fingerprint: str) -> str:
return f"prompt-cache-observation:{scope}:{fingerprint}"
async def lookup(
cache: DualCache, scope: str, prefix: PromptPrefix, now: float | None = None
) -> CacheObservation | None:
checked_at: Final = time.time() if now is None else now
exact: Final = await _read_exact(cache, scope, prefix.fingerprint)
if exact is not None and exact.expires_at > checked_at:
return exact
older: Final = await asyncio.gather(
*(_read_exact(cache, scope, fingerprint) for fingerprint in prefix.fingerprints[1:])
)
observations: Final = tuple(observation for observation in (exact, *older) if observation is not None)
return next(
(observation for observation in observations if observation.expires_at > checked_at),
next(iter(observations), None),
)
async def _read_exact(cache: DualCache, scope: str, fingerprint: str) -> CacheObservation | None:
try:
value: Final = _CACHE_ENTRY.validate_python(await cache.async_get_cache(_cache_key(scope, fingerprint), ttl=1)) # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # validate the legacy cache's untyped result at the I/O boundary
if value is None:
return None
observation: Final = CacheObservation.model_validate_json(value) if isinstance(value, str) else value
except ValidationError:
return None
return observation if observation.fingerprint == fingerprint else None

View file

@ -0,0 +1,100 @@
from collections.abc import Mapping
from datetime import datetime, timezone
from math import isfinite
from typing import Final
from pydantic import TypeAdapter
import litellm
from litellm.cost_calculator import (
completion_cost, # pyright: ignore[reportUnknownVariableType] # legacy optional parameters are untyped
select_model_name_for_cost_calc,
)
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.types.management_endpoints.prompt_cache_prediction import CacheTokenBuckets
from litellm.types.utils import CacheCreationTokenDetails, ModelResponse, PromptTokensDetailsWrapper, Usage
_PRICE_ENTRY: Final = TypeAdapter(Mapping[str, object])
def _valid_price(value: object) -> bool:
return isinstance(value, (int, float)) and not isinstance(value, bool) and isfinite(value) and value >= 0
def _has_required_prices(prices: Mapping[str, object], tokens: CacheTokenBuckets, completion_tokens: int = 0) -> bool:
required: Final = (
("input_cost_per_token", True),
("cache_read_input_token_cost", tokens.cache_read_input_tokens > 0),
("cache_creation_input_token_cost", tokens.cache_creation_5m_input_tokens > 0),
("cache_creation_input_token_cost_above_1hr", tokens.cache_creation_1h_input_tokens > 0),
("output_cost_per_token", completion_tokens > 0),
)
if any(needed and not _valid_price(prices.get(key)) for key, needed in required):
return False
return all(
_valid_price(value)
for key, value in prices.items()
if value is not None and any(needed and key.startswith(f"{base}_above_") for base, needed in required)
)
def price_cache_tokens(
model: str, deployment_id: str, tokens: CacheTokenBuckets, completion_tokens: int = 0
) -> float | None:
try:
selected_model: Final = select_model_name_for_cost_calc(
model=model,
completion_response=None,
custom_pricing=True,
custom_llm_provider="anthropic",
router_model_id=deployment_id,
)
if selected_model is None:
return None
model_info: Final = litellm.get_model_info(model=selected_model, custom_llm_provider="anthropic")
registry: Final = _PRICE_ENTRY.validate_python(litellm.model_cost) # pyright: ignore[reportUnknownMemberType] # legacy registry is validated at this boundary
price_entry: Final = registry.get(model_info["key"])
if price_entry is None:
return None
prices: Final = _PRICE_ENTRY.validate_python(price_entry)
if completion_tokens < 0 or not _has_required_prices(prices, tokens, completion_tokens):
return None
usage: Final = Usage(
prompt_tokens=tokens.total_tokens,
completion_tokens=completion_tokens,
total_tokens=tokens.total_tokens + completion_tokens,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=tokens.cache_read_input_tokens,
cache_creation_tokens=tokens.cache_creation_5m_input_tokens + tokens.cache_creation_1h_input_tokens,
cache_creation_token_details=CacheCreationTokenDetails(
ephemeral_5m_input_tokens=tokens.cache_creation_5m_input_tokens,
ephemeral_1h_input_tokens=tokens.cache_creation_1h_input_tokens,
),
),
)
logging_obj: Final = Logging(
model=model,
messages=[],
stream=False,
call_type="completion",
start_time=datetime.now(timezone.utc),
litellm_call_id="prompt-cache-prediction",
function_id="prompt-cache-prediction",
)
completion_cost(
completion_response=ModelResponse(model=model, usage=usage),
model=model,
custom_llm_provider="anthropic",
custom_pricing=True,
router_model_id=deployment_id,
litellm_logging_obj=logging_obj,
)
breakdown: Final = logging_obj.cost_breakdown
input_cost: Final = breakdown.get("input_cost") if breakdown is not None else None
output_cost: Final = breakdown.get("output_cost") if breakdown is not None else None
if input_cost is None or output_cost is None:
return None
cost: Final = input_cost + output_cost
return cost if _valid_price(cost) else None
except Exception: # noqa: BLE001 # the shared pricing owners raise plain Exception for unpriceable models
return None

View file

@ -14,7 +14,7 @@ from urllib.parse import quote, urlparse
from litellm._logging import verbose_logger
from litellm._uuid import uuid
from litellm.integrations.azure_storage.azure_storage import AzureBlobStorageLogger
from litellm.proxy.common_utils.path_utils import safe_filename
from litellm.litellm_core_utils.path_utils import safe_filename
from .storage_backend import BaseFileStorageBackend

View file

@ -29,7 +29,7 @@ def resolve_resource_owner_id(
A key with neither a user_id nor a team_id would otherwise stamp
``created_by=None`` and be locked out of its own resources, so it owns
them under its hashed token instead, using the ``key:`` scope prefix
already used by ``proxy/common_utils/resource_ownership.py``. ``None``
already used by ``litellm_core_utils/resource_ownership.py``. ``None``
means the caller has no usable identity of its own and must fall back
to team scoping, or be denied.
"""

View file

@ -92,7 +92,7 @@ def _translate_message(decoder: "AWSEventStreamDecoder", message: str) -> ModelR
def _build_logged_response(
chunks: Sequence[ModelResponseStream], litellm_logging_obj: "LiteLLMLoggingObj"
) -> Optional["CostResponseTypes"]:
from litellm.main import stream_chunk_builder
from litellm.litellm_core_utils.streaming_chunk_builder_utils import stream_chunk_builder
if len(chunks) == 0:
return None

View file

@ -146,11 +146,11 @@ class GigaChatPassthroughConfig(BasePassthroughConfig):
3. Return the model_response
"""
from litellm.litellm_core_utils.streaming_chunk_builder_utils import stream_chunk_builder
from litellm.litellm_core_utils.streaming_handler import (
convert_generic_chunk_to_model_response_stream,
generic_chunk_has_all_required_fields,
)
from litellm.main import stream_chunk_builder
from litellm.types.utils import ModelResponseStream
all_translated_chunks: Final[list[object]] = [] # mutable-ok: accumulator

Some files were not shown because too many files have changed in this diff Show more