refactoring

Author: Andrei Bratu

src/humanloop/client.py

@@ -8,7 +8,6 @@
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.trace import Tracer
 
-from humanloop.context import PromptContext, reset_prompt_context, set_prompt_context
 from humanloop.core.client_wrapper import SyncClientWrapper
 
 from humanloop.eval_utils import run_eval
@@ -17,8 +16,8 @@
 from humanloop.base_client import AsyncBaseHumanloop, BaseHumanloop
 from humanloop.overload import overload_call, overload_log
 from humanloop.utilities.flow import flow as flow_decorator_factory
-from humanloop.utilities.prompt import prompt
-from humanloop.utilities.tool import tool as tool_decorator_factory
+from humanloop.utilities.prompt import prompt_decorator_factory
+from humanloop.utilities.tool import tool_decorator_factory as tool_decorator_factory
 from humanloop.environment import HumanloopEnvironment
 from humanloop.evaluations.client import EvaluationsClient
 from humanloop.otel import instrument_provider
@@ -223,9 +222,7 @@ def call_llm(messages):
 
         :param prompt_kernel: Attributes that define the Prompt. See `class:DecoratorPromptKernelRequestParams`
         """
-
-        with prompt(path=path, template=template):
-            yield
+        return prompt_decorator_factory(path=path, template=template)
 
     def tool(
         self,

src/humanloop/eval_utils/run.py

@@ -104,6 +104,10 @@ def run_eval(
     :param workers: the number of threads to process datapoints using your function concurrently.
     :return: per Evaluator checks.
     """
+    if workers > 32:
+        logger.warning("Too many workers requested, capping the number to 32.")
+    workers = min(workers, 32)
+
     evaluators_worker_pool = ThreadPoolExecutor(max_workers=workers)
 
     file_ = _file_or_file_inside_hl_utility(file)
@@ -183,7 +187,7 @@ def upload_callback(log_id: str):
             start_time = datetime.now()
             try:
                 output = _call_function(function_, hl_file.type, dp)
