Merge branch 'main' into deprecate-trace-updates

Author: hassiebp

.github/workflows/ci.yml

@@ -19,9 +19,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
-      - uses: chartboost/ruff-action@v1
-        with:
-          args: check --config ci.ruff.toml
+      - uses: astral-sh/ruff-action@v3
 
   type-checking:
     runs-on: ubuntu-latest

.pre-commit-config.yaml

@@ -1,16 +1,15 @@
 repos:
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.14.4
+    rev: v0.15.2
     hooks:
       # Run the linter and fix
-      - id: ruff
+      - id: ruff-check
         types_or: [python, pyi, jupyter]
-        args: [--fix, --config=ci.ruff.toml]
+        args: [--fix]
 
       # Run the formatter.
       - id: ruff-format
         types_or: [python, pyi, jupyter]
-        args: [--config=ci.ruff.toml]
 
   - repo: https://github.com/pre-commit/mirrors-mypy
     rev: v1.18.2

CLAUDE.md

@@ -114,8 +114,6 @@ Environment variables (defined in `_client/environment_variables.py`):
 ## Important Files
 
 - `pyproject.toml`: Poetry configuration, dependencies, and tool settings
-- `ruff.toml`: Local development linting config (stricter)
-- `ci.ruff.toml`: CI linting config (more permissive)
 - `langfuse/version.py`: Version string (updated by CI release workflow)
 
 ## API Generation

ci.ruff.toml

@@ -95,6 +95,7 @@
     Prompt_Chat,
     Prompt_Text,
     ScoreBody,
+    TraceBody,
 )
 from langfuse.batch_evaluation import (
     BatchEvaluationResult,
@@ -1839,6 +1840,39 @@ def create_score(
                 f"Error creating score: Failed to process score event for trace_id={trace_id}, name={name}. Error: {e}"
             )
 
+    def _create_trace_tags_via_ingestion(
+        self,
+        *,
+        trace_id: str,
+        tags: List[str],
+    ) -> None:
+        """Private helper to enqueue trace tag updates via ingestion API events."""
+        if not self._tracing_enabled:
+            return
+
+        if len(tags) == 0:
+            return
+
+        try:
+            new_body = TraceBody(
+                id=trace_id,
+                tags=tags,
+            )
+
+            event = {
+                "id": self.create_trace_id(),
+                "type": "trace-create",
+                "timestamp": _get_timestamp(),
+                "body": new_body,
+            }
+
+            if self._resources is not None:
+                self._resources.add_trace_task(event)
+        except Exception as e:
+            langfuse_logger.exception(
+                f"Error updating trace tags: Failed to process trace update event for trace_id={trace_id}. Error: {e}"
+            )
+
     @overload
     def score_current_span(
         self,
@@ -2866,8 +2900,10 @@ def run_batched_evaluation(
         max_retries: int = 3,
         evaluators: List[EvaluatorFunction],
         composite_evaluator: Optional[CompositeEvaluatorFunction] = None,
-        max_concurrency: int = 50,
+        max_concurrency: int = 5,
         metadata: Optional[Dict[str, Any]] = None,
+        _add_observation_scores_to_trace: bool = False,
+        _additional_trace_tags: Optional[List[str]] = None,
         resume_from: Optional[BatchEvaluationResumeToken] = None,
         verbose: bool = False,
     ) -> BatchEvaluationResult:
@@ -2909,7 +2945,7 @@ def run_batched_evaluation(
                 items matching the filter. Useful for testing or limiting evaluation runs.
                 Default: None (process all).
             max_concurrency: Maximum number of items to evaluate concurrently. Controls
-                parallelism and resource usage. Default: 50.
+                parallelism and resource usage. Default: 5.
             composite_evaluator: Optional function that creates a composite score from
                 item-level evaluations. Receives the original item and its evaluations,
                 returns a single Evaluation. Useful for weighted averages or combined metrics.
@@ -3078,6 +3114,8 @@ def composite_evaluator(*, item, evaluations):
                     max_concurrency=max_concurrency,
                     composite_evaluator=composite_evaluator,
                     metadata=metadata,
+                    _add_observation_scores_to_trace=_add_observation_scores_to_trace,
+                    _additional_trace_tags=_additional_trace_tags,
                     max_retries=max_retries,
                     verbose=verbose,
                     resume_from=resume_from,

langfuse/_client/client.py

@@ -254,6 +254,7 @@ def _initialize_instance(
         self._media_upload_queue: Queue[Any] = Queue(100_000)
         self._media_manager = MediaManager(
             api_client=self.api,
+            httpx_client=self.httpx_client,
             media_upload_queue=self._media_upload_queue,
             max_retries=3,
         )
@@ -356,6 +357,29 @@ def add_score_task(self, event: dict, *, force_sample: bool = False) -> None:
 
             return
 
+    def add_trace_task(
+        self,
+        event: dict,
+    ) -> None:
+        try:
+            langfuse_logger.debug(
+                f"Trace: Enqueuing event type={event['type']} for trace_id={event['body'].id}"
+            )
+            self._score_ingestion_queue.put(event, block=False)
+
+        except Full:
+            langfuse_logger.warning(
+                "System overload: Trace ingestion queue has reached capacity (100,000 items). Trace update will be dropped. Consider increasing flush frequency or decreasing event volume."
+            )
+
+            return
+        except Exception as e:
+            langfuse_logger.error(
+                f"Unexpected error: Failed to process trace event. The trace update will be dropped. Error details: {e}"
+            )
+
+            return
+
     @property
     def tracer(self) -> Optional[Tracer]:
         return self._otel_tracer

langfuse/_client/resource_manager.py

@@ -5,7 +5,7 @@
 from typing import Any, Callable, Optional, TypeVar, cast
 
 import backoff
-import requests
+import httpx
 from typing_extensions import ParamSpec
 
 from langfuse._client.environment_variables import LANGFUSE_MEDIA_UPLOAD_ENABLED
@@ -27,10 +27,12 @@ def __init__(
         self,
         *,
         api_client: LangfuseAPI,
+        httpx_client: httpx.Client,
         media_upload_queue: Queue,
         max_retries: Optional[int] = 3,
     ):
         self._api_client = api_client
+        self._httpx_client = httpx_client
         self._queue = media_upload_queue
         self._max_retries = max_retries
         self._enabled = os.environ.get(
@@ -252,10 +254,10 @@ def _process_upload_media_job(
 
         upload_start_time = time.time()
         upload_response = self._request_with_backoff(
-            requests.put,
+            self._httpx_client.put,
             upload_url,
             headers=headers,
-            data=data["content_bytes"],
+            content=data["content_bytes"],
         )
         upload_time_ms = int((time.time() - upload_start_time) * 1000)
 
@@ -282,7 +284,7 @@ def _should_give_up(e: Exception) -> bool:
                     and 400 <= e.status_code < 500
                     and e.status_code != 429
                 )
-            if isinstance(e, requests.exceptions.RequestException):
+            if isinstance(e, httpx.HTTPStatusError):
                 return (
                     e.response is not None
                     and e.response.status_code < 500