draft

Author: Andrei Bratu

.gitignore

@@ -4,4 +4,4 @@ __pycache__/
 poetry.toml
 .ruff_cache/
 .vscode
-.env
+.env

src/humanloop/client.py

@@ -49,7 +49,6 @@ def run(
         name: Optional[str],
         dataset: Dataset,
         evaluators: Optional[Sequence[Evaluator]] = None,
-        # logs: typing.Sequence[dict] | None = None,
         workers: int = 4,
     ) -> List[EvaluatorCheck]:
         """Evaluate your function for a given `Dataset` and set of `Evaluators`.

src/humanloop/eval_utils/run.py

@@ -212,10 +212,6 @@ def increment(self):
                 sys.stderr.write("\n")
 
 
-# Module-level so it can be shared by threads.
-_PROGRESS_BAR: Optional[_SimpleProgressBar] = None
-
-
 def run_eval(
     client: "BaseHumanloop",
     file: File,
@@ -236,7 +232,6 @@ def run_eval(
     :param workers: the number of threads to process datapoints using your function concurrently.
     :return: per Evaluator checks.
     """
-    global _PROGRESS_BAR
 
     if hasattr(file["callable"], "file"):
         # When the decorator inside `file` is a decorated function,

src/humanloop/otel/__init__.py

@@ -2,6 +2,7 @@
 
 from opentelemetry.sdk.trace import TracerProvider
 from typing_extensions import NotRequired
+from opentelemetry.sdk.trace import TracerProvider
 
 from humanloop.otel.helpers import module_is_installed
 
@@ -41,12 +42,3 @@ def instrument_provider(provider: TracerProvider):
         from opentelemetry.instrumentation.bedrock import BedrockInstrumentor
 
         BedrockInstrumentor().instrument(tracer_provider=provider)
-
-
-class FlowContext(TypedDict):
-    trace_id: NotRequired[str]
-    trace_parent_id: NotRequired[Optional[int]]
-    is_flow_log: NotRequired[bool]
-
-
-TRACE_FLOW_CONTEXT: dict[int, FlowContext] = {}

src/humanloop/otel/constants.py

@@ -4,3 +4,5 @@
 HUMANLOOP_LOG_KEY = "humanloop.log"
 HUMANLOOP_FILE_TYPE_KEY = "humanloop.file.type"
 HUMANLOOP_PATH_KEY = "humanloop.file.path"
+# Required for the exporter to know when to mark the Flow Log as complete
+HUMANLOOP_FLOW_PREREQUISITES_KEY = "humanloop.flow.prerequisites"

src/humanloop/otel/exporter.py

@@ -1,7 +1,7 @@
 import contextvars
-import json
 import logging
 import threading
+import time
 import typing
 from queue import Empty as EmptyQueue
 from queue import Queue
@@ -14,16 +14,18 @@
 
 from humanloop.core import ApiError as HumanloopApiError
 from humanloop.eval_utils.context import EVALUATION_CONTEXT_VARIABLE_NAME, EvaluationContext
-from humanloop.otel import TRACE_FLOW_CONTEXT, FlowContext
+from humanloop.otel import TRACE_FLOW_CONTEXT
 from humanloop.otel.constants import (
     HUMANLOOP_FILE_KEY,
     HUMANLOOP_FILE_TYPE_KEY,
+    HUMANLOOP_FLOW_PREREQUISITES_KEY,
     HUMANLOOP_LOG_KEY,
     HUMANLOOP_PATH_KEY,
 )
 from humanloop.otel.helpers import is_humanloop_span, read_from_opentelemetry_span
 from humanloop.requests.flow_kernel_request import FlowKernelRequestParams
 from humanloop.requests.prompt_kernel_request import PromptKernelRequestParams
+from humanloop.requests.tool_kernel_request import ToolKernelRequestParams
 
 if typing.TYPE_CHECKING:
     from humanloop.client import Humanloop
@@ -69,7 +71,8 @@ def __init__(
         for thread in self._threads:
             thread.start()
             logger.debug("Exporter Thread %s started", thread.ident)
-        self._flow_logs_to_complete: list[str] = []
+        # Flow Log Span ID mapping to children Spans that must be uploaded first
+        self._flow_log_prerequisites: dict[int, set[int]] = {}
 
     def export(self, spans: trace.Sequence[ReadableSpan]) -> SpanExportResult:
         def is_evaluated_file(
@@ -133,11 +136,6 @@ def shutdown(self) -> None:
         for thread in self._threads:
             thread.join()
             logger.debug("Exporter Thread %s joined", thread.ident)
-        for log_id in self._flow_logs_to_complete:
-            self._client.flows.update_log(
-                log_id=log_id,
-                trace_status="complete",
-            )
 
     def force_flush(self, timeout_millis: int = 3000) -> bool:
         self._shutdown = True
@@ -211,9 +209,22 @@ def _do_work(self):
                 self._upload_queue.put((span_to_export, evaluation_context))
             self._upload_queue.task_done()
 
+    def _complete_flow_log(self, span_id: int) -> None:
+        for flow_log_span_id, flow_children_span_ids in self._flow_log_prerequisites.items():
+            if span_id in flow_children_span_ids:
+                flow_children_span_ids.remove(span_id)
+                if len(flow_children_span_ids) == 0:
+                    flow_log_id = self._span_id_to_uploaded_log_id[flow_log_span_id]
+                    self._client.flows.update_log(log_id=flow_log_id, trace_status="complete")
+            break
+
     def _export_span_dispatch(self, span: ReadableSpan) -> None:
         hl_file = read_from_opentelemetry_span(span, key=HUMANLOOP_FILE_KEY)
         file_type = span._attributes.get(HUMANLOOP_FILE_TYPE_KEY)  # type: ignore
+        parent_span_id = span.parent.span_id if span.parent else None
+
+        while parent_span_id and self._span_id_to_uploaded_log_id.get(parent_span_id) is None:
+            time.sleep(0.1)
 
         if file_type == "prompt":
             export_func = self._export_prompt
@@ -242,25 +253,16 @@ def _export_prompt(self, span: ReadableSpan) -> None:
             log_object["messages"] = []
         if "tools" not in file_object["prompt"]:
             file_object["prompt"]["tools"] = []
-        trace_metadata = TRACE_FLOW_CONTEXT.get(span.get_span_context().span_id)
-        if trace_metadata and "trace_parent_id" in trace_metadata and trace_metadata["trace_parent_id"]:
-            trace_parent_id = self._span_id_to_uploaded_log_id[trace_metadata["trace_parent_id"]]
-            if trace_parent_id is None:
-                # Parent Log in Trace upload failed
-                file_path = read_from_opentelemetry_span(span, key=HUMANLOOP_PATH_KEY)
-                logger.error(f"Skipping log for {file_path}: parent Log upload failed")
-                return
-        else:
-            trace_parent_id = None
-        prompt: PromptKernelRequestParams = file_object["prompt"]
+
         path: str = file_object["path"]
-        if "output" in log_object:
-            if not isinstance(log_object["output"], str):
-                # Output expected to be a string, if decorated function
-                # does not return one, jsonify it
-                log_object["output"] = json.dumps(log_object["output"])
+        prompt: PromptKernelRequestParams = file_object["prompt"]
+
+        span_parent_id = span.parent.span_id if span.parent else None
+        trace_parent_id = self._span_id_to_uploaded_log_id[span_parent_id] if span_parent_id else None
+
         if "attributes" not in prompt or not prompt["attributes"]:
             prompt["attributes"] = {}
+
         try:
             log_response = self._client.prompts.log(
                 path=path,
@@ -271,34 +273,32 @@ def _export_prompt(self, span: ReadableSpan) -> None:
             self._span_id_to_uploaded_log_id[span.context.span_id] = log_response.id
         except HumanloopApiError:
             self._span_id_to_uploaded_log_id[span.context.span_id] = None
+        self._complete_flow_log(span_id=span.context.span_id)
 
     def _export_tool(self, span: ReadableSpan) -> None:
-        file_object: dict[str, Any] = read_from_opentelemetry_span(span, key=HUMANLOOP_FILE_KEY)
-        log_object: dict[str, Any] = read_from_opentelemetry_span(span, key=HUMANLOOP_LOG_KEY)
-        trace_metadata: FlowContext = TRACE_FLOW_CONTEXT.get(span.get_span_context().span_id, {})
-        if "trace_parent_id" in trace_metadata and trace_metadata["trace_parent_id"]:
-            trace_parent_id = self._span_id_to_uploaded_log_id.get(
-                trace_metadata["trace_parent_id"],
-            )
-            if trace_parent_id is None:
-                # Parent Log in Trace upload failed
-                file_path = read_from_opentelemetry_span(span, key=HUMANLOOP_PATH_KEY)
-                logger.error(f"Skipping log for {file_path}: parent Log upload failed")
-                return
-        else:
-            trace_parent_id = None
-        tool = file_object["tool"]
+        file_object: dict[str, Any] = read_from_opentelemetry_span(
+            span,
+            key=HUMANLOOP_FILE_KEY,
+        )
+        log_object: dict[str, Any] = read_from_opentelemetry_span(
+            span,
+            key=HUMANLOOP_LOG_KEY,
+        )
+
+        path: str = file_object["path"]
+        tool: ToolKernelRequestParams = file_object["tool"]
+
+        span_parent_id = span.parent.span_id if span.parent else None
+        trace_parent_id = self._span_id_to_uploaded_log_id[span_parent_id] if span_parent_id else None
+
+        # API expects an empty dictionary if user does not supply attributes
         if not tool.get("attributes"):
             tool["attributes"] = {}
         if not tool.get("setup_values"):
             tool["setup_values"] = {}
-        path: str = file_object["path"]
         if "parameters" in tool["function"] and "properties" not in tool["function"]["parameters"]:
             tool["function"]["parameters"]["properties"] = {}
-        if not isinstance(log_object["output"], str):
-            # Output expected to be a string, if decorated function
-            # does not return one, jsonify it
-            log_object["output"] = json.dumps(log_object["output"])
+
         try:
             log_response = self._client.tools.log(
                 path=path,
@@ -309,33 +309,34 @@ def _export_tool(self, span: ReadableSpan) -> None:
             self._span_id_to_uploaded_log_id[span.context.span_id] = log_response.id
         except HumanloopApiError:
             self._span_id_to_uploaded_log_id[span.context.span_id] = None
+        self._complete_flow_log(span_id=span.context.span_id)
 
     def _export_flow(self, span: ReadableSpan) -> None:
-        file_object: dict[str, Any] = read_from_opentelemetry_span(span, key=HUMANLOOP_FILE_KEY)
-        log_object: dict[str, Any] = read_from_opentelemetry_span(span, key=HUMANLOOP_LOG_KEY)
-        trace_metadata: FlowContext = TRACE_FLOW_CONTEXT.get(
-            span.get_span_context().span_id,
-            {},
+        file_object: dict[str, Any] = read_from_opentelemetry_span(
+            span,
+            key=HUMANLOOP_FILE_KEY,
         )
-        if "trace_parent_id" in trace_metadata:
-            trace_parent_id = self._span_id_to_uploaded_log_id.get(
-                trace_metadata["trace_parent_id"],  # type: ignore
-            )
-            if trace_parent_id is None and trace_metadata["trace_id"] != span.get_span_context().span_id:
-                # Parent Log in Trace upload failed
-                # NOTE: Check if the trace_id metadata field points to the
-                # span itself. This signifies the span is the head of the Trace
-                file_path = read_from_opentelemetry_span(span, key=HUMANLOOP_PATH_KEY)
-                logger.error(f"Skipping log for {file_path}: parent Log upload failed")
-                return
-        else:
-            trace_parent_id = None
+        log_object: dict[str, Any] = read_from_opentelemetry_span(
+            span,
+            key=HUMANLOOP_LOG_KEY,
+        )
+        # Spans that must be uploaded before the Flow Span is completed
+        prerequisites = read_from_opentelemetry_span(
+            span=span,
+            key=HUMANLOOP_FLOW_PREREQUISITES_KEY,
+        )
+        self._flow_log_prerequisites[span.context.span_id] = set(prerequisites)
+
+        path: str = file_object["path"]
         flow: FlowKernelRequestParams
         if not file_object.get("flow"):
             flow = {"attributes": {}}
         else:
             flow = file_object["flow"]
-        path: str = file_object["path"]
+
+        span_parent_id = span.parent.span_id if span.parent else None
+        trace_parent_id = self._span_id_to_uploaded_log_id[span_parent_id] if span_parent_id else None
+
         if "output" not in log_object:
             log_object["output"] = None
         try:
@@ -350,3 +351,4 @@ def _export_flow(self, span: ReadableSpan) -> None:
         except HumanloopApiError as e:
             logger.error(str(e))
             self._span_id_to_uploaded_log_id[span.context.span_id] = None
+        self._complete_flow_log(span_id=span.context.span_id)

src/humanloop/otel/helpers.py

@@ -267,13 +267,7 @@ def is_llm_provider_call(span: ReadableSpan) -> bool:
 
 def is_humanloop_span(span: ReadableSpan) -> bool:
     """Check if the Span was created by the Humanloop SDK."""
-    try:
-        # Valid spans will have keys with the HL_FILE_OT_KEY and HL_LOG_OT_KEY prefixes present
-        read_from_opentelemetry_span(span, key=HUMANLOOP_FILE_KEY)
-        read_from_opentelemetry_span(span, key=HUMANLOOP_LOG_KEY)
-    except KeyError:
-        return False
-    return True
+    return span.name.startswith("humanloop.")
 
 
 def module_is_installed(module_name: str) -> bool:
@@ -288,10 +282,6 @@ def module_is_installed(module_name: str) -> bool:
     return True
 
 
-def generate_span_id() -> str:
-    return str(uuid.uuid4())
-
-
 def jsonify_if_not_string(func: Callable, output: Any) -> str:
     if not isinstance(output, str):
         try:

src/humanloop/otel/processor.py

@@ -2,12 +2,16 @@
 from collections import defaultdict
 from typing import Any
 
-# No typing stubs for parse
 from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExporter
 from pydantic import ValidationError as PydanticValidationError
 
-from humanloop.otel.constants import HUMANLOOP_FILE_KEY, HUMANLOOP_FILE_TYPE_KEY, HUMANLOOP_LOG_KEY
+from humanloop.otel.constants import (
+    HUMANLOOP_FILE_KEY,
+    HUMANLOOP_FILE_TYPE_KEY,
+    HUMANLOOP_FLOW_PREREQUISITES_KEY,
+    HUMANLOOP_LOG_KEY,
+)
 from humanloop.otel.helpers import (
     is_humanloop_span,
     is_llm_provider_call,
@@ -40,10 +44,17 @@ def __init__(self, exporter: SpanExporter) -> None:
         super().__init__(exporter)
         # Span parent to Span children map
         self._children: dict[int, list] = defaultdict(list)
-
-    # NOTE: Could override on_start and process Flow spans ahead of time
-    # and PATCH the created Logs in on_end. A special type of ReadableSpan could be
-    # used for this
+        self._prerequisites: dict[int, list[int]] = {}
+
+    def on_start(self, span, parent_context=None):
+        span_id = span.context.span_id
+        if span.name == "humanloop.flow":
+            self._prerequisites[span_id] = []
+        if span.parent and is_humanloop_span(span):
+            parent_span_id = span.parent.span_id
+            for trace_head, all_trace_nodes in self._prerequisites.items():
+                if parent_span_id == trace_head or parent_span_id in all_trace_nodes:
+                    all_trace_nodes.append(span_id)
 
     def on_end(self, span: ReadableSpan) -> None:
         if is_humanloop_span(span=span):
@@ -57,6 +68,12 @@ def on_end(self, span: ReadableSpan) -> None:
                 # arrives in order to enrich it
                 self._children[span.parent.span_id].append(span)
         # Pass the Span to the Exporter
+        if span.name == "humanloop.flow":
+            write_to_opentelemetry_span(
+                span=span,
+                key=HUMANLOOP_FLOW_PREREQUISITES_KEY,
+                value=self._prerequisites[span.context.span_id],
+            )
         self.span_exporter.export([span])