-                if not _callable_is_hl_utility(file):
+                if not _callable_is_decorated(file):
                     # function_ is a plain callable so we need to create a Log
                     log_func(
                         inputs=dp.inputs,
@@ -277,7 +281,7 @@ class _LocalEvaluator:
     function: Callable
 
 
-def _callable_is_hl_utility(file: File) -> bool:
+def _callable_is_decorated(file: File) -> bool:
     """Check if a File is a decorated function."""
     return hasattr(file["callable"], "file")
 
@@ -348,34 +352,29 @@ def _get_checks(
 
 
 def _file_or_file_inside_hl_utility(file: File) -> File:
-    if _callable_is_hl_utility(file):
+    if _callable_is_decorated(file):
         # When the decorator inside `file` is a decorated function,
         # we need to validate that the other parameters of `file`
         # match the attributes of the decorator
+        decorated_fn_name = file["callable"].__name__
         inner_file: File = file["callable"].file
-        if "path" in file and inner_file["path"] != file["path"]:
-            raise ValueError(
-                "`path` attribute specified in the `file` does not match the File path of the decorated function."
-            )
-        if "version" in file and inner_file["version"] != file["version"]:
-            raise ValueError(
-                "`version` attribute in the `file` does not match the File version of the decorated function."
-            )
-        if "type" in file and inner_file["type"] != file["type"]:
-            raise ValueError(
-                "`type` attribute of `file` argument does not match the File type of the decorated function."
-            )
-        if "id" in file:
-            raise ValueError("Do not specify an `id` attribute in `file` argument when using a decorated function.")
-        # file on decorated function holds at least
-        # or more information than the `file` argument
+        for argument in ["version", "path", "type", "id"]:
+            if argument in file:
+                logger.warning(
+                    f"Argument `file.{argument}` will be ignored: "
+                    f"callable `{decorated_fn_name}` is managed by "
+                    "the @{inner_file['type']} decorator."
+                )
+
+        # Use the file manifest in the decorated function
         file_ = copy.deepcopy(inner_file)
+
     else:
+        # Simple function
+        # Raise error if one of path or id not provided
         file_ = file
-
-    # Raise error if one of path or id not provided
-    if not file_.get("path") and not file_.get("id"):
-        raise ValueError("You must provide a path or id in your `file`.")
+        if not file_.get("path") and not file_.get("id"):
+            raise ValueError("You must provide a path or id in your `file`.")
 
     return file_
 

src/humanloop/otel/exporter.py

@@ -26,29 +26,6 @@
 
 
 class HumanloopSpanExporter(SpanExporter):
-    """Upload Spans created by SDK decorators to Humanloop.
-
-    Spans not created by Humanloop SDK decorators will be dropped.
-
-    Each Humanloop Span contains information about the File to log against and
-    the Log to create. We are using the .log actions that pass the kernel in the
-    request. This allows us to create new Versions if the decorated function
-    is changed.
-
-    The exporter uploads Spans top-to-bottom, where a Span is uploaded only after
-    its parent Span has been uploaded. This is necessary for Flow Traces, where
-    the parent Span is a Flow Log and the children are the Logs in the Trace.
-
-    The exporter keeps an upload queue and only uploads a Span if its direct parent has
-    been uploaded.
-    """
-
-    # NOTE: LLM Instrumentors will only intercept calls to the provider made via the
-    # official libraries e.g. import openai from openai. This is 100% the reason why
-    # prompt call is not intercepted by the Instrumentor. The way to fix this is likely
-    # overriding the hl_client.prompt.call utility. @James I'll do this since it will
-    # involve looking at the EvaluationContext deep magic.
-
     DEFAULT_NUMBER_THREADS = 4
 
     def __init__(
@@ -62,8 +39,6 @@ def __init__(
         """
         super().__init__()
         self._client = client
-        # Uploaded spans translate to a Log on Humanloop. The IDs are required to link Logs in a Flow Trace
-        self._span_to_uploaded_log_id: dict[int, Optional[str]] = {}
         # Work queue for the threads uploading the spans
         self._upload_queue: Queue = Queue()
         # Worker threads to export the spans
@@ -81,9 +56,6 @@ def __init__(
         for thread in self._threads:
             thread.start()
             logger.debug("Exporter Thread %s started", thread.ident)
-        # Flow Log Span ID mapping to children Spans that must be uploaded first
-        self._spans_left_in_trace: dict[int, set[int]] = {}
-        self._traces: list[set[str]] = []
 
     def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
         if self._shutdown:
@@ -122,40 +94,29 @@ def _do_work(self):
                 continue
 
             span_to_export, evaluation_context = thread_args
-            span_file_type = span_to_export.attributes.get(HUMANLOOP_FILE_TYPE_KEY)
-            if span_file_type is None:
+            file_type = span_to_export.attributes.get(HUMANLOOP_FILE_TYPE_KEY)
+            file_path = span_to_export.attributes.get(HUMANLOOP_PATH_KEY)
+            if file_type is None:
                 raise ValueError("Span does not have type set")
 
-            if span_file_type == "flow":
-                log_args = read_from_opentelemetry_span(
-                    span=span_to_export,
-                    key=HUMANLOOP_LOG_KEY,
-                )
-                log_args = {
-                    **log_args,
-                    "log_status": "complete",
-                }
+            log_args = read_from_opentelemetry_span(
+                span=span_to_export,
+                key=HUMANLOOP_LOG_KEY,
+            )
 
             if evaluation_context:
-                log_args = read_from_opentelemetry_span(
-                    span=span_to_export,
-                    key=HUMANLOOP_LOG_KEY,
-                )
-                span_file_path = read_from_opentelemetry_span(
-                    span=span_to_export,
-                    key=HUMANLOOP_PATH_KEY,
-                )
-                if span_file_path == evaluation_context.path:
+                if file_path == evaluation_context.path:
                     log_args = {
                         **log_args,
                         "source_datapoint_id": evaluation_context.source_datapoint_id,
                         "run_id": evaluation_context.run_id,
                     }
-                    write_to_opentelemetry_span(
-                        span=span_to_export,
-                        key=HUMANLOOP_LOG_KEY,
-                        value=log_args,
-                    )
+
+            write_to_opentelemetry_span(
+                span=span_to_export,
+                key=HUMANLOOP_LOG_KEY,
+                value=log_args,
+            )
 
             response = requests.post(
                 f"{self._client._client_wrapper.get_base_url()}/import/otel",
@@ -166,7 +127,7 @@ def _do_work(self):
                 # TODO: handle
                 pass
             else:
-                if evaluation_context and span_file_path == evaluation_context.path:
+                if evaluation_context and file_path == evaluation_context.path:
                     log_id = response.json()["log_id"]
                     evaluation_context.callback(log_id)
 

src/humanloop/otel/processor.py

@@ -23,39 +23,24 @@ class CompletableSpan(TypedDict):
 
 
 class HumanloopSpanProcessor(SimpleSpanProcessor):
-    """Enrich Humanloop spans with data from their children spans.
-
-    The decorators add Instrumentors to the OpenTelemetry TracerProvider
-    that log interactions with common LLM libraries. These Instrumentors
-    produce Spans which contain information that can be used to enrich the
-    Humanloop File Kernels.
-
-    For example, Instrumentors for LLM provider libraries intercept
-    hyperparameters used in the API call to the model to build the
-    Prompt File definition when using the @prompt decorator.
-
-    Spans created that are not created by Humanloop decorators, such as
-    those created by the Instrumentors mentioned above, will be passed
-    to the Exporter as they are.
-    """
-
     def __init__(self, exporter: SpanExporter) -> None:
         super().__init__(exporter)
 
     def on_start(self, span: Span, parent_context):
         if is_llm_provider_call(span):
-            context = get_prompt_context()
-            prompt_path, prompt_template = context.path, context.template
-            if context:
-                span.set_attribute(HUMANLOOP_PATH_KEY, context.path)
+            prompt_context = get_prompt_context()
+            if prompt_context:
+                path, template = prompt_context.path, prompt_context.template
+                span.set_attribute(HUMANLOOP_PATH_KEY, path)
                 span.set_attribute(HUMANLOOP_FILE_TYPE_KEY, "prompt")
-                if prompt_template:
+                if template:
                     span.set_attribute(
                         f"{HUMANLOOP_FILE_KEY}.template",
-                        prompt_template,
+                        template,
                     )
             else:
-                raise ValueError(f"Provider call outside @prompt context manager: {prompt_path}")
+                # TODO: handle
+                raise ValueError("Provider call outside @prompt context manager")
             trace_id = get_trace_id()
             if trace_id:
                 span.set_attribute(f"{HUMANLOOP_LOG_KEY}.trace_parent_id", trace_id)

src/humanloop/overload.py

@@ -5,7 +5,6 @@
 
 from humanloop.context import get_trace_id
 from humanloop.eval_utils.run import HumanloopUtilityError
-from humanloop.flows.client import FlowsClient
 
 from humanloop.prompts.client import PromptsClient
 from humanloop.types.create_evaluator_log_response import CreateEvaluatorLogResponse

src/humanloop/utilities/flow.py

@@ -1,13 +1,15 @@
 import logging
 from functools import wraps
-from typing import Any, Callable, Mapping, Optional, Sequence
+from typing import Any, Callable, Mapping, Optional, Sequence, TypeVar
+from typing_extensions import ParamSpec
 
 from opentelemetry.trace import Span, Tracer
 from opentelemetry import context as context_api
 import requests
 
 from humanloop.base_client import BaseHumanloop
 from humanloop.context import get_trace_id, set_trace_id
+from humanloop.types.chat_message import ChatMessage
 from humanloop.utilities.helpers import bind_args
 from humanloop.eval_utils.types import File
 from humanloop.otel.constants import (
@@ -21,6 +23,10 @@
 logger = logging.getLogger("humanloop.sdk")
 
 
+P = ParamSpec("P")
+R = TypeVar("R")
+
+
 def flow(
     client: "BaseHumanloop",
     opentelemetry_tracer: Tracer,
@@ -29,19 +35,19 @@ def flow(
 ):
     flow_kernel = {"attributes": attributes or {}}
 
-    def decorator(func: Callable):
+    def decorator(func: Callable[P, R]) -> Callable[P, R]:
         decorator_path = path or func.__name__
         file_type = "flow"
 
         @wraps(func)
-        def wrapper(*args: Sequence[Any], **kwargs: Mapping[str, Any]) -> Any:
+        def wrapper(*args: P.args, **kwargs: P.kwargs) -> Optional[R]:
             span: Span
             with opentelemetry_tracer.start_as_current_span("humanloop.flow") as span:  # type: ignore
                 trace_id = get_trace_id()
                 args_to_func = bind_args(func, args, kwargs)
 
                 # Create the trace ahead so we have a parent ID to reference
-                log_inputs = {
+                init_log_inputs = {
                     "inputs": {k: v for k, v in args_to_func.items() if k != "messages"},
                     "messages": args_to_func.get("messages"),
                     "trace_parent_id": trace_id,
@@ -53,7 +59,7 @@ def wrapper(*args: Sequence[Any], **kwargs: Mapping[str, Any]) -> Any:
                         "path": path,
                         "flow": flow_kernel,
                         "log_status": "incomplete",
-                        **log_inputs,
+                        **init_log_inputs,
                     },
                 ).json()
                 # log = client.flows.log(
@@ -66,34 +72,37 @@ def wrapper(*args: Sequence[Any], **kwargs: Mapping[str, Any]) -> Any:
                 span.set_attribute(HUMANLOOP_PATH_KEY, decorator_path)
                 span.set_attribute(HUMANLOOP_FILE_TYPE_KEY, file_type)
 
-                # Call the decorated function
+                func_output: Optional[R]
+                log_output: str
+                log_error: Optional[str]
+                log_output_message: ChatMessage
                 try:
-                    output = func(*args, **kwargs)
+                    func_output = func(*args, **kwargs)
                     if (
-                        isinstance(output, dict)
-                        and len(output.keys()) == 2
-                        and "role" in output
-                        and "content" in output
+                        isinstance(func_output, dict)
+                        and len(func_output.keys()) == 2
+                        and "role" in func_output
+                        and "content" in func_output
                     ):
-                        output_message = output
-                        output = None
+                        log_output_message = ChatMessage(**func_output)
+                        log_output = None
                     else:
-                        output = process_output(func=func, output=output)
-                        output_message = None
-                    error = None
+                        log_output = process_output(func=func, output=func_output)
+                        log_output_message = None
+                    log_error = None
                 except Exception as e:
                     logger.error(f"Error calling {func.__name__}: {e}")
                     output = None
-                    output_message = None
-                    error = str(e)
+                    log_output_message = None
+                    log_error = str(e)
 
                 flow_log = {
                     "inputs": {k: v for k, v in args_to_func.items() if k != "messages"},
                     "messages": args_to_func.get("messages"),
                     "log_status": "complete",
-                    "output": output,
-                    "error": error,
-                    "output_message": output_message,
+                    "output": log_output,
+                    "error": log_error,
+                    "output_message": log_output_message,
                     "id": init_log["id"],
                 }