feat(lens): analyze agent activity with a separate worker

.github/workflows/lens-worker.yml

@@ -0,0 +1,54 @@
+name: Lens Worker Image
+
+on:
+  pull_request:
+    branches: [main, litellm_oss_branch, "litellm_**"]
+    paths:
+      - deploy/lens/**
+      - litellm/proxy/engine/**
+      - .github/workflows/lens-worker.yml
+  push:
+    branches: [main, litellm_agent_engine]
+    paths:
+      - deploy/lens/**
+      - litellm/proxy/engine/**
+      - .github/workflows/lens-worker.yml
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  lens-worker-image:
+    permissions:
+      contents: read
+      packages: write
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
+        with:
+          persist-credentials: false
+      - name: Build Lens worker
+        run: docker build -f deploy/lens/Dockerfile -t lens-worker:${{ github.sha }} .
+      - name: Verify standalone imports with a read-only filesystem
+        run: >-
+          docker run --rm --network none --read-only --cap-drop ALL
+          --security-opt no-new-privileges --entrypoint python
+          lens-worker:${{ github.sha }}
+          -c 'import os; import engine.worker; assert os.getuid() == 65532'
+      - name: Publish versioned Lens worker
+        if: github.event_name != 'pull_request' && github.repository == 'BerriAI/litellm'
+        env:
+          REGISTRY_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          REGISTRY_USER: ${{ github.actor }}
+          IMAGE: ghcr.io/berriai/litellm-lens-worker:sha-${{ github.sha }}
+        run: |
+          printf '%s' "$REGISTRY_TOKEN" | docker login ghcr.io -u "$REGISTRY_USER" --password-stdin
+          docker tag lens-worker:${{ github.sha }} "$IMAGE"
+          docker push "$IMAGE"
+          printf 'Lens worker image: `%s`\n' "$IMAGE" >> "$GITHUB_STEP_SUMMARY"

deploy/lens/Dockerfile

@@ -0,0 +1,6 @@
+FROM python:3.12-slim
+WORKDIR /app
+RUN pip install --no-cache-dir httpx==0.28.1 pydantic==2.11.7
+COPY litellm/proxy/engine/__init__.py litellm/proxy/engine/models.py litellm/proxy/engine/analysis.py litellm/proxy/engine/worker.py /app/engine/
+USER 65532:65532
+CMD ["python", "-m", "engine.worker"]

deploy/lens/Dockerfile.dockerignore

@@ -0,0 +1,8 @@
+**
+!litellm/
+!litellm/proxy/
+!litellm/proxy/engine/
+!litellm/proxy/engine/__init__.py
+!litellm/proxy/engine/models.py
+!litellm/proxy/engine/analysis.py
+!litellm/proxy/engine/worker.py

deploy/lens/README.md

@@ -0,0 +1,59 @@
+# Lens worker
+
+Lens reviews recorded activity and saves evidence-linked findings in the LiteLLM dashboard under Observability, Lens (`/ui/lens/`)
+
+## Start a worker
+
+Upgrade your existing LiteLLM proxy to a release that includes Lens with PostgreSQL, agent tracing (`general_settings.tracing: {store: clickhouse}`), and ClickHouse configured through `CLICKHOUSE_URL` and a separate SELECT-only `CLICKHOUSE_READER_URL`. Enable the ClickHouse callback and request/response logging to analyze LLM requests. Lens can only inspect content you actually retain
+
+In Lens, click **Connect worker**, then **Generate setup command**. The LiteLLM address is filled in for you; change it only if the server running Docker needs a different network address. Copy the command and run it on your server. The dialog changes to **Worker connected** when the container checks in
+
+The command already contains the compatible worker image and one worker token. No separate API key, source checkout, environment file, or second LiteLLM deployment is needed. Keep the command private because it includes the token. The LiteLLM release provides the dashboard and APIs; the container only runs background analysis
+
+The dashboard and Compose file pin a verified worker image by digest. The image uses Linux amd64, and the generated command selects that platform. Worker image releases are independent of proxy releases: update the pinned image when changing their API contract. CI also publishes immutable commit tags for reproducible builds
+
+For deployments managed with Compose, download `compose.yaml` and provide `LITELLM_URL` and `LENS_WORKER_TOKEN` in an environment file. Its default image is already selected:
+
+```bash
+docker compose --env-file /path/to/lens.env -f compose.yaml up -d
+```
+
+Developers can build locally with `LENS_WORKER_IMAGE=litellm-lens-worker:local docker compose -f deploy/lens/compose.yaml -f deploy/lens/compose.build.yaml up -d --build`
+
+The worker needs outbound HTTPS access to LiteLLM. It needs no inbound ports, provider keys, direct database access, or GPU. The proxy calls your selected model through its configured router; trace content reaches that model provider. Use a model with JSON output support and known token prices. One worker handles one scan at a time and can serve multiple lenses. For more throughput, start another worker with a separate credential
+
+V1 setup, manual runs, feedback, and worker credentials are restricted to proxy administrators. Admin viewers can inspect results. Worker credentials can serve the administrator’s lenses. Revoke it in the connection dialog when retiring a worker. Redeploy the worker alongside proxy upgrades so their API versions match
+
+## Configure a lens
+
+Choose agent runs, individual LLM requests, or both. The matching-activity preview updates as you choose an application (the recorded OpenTelemetry service.name) or, for request activity, a LiteLLM model group and add metadata conditions. It shows run names, timestamps, and trace IDs; open a run to inspect its original steps before starting analysis. Suggestions come from up to 100 recent executions and may not include every recorded attribute. You can enter other exact keys and values. Leave service and filters blank for all activity your account can access. Filters are exact key/value matches, combined with AND. Trace filters match span or resource attributes on the same span. Request filters match logged metadata, including caller metadata stored under `requester_metadata`; `tag=value` matches request tags. `swarm=research` works only if your instrumentation records that attribute
+
+Write a few questions, give context about a successful run, choose a model, and set the monthly limit and sample size. Choose an initial history window from 1 hour to 30 days, in hours or days. Creation queues the first scan over that window. New lenses run once by default; opt into background monitoring for a custom interval from 1 minute to 7 days, entered in minutes, hours, or days. **Analyze now** checks activity since the last successful scan; **Recheck the last 24 hours** revisits recent history. The runs API accepts `lookback_hours` from 1 to 720 for other historical windows
+
+Pausing stops future scheduled scans; cancel the active scan separately if needed. The worker polls every 10 seconds; creating a lens or clicking Analyze now queues a scan, and due schedules are queued when the worker polls. Scans for the same lens never overlap, and its next interval starts after completion. Closing the browser does not stop the worker. Configuration edits apply to the next scan. A running scan retains its settings and selected execution IDs across retries
+
+## Read the results
+
+Needs attention shows issues, highest priority first. Patterns contains useful trends and successful behavior that may not need a fix. Each finding starts with a short explanation and a next step when useful. Expand the limitations for uncertainty and counterexamples. Evidence is grouped by run and collapsed until you need it; each quote opens the original step
+
+The Runs tab lists the actual sample frozen for the latest scan. Linked-run counts on findings include cited counterexamples, so they are not failure counts. The Scans tab shows history and coverage. Existing findings retain their original wording; the shorter summaries apply to new analysis
+
+## What a scan does
+
+The proxy selects newly received or updated executions with a two-minute settling period and a five-minute overlap. Older rows without receipt timestamps use execution end time. Overlapping scans do not increment a finding's occurrence count for the same execution ID
+
+A trace is spans sharing a trace ID within one team, not an automatically reconstructed conversation session. Requests are individual LLM calls. When both sources are enabled, requests correlated to a recorded span by response ID are excluded to reduce double counting
+
+The worker screens a deterministic sample, at most the configured 1–500 executions. For each execution it reads up to 160 spans, with 8,000 characters per span section, and splits these into model calls. It consolidates observations across batches, then investigates at most 10 candidate patterns using up to five model turns each. The dashboard shows these three stages, completed work counts, and elapsed time; progress is based on the selected sample, not every eligible execution. The investigator can read more original content from the selected executions. It has no shell, browsing, code-editing, or production-action tools
+
+Each model response must match a bounded JSON schema. A malformed response gets one repair attempt through the same budget controls; repeated invalid output fails the scan. Both the worker and proxy validate quoted evidence. Findings retain exact quotes and open the source trace or request. Resolve a finding after a fix, or dismiss it with a reason. A resolved finding reopens when new execution IDs support the same pattern; dismissed findings remain dismissed
+
+Coverage distinguishes eligible, sampled, reviewed, partial, and unassessable executions. Findings describe observations in the sample, not population-wide success rates or proven causes. A root span does not prove that a trace contains every expected span. Long, missing, redacted, or expired content limits the conclusions
+
+## Operations and limits
+
+PostgreSQL stores configurations, findings and the latest 50 jobs. Workers claim jobs with optimistic concurrency and a five-minute lease, renewed every 30 seconds. A disconnected job can be reclaimed up to three times. Cancellation stops subsequent work; a model call already in flight may finish and incur cost
+
+Before every model call, Lens reserves a conservative amount against the monthly lens budget. Successful calls reconcile to reported cost where pricing is available. Interrupted calls retain their reservation because the provider may have charged. A scan stops when the next reservation would exceed the limit, so it can stop with some budget remaining. Lens budgets are separate from virtual-key budgets; analysis calls use the proxy router directly
+
+V1 requires ClickHouse for both sources. It does not reconstruct sessions from unrelated trace IDs, guarantee exhaustive reviews, cache all per-execution observations across scans, or automatically fix agent code. Trace contents can change as late spans arrive, even though a job's selected IDs are fixed. Findings should be reviewed by a person before acting on them

deploy/lens/compose.build.yaml

@@ -0,0 +1,6 @@
+services:
+  lens-worker:
+    build:
+      context: ../..
+      dockerfile: deploy/lens/Dockerfile
+    image: litellm-lens-worker:local

deploy/lens/compose.yaml

@@ -0,0 +1,10 @@
+services:
+  lens-worker:
+    image: ${LENS_WORKER_IMAGE:-ghcr.io/berriai/litellm-lens-worker@sha256:654bdb62df533402cc778d318db0e9c56ed1bce7d866a6c684c3436bedc6a7eb}
+    environment:
+      LITELLM_URL: ${LITELLM_URL:?Set the URL reachable from this container}
+      LENS_WORKER_TOKEN: ${LENS_WORKER_TOKEN:?Create a worker credential in the Lens UI}
+    restart: unless-stopped
+    read_only: true
+    cap_drop: [ALL]
+    security_opt: [no-new-privileges:true]

litellm-proxy-extras/litellm_proxy_extras/migrations/20260930000000_agent_engine/migration.sql

@@ -0,0 +1,10 @@
+CREATE TABLE "LiteLLM_Engine" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "version" INTEGER NOT NULL DEFAULT 0,
+    "data" JSONB NOT NULL
+);
+CREATE TABLE "LiteLLM_EngineWorker" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "token_hash" TEXT NOT NULL UNIQUE,
+    "data" JSONB NOT NULL
+);

litellm-proxy-extras/litellm_proxy_extras/schema.prisma

@@ -1894,3 +1894,15 @@ model LiteLLM_WorkflowMessage {
   @@unique([run_id, sequence_number])
   @@index([run_id])
 }
+
+model LiteLLM_Engine {
+  id      String @id
+  version Int @default(0)
+  data    Json
+}
+
+model LiteLLM_EngineWorker {
+  id         String @id
+  token_hash String @unique
+  data       Json
+}

litellm-rust/crates/python-bridge/src/routes/traces.rs

@@ -9,9 +9,11 @@ use pyo3::{
 
 fn map_error(error: Error) -> PyErr {
     match error {
-        Error::InvalidRow | Error::InvalidTable | Error::InvalidSchema | Error::EmptySql => {
-            PyValueError::new_err(error.to_string())
-        }
+        Error::InvalidRow
+        | Error::InvalidTable
+        | Error::InvalidSchema
+        | Error::EmptySql
+        | Error::InvalidQuery => PyValueError::new_err(error.to_string()),
         Error::InsertTooLarge => PyOverflowError::new_err(error.to_string()),
         Error::InvalidUrl
         | Error::QueryFailed(_)
@@ -91,6 +93,19 @@ impl NativeTraceStorage {
         )
     }
 
+    fn lens_query<'py>(
+        &self,
+        py: Python<'py>,
+        name: &str,
+        #[pyo3(from_py_with = litellm_host_python::from_py_argument)] parameters: BTreeMap<
+            String,
+            Parameter,
+        >,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let query = litellm_traces::LensQuery::parse(name).map_err(map_error)?;
+        self.query(py, query.sql().to_owned(), parameters)
+    }
+
     fn query<'py>(
         &self,
         py: Python<'py>,

litellm-rust/crates/traces/Cargo.toml

@@ -12,6 +12,7 @@ opentelemetry-proto = { version = "0.33.0", default-features = false, features =
 prost = "0.14.4"
 time = { workspace = true, features = ["formatting"] }
 litellm-http.workspace = true
+sha2.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true

litellm-rust/crates/traces/migrations/0008_trace_received.sql

@@ -0,0 +1 @@
+ALTER TABLE {database}.otel_traces ADD COLUMN IF NOT EXISTS EngineReceivedMs UInt64 DEFAULT 0

litellm-rust/crates/traces/migrations/0009_spend_received.sql

@@ -0,0 +1 @@
+ALTER TABLE {database}.spend_logs ADD COLUMN IF NOT EXISTS EngineReceivedMs UInt64 DEFAULT 0

litellm-rust/crates/traces/query/lens_content.sql

@@ -0,0 +1,26 @@
+SELECT * FROM (
+    SELECT SpanId AS span_id, ParentSpanId AS parent_span_id, SpanName AS name,
+        ObservationType AS kind,
+        substringUTF8(concat('Input: ',Input,'\nOutput: ',Output,'\nStatus: ',StatusCode,' ',StatusMessage),
+            {offset:UInt32},8000) AS content,
+        lengthUTF8(concat('Input: ',Input,'\nOutput: ',Output,'\nStatus: ',StatusCode,' ',StatusMessage))
+            >= {offset:UInt32}+8000 AS truncated
+    FROM otel_traces WHERE {source:String}='traces'
+      AND ({all_teams:UInt8}=1 OR TeamId={team:String})
+      AND ({key_hash:String}='' OR ApiKeyHash={key_hash:String})
+      AND ({trace_ref:String}='' OR hex(SHA256(concat(TeamId, char(0), ApiKeyHash, char(0), TraceId)))={trace_ref:String})
+      AND TraceId={id:String} AND TeamId={record_team:String} AND SpanId > {cursor:String}
+    ORDER BY SpanId LIMIT 1 BY SpanId LIMIT 40
+)
+UNION ALL
+SELECT * FROM (
+    SELECT request_id AS span_id, '' AS parent_span_id, model AS name, 'llm' AS kind,
+        substringUTF8(concat('Input: ',messages,'\nOutput: ',response,'\nError: ',error_str),
+            {offset:UInt32},8000) AS content,
+        lengthUTF8(concat('Input: ',messages,'\nOutput: ',response,'\nError: ',error_str))
+            >= {offset:UInt32}+8000 AS truncated
+    FROM spend_logs FINAL WHERE {source:String}='requests'
+      AND ({all_teams:UInt8}=1 OR team_id={team:String})
+      AND ({key_hash:String}='' OR api_key={key_hash:String})
+      AND request_id={id:String} AND team_id={record_team:String} LIMIT 1
+)

litellm-rust/crates/traces/query/lens_evidence.sql

@@ -0,0 +1,14 @@
+SELECT sum(matches) AS count FROM (
+    SELECT count() AS matches FROM otel_traces WHERE {source:String}='traces'
+      AND ({all_teams:UInt8}=1 OR TeamId={team:String})
+      AND ({key_hash:String}='' OR ApiKeyHash={key_hash:String})
+      AND ({trace_ref:String}='' OR hex(SHA256(concat(TeamId, char(0), ApiKeyHash, char(0), TraceId)))={trace_ref:String})
+      AND TraceId={id:String} AND TeamId={record_team:String} AND SpanId={span:String}
+      AND position(concat('Input: ',Input,'\nOutput: ',Output,'\nStatus: ',StatusCode,' ',StatusMessage),{quote:String})>0
+    UNION ALL
+    SELECT count() AS matches FROM spend_logs FINAL WHERE {source:String}='requests'
+      AND ({all_teams:UInt8}=1 OR team_id={team:String})
+      AND ({key_hash:String}='' OR api_key={key_hash:String})
+      AND request_id={id:String} AND team_id={record_team:String} AND request_id={span:String}
+      AND position(concat('Input: ',messages,'\nOutput: ',response,'\nError: ',error_str),{quote:String})>0
+)

litellm-rust/crates/traces/query/lens_sample.sql

@@ -0,0 +1,50 @@
+SELECT *, count() OVER () AS eligible FROM (
+    SELECT 'traces' AS source, TraceId AS trace_id, TeamId AS team_id, hex(SHA256(concat(TeamId, char(0), ApiKeyHash, char(0), TraceId))) AS trace_ref,
+        coalesce(nullIf(argMin(ResourceAttributes['run.name'], Timestamp), ''),
+            argMin(SpanName, Timestamp)) AS name, toString(min(Timestamp)) AS start_time,
+        uniqExact(SpanId) AS span_count, countIf(ParentSpanId='') > 0 AS root_seen,
+        argMin(ServiceName, Timestamp) AS service,
+        arrayZip(mapKeys(argMin(mapConcat(ResourceAttributes, SpanAttributes), tuple(ParentSpanId!='',Timestamp))),
+            mapValues(argMin(mapConcat(ResourceAttributes, SpanAttributes), tuple(ParentSpanId!='',Timestamp)))) AS attributes
+    FROM otel_traces
+    WHERE {source:String} IN ('traces','both')
+      AND ({all_teams:UInt8}=1 OR TeamId={team:String})
+      AND ({key_hash:String}='' OR ApiKeyHash={key_hash:String})
+      AND (TeamId,ApiKeyHash,TraceId) IN (
+          SELECT TeamId,ApiKeyHash,TraceId FROM otel_traces
+          WHERE ({all_teams:UInt8}=1 OR TeamId={team:String})
+            AND ({key_hash:String}='' OR ApiKeyHash={key_hash:String})
+            AND if(EngineReceivedMs>0,toInt64(EngineReceivedMs),
+                toUnixTimestamp64Milli(Timestamp)+toInt64(intDiv(Duration,1000000))) >= {start:UInt64}
+      )
+    GROUP BY TeamId,ApiKeyHash,TraceId
+    HAVING max(EngineReceivedMs) < {end:UInt64}
+       AND max(toUnixTimestamp64Milli(Timestamp)+toInt64(intDiv(Duration,1000000))) < {end:UInt64}
+       AND countIf(arrayAll((k,v) -> ResourceAttributes[k]=v OR SpanAttributes[k]=v,
+           {filter_keys:Array(String)},{filter_values:Array(String)})
+           AND ({service:String}='' OR ServiceName={service:String})) > 0
+    UNION ALL
+    SELECT 'requests' AS source, request_id AS trace_id, team_id, '' AS trace_ref, model AS name,
+        toString(start_time) AS start_time, toUInt64(1) AS span_count, toUInt8(1) AS root_seen,
+        model_group AS service,
+        arrayConcat(JSONExtractKeysAndValues(metadata, 'requester_metadata', 'String'),
+            arrayMap(t -> tuple('tag', t), request_tags)) AS attributes
+    FROM spend_logs FINAL
+    WHERE {source:String} IN ('requests','both')
+      AND ({all_teams:UInt8}=1 OR team_id={team:String})
+      AND ({key_hash:String}='' OR api_key={key_hash:String})
+      AND if(EngineReceivedMs>0,toInt64(EngineReceivedMs),toUnixTimestamp64Milli(end_time)) >= {start:UInt64}
+      AND EngineReceivedMs < {end:UInt64}
+      AND toUnixTimestamp64Milli(end_time) < {end:UInt64}
+      AND arrayAll((k,v) -> JSONExtractString(metadata,k)=v
+          OR JSONExtractString(metadata,'requester_metadata',k)=v OR (k='tag' AND has(request_tags,v)),
+          {filter_keys:Array(String)},{filter_values:Array(String)})
+      AND ({service:String}='' OR model_group={service:String})
+      AND NOT has(request_tags,'litellm-engine')
+      AND ({source:String}!='both' OR (team_id,api_key,response_id) NOT IN (
+          SELECT TeamId,ApiKeyHash,LiteLLMRequestId FROM otel_traces
+          WHERE ({all_teams:UInt8}=1 OR TeamId={team:String})
+            AND ({key_hash:String}='' OR ApiKeyHash={key_hash:String}) AND LiteLLMRequestId!=''
+      ))
+)
+ORDER BY cityHash64(concat(source,team_id,trace_id)) LIMIT {limit:UInt32}

litellm-rust/crates/traces/src/error.rs

@@ -10,6 +10,8 @@ pub enum Error {
     InvalidSchema,
     #[error("SQL query must not be empty")]
     EmptySql,
+    #[error("unknown named trace query")]
+    InvalidQuery,
     #[error("ClickHouse query failed with HTTP status {0}")]
     QueryFailed(u16),
     #[error("ClickHouse insert failed with HTTP status {0}")]