Use /otel endpoint to process endpoints

Author: Andrei Bratu

src/humanloop/client.py

@@ -8,15 +8,16 @@
 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.run import prompt_call_evaluation_aware
 
-from humanloop.eval_utils import log_with_evaluation_context, run_eval
+from humanloop.eval_utils import run_eval
 from humanloop.eval_utils.types import Dataset, Evaluator, EvaluatorCheck, File
 
 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 as prompt_decorator_factory
+from humanloop.utilities.prompt import prompt
 from humanloop.utilities.tool import tool as tool_decorator_factory
 from humanloop.environment import HumanloopEnvironment
 from humanloop.evaluations.client import EvaluationsClient
@@ -25,8 +26,6 @@
 from humanloop.otel.processor import HumanloopSpanProcessor
 from humanloop.prompt_utils import populate_template
 from humanloop.prompts.client import PromptsClient
-from humanloop.requests.flow_kernel_request import FlowKernelRequestParams
-from humanloop.requests.tool_kernel_request import ToolKernelRequestParams
 
 
 class ExtendedEvalsClient(EvaluationsClient):
@@ -119,9 +118,10 @@ def __init__(
         self.prompts = ExtendedPromptsClient(client_wrapper=self._client_wrapper)
 
         # Overload the .log method of the clients to be aware of Evaluation Context
-        self.prompts = log_with_evaluation_context(client=self.prompts)
-        self.prompts = prompt_call_evaluation_aware(client=self.prompts)
-        self.flows = log_with_evaluation_context(client=self.flows)
+        self.prompts = overload_log(client=self.prompts)
+        self.prompts = overload_call(client=self.prompts)
+        self.flows = overload_log(client=self.flows)
+        self.tools = overload_log(client=self.tools)
 
         if opentelemetry_tracer_provider is not None:
             self._tracer_provider = opentelemetry_tracer_provider
@@ -135,10 +135,7 @@ def __init__(
             )
         instrument_provider(provider=self._tracer_provider)
         self._tracer_provider.add_span_processor(
-            HumanloopSpanProcessor(
-                client=self,
-                exporter=HumanloopSpanExporter(client=self),
-            ),
+            HumanloopSpanProcessor(exporter=HumanloopSpanExporter(client=self)),
         )
 
         if opentelemetry_tracer is None:
@@ -151,6 +148,7 @@ def prompt(
         self,
         *,
         path: str,
+        template: Optional[str] = None,
     ):
         """Decorator for declaring a [Prompt](https://humanloop.com/docs/explanation/prompts) in code.
 
@@ -225,14 +223,16 @@ def call_llm(messages):
 
         :param prompt_kernel: Attributes that define the Prompt. See `class:DecoratorPromptKernelRequestParams`
         """
-        return prompt_decorator_factory(path=path)
+
+        with prompt(path=path, template=template):
+            yield
 
     def tool(
         self,
         *,
         path: str,
-        attributes: dict[str, Any] | None = None,
-        setup_values: dict[str, Any] | None = None,
+        attributes: Optional[dict[str, Any]] = None,
+        setup_values: Optional[dict[str, Any]] = None,
     ):
         """Decorator for declaring a [Tool](https://humanloop.com/docs/explanation/tools) in code.
 
@@ -312,8 +312,8 @@ def calculator(a: int, b: Optional[int]) -> int:
     def flow(
         self,
         *,
-        path: str = None,
-        attributes: dict[str, Any] | None = None,
+        path: str,
+        attributes: Optional[dict[str, Any]] = None,
     ):
         """Decorator for declaring a [Flow](https://humanloop.com/docs/explanation/flows) in code.
 
@@ -366,6 +366,7 @@ def entrypoint():
         :param flow_kernel: Attributes that define the Flow. See `class:ToolKernelRequestParams`
         """
         return flow_decorator_factory(
+            client=self,
             opentelemetry_tracer=self._opentelemetry_tracer,
             path=path,
             attributes=attributes,

src/humanloop/context.py

@@ -15,7 +15,6 @@
 
 def get_trace_id() -> Optional[str]:
     key = hash((HUMANLOOP_CONTEXT_TRACE_ID, threading.get_ident()))
-    context_api.get_value()
     return context_api.get_value(key=key)
 
 
@@ -28,16 +27,27 @@ def reset_trace_id_context(token: ResetToken):
     context_api.detach(token=token)
 
 
-def set_prompt_path(path: str) -> ResetToken:
+@dataclass
+class PromptContext:
+    path: str
+    template: Optional[str]
+
+
+def set_prompt_context(prompt_context: PromptContext) -> ResetToken:
     key = hash((HUMANLOOP_CONTEXT_PROMPT_PATH, threading.get_ident()))
-    return context_api.set_value(key=key, value=path)
+    return context_api.attach(
+        context_api.set_value(
+            key=key,
+            value=prompt_context,
+        )
+    )
 
 
-def reset_prompt_path(token: ResetToken):
+def reset_prompt_context(token: ResetToken):
     context_api.detach(token=token)
 
 
-def get_prompt_path() -> Optional[str]:
+def get_prompt_context() -> Optional[PromptContext]:
     key = hash((HUMANLOOP_CONTEXT_PROMPT_PATH, threading.get_ident()))
     return context_api.get_value(key)
 
@@ -51,9 +61,9 @@ class EvaluationContext:
     path: str
 
 
-def set_evaluation_context(evaluation_context: EvaluationContext):
+def set_evaluation_context(evaluation_context: EvaluationContext) -> ResetToken:
     key = hash((HUMANLOOP_CONTEXT_EVALUATION, threading.get_ident()))
-    context_api.set_value(key, evaluation_context)
+    return context_api.attach(context_api.set_value(key, evaluation_context))
 
 
 def get_evaluation_context() -> Optional[EvaluationContext]:

src/humanloop/eval_utils/__init__.py

@@ -1,4 +1,4 @@
-from .run import log_with_evaluation_context, run_eval
+from .run import run_eval
 from .types import File
 
-__all__ = ["run_eval", "log_with_evaluation_context", "File"]
+__all__ = ["run_eval", "File"]

src/humanloop/eval_utils/run.py

@@ -166,7 +166,7 @@ def upload_callback(log_id: str):
             set_evaluation_context(
                 EvaluationContext(
                     source_datapoint_id=dp.id,
-                    upload_callback=upload_callback,
+                    callback=upload_callback,
                     file_id=hl_file.id,
                     run_id=run.id,
                     path=hl_file.path,
@@ -741,32 +741,32 @@ def _run_local_evaluators(
         else:
             log_dict = log
     datapoint_dict = datapoint.dict() if datapoint else None
-    for local_evaluator, eval_function in local_evaluators:
+    for local_evaluator in local_evaluators:
         start_time = datetime.now()
         try:
-            if local_evaluator.spec.arguments_type == "target_required":
-                judgement = eval_function(
+            if local_evaluator.hl_evaluator.spec.arguments_type == "target_required":
+                judgement = local_evaluator.function(
                     log_dict,
                     datapoint_dict,
                 )
             else:
-                judgement = eval_function(log_dict)
+                judgement = local_evaluator.function(log_dict)
 
             _ = client.evaluators.log(
-                version_id=local_evaluator.version_id,
+                version_id=local_evaluator.hl_evaluator.version_id,
                 parent_id=log_id,
                 judgment=judgement,
-                id=local_evaluator.id,
+                id=local_evaluator.hl_evaluator.id,
                 start_time=start_time,
                 end_time=datetime.now(),
             )
         except Exception as e:
             _ = client.evaluators.log(
                 parent_id=log_id,
-                id=local_evaluator.id,
+                id=local_evaluator.hl_evaluator.id,
                 error=str(e),
                 start_time=start_time,
                 end_time=datetime.now(),
             )
-            logger.warning(f"\nEvaluator {local_evaluator.path} failed with error {str(e)}")
+            logger.warning(f"\nEvaluator {local_evaluator.hl_evaluator.path} failed with error {str(e)}")
     progress_bar.increment()

src/humanloop/otel/constants.py

@@ -5,10 +5,7 @@
 HUMANLOOP_LOG_ID_KEY = "humanloop.log_id"
 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"
-HUMANLOOP_INTERCEPTED_PROMPT_CALL_SPAN_NAME = "humanloop_intercepted_hl_call"
-HUMANLOOP_INTERCEPTED_PROMPT_CALL_RESPONSE = "intercepted_call_response"
+# Opentelemetry context
 HUMANLOOP_CONTEXT_PROMPT_PATH = "humanloop.context.prompt.path"
 HUMANLOOP_CONTEXT_TRACE_ID = "humanloop.context.flow.trace_id"
 HUMANLOOP_CONTEXT_EVALUATION = "humanloop.context.evaluation"

src/humanloop/otel/exporter.py

@@ -9,6 +9,7 @@
 from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
 
+import requests
 from humanloop.context import get_evaluation_context, EvaluationContext
 from humanloop.otel.constants import (
     HUMANLOOP_FILE_TYPE_KEY,
@@ -156,8 +157,17 @@ def _do_work(self):
                         value=log_args,
                     )
 
-            id = self._client.export(span_to_export.to_json())
-            if evaluation_context:
-                evaluation_context.callback(id)
+            response = requests.post(
+                f"{self._client._client_wrapper.get_base_url()}/import/otel",
+                headers=self._client._client_wrapper.get_headers(),
+                data=span_to_export.to_json().encode("ascii"),
+            )
+            if response.status_code != 200:
+                # TODO: handle
+                pass
+            else:
+                if evaluation_context and span_file_path == evaluation_context.path:
+                    log_id = response.json()["log_id"]
+                    evaluation_context.callback(log_id)
 
             self._upload_queue.task_done()

src/humanloop/otel/helpers.py

@@ -5,7 +5,6 @@
 from opentelemetry.trace import SpanKind
 from opentelemetry.util.types import AttributeValue
 
-from humanloop.otel.constants import HUMANLOOP_INTERCEPTED_PROMPT_CALL_SPAN_NAME
 
 NestedDict = dict[str, Union["NestedDict", AttributeValue]]
 NestedList = list[Union["NestedList", NestedDict]]
@@ -178,12 +177,6 @@ def read_from_opentelemetry_span(span: ReadableSpan, key: str = "") -> NestedDic
             # Remove the key prefix and the first dot
             to_process.append((span_key, span_value))
 
-    if not to_process:
-        if key == "":
-            # Empty span attributes
-            return result
-        raise KeyError(f"Key {key} not found in span attributes")
-
     for span_key, span_value in to_process:  # type: ignore
         parts = span_key.split(".")
         len_parts = len(parts)
@@ -264,10 +257,6 @@ def is_llm_provider_call(span: ReadableSpan) -> bool:
     )
 
 
-def is_intercepted_call(span: ReadableSpan) -> bool:
-    return span.name == HUMANLOOP_INTERCEPTED_PROMPT_CALL_SPAN_NAME
-
-
 def is_humanloop_span(span: ReadableSpan) -> bool:
     """Check if the Span was created by the Humanloop SDK."""
     return span.name.startswith("humanloop.")
@@ -285,7 +274,7 @@ def module_is_installed(module_name: str) -> bool:
     return True
 
 
-def jsonify_if_not_string(func: Callable, output: Any) -> str:
+def process_output(func: Callable, output: Any) -> str:
     if not isinstance(output, str):
         try:
             output = json.dumps(output)

src/humanloop/otel/processor.py

@@ -5,8 +5,11 @@
 from opentelemetry.sdk.trace import ReadableSpan, Span
 from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExporter
 
-from humanloop.context import get_prompt_path
+from humanloop.context import get_prompt_context, get_trace_id
 from humanloop.otel.constants import (
+    HUMANLOOP_FILE_KEY,
+    HUMANLOOP_FILE_TYPE_KEY,
+    HUMANLOOP_LOG_KEY,
     HUMANLOOP_PATH_KEY,
 )
 from humanloop.otel.helpers import is_llm_provider_call
@@ -39,10 +42,21 @@ class HumanloopSpanProcessor(SimpleSpanProcessor):
     def __init__(self, exporter: SpanExporter) -> None:
         super().__init__(exporter)
 
-    def on_start(self, span: Span):
+    def on_start(self, span: Span, parent_context):
         if is_llm_provider_call(span):
-            prompt_path = get_prompt_path()
-            if prompt_path:
-                span.set_attribute(HUMANLOOP_PATH_KEY, prompt_path)
+            context = get_prompt_context()
+            prompt_path, prompt_template = context.path, context.template
+            if context:
+                span.set_attribute(HUMANLOOP_PATH_KEY, context.path)
+                span.set_attribute(HUMANLOOP_FILE_TYPE_KEY, "prompt")
+                if prompt_template:
+                    span.set_attribute(
+                        f"{HUMANLOOP_FILE_KEY}.template",
+                        prompt_template,
+                    )
             else:
-                raise ValueError("Provider call outside @prompt context manager")
+                raise ValueError(f"Provider call outside @prompt context manager: {prompt_path}")
+            trace_id = get_trace_id()
+            if trace_id:
+                span.set_attribute(f"{HUMANLOOP_LOG_KEY}.trace_parent_id", trace_id)
+                print(span)

src/humanloop/overload.py

@@ -17,7 +17,7 @@
 logger = logging.getLogger("humanloop.sdk")
 
 
-CLIENT_TYPE = TypeVar("CLIENT_TYPE", PromptsClient, FlowsClient)
+CLIENT_TYPE = TypeVar("CLIENT_TYPE")
 
 
 def overload_log(client: CLIENT_TYPE) -> CLIENT_TYPE:
@@ -51,8 +51,7 @@ def _overload_log(
                 "trace_parent_id": trace_id,
             }
         try:
-            response = self._call(**kwargs)
-            response = typing.cast(PromptCallResponse, response)
+            response = self._log(**kwargs)
         except Exception as e:
             # TODO handle
             # TODO: Bug found in backend: not specifying a model 400s but creates a File
@@ -67,7 +66,7 @@ def _overload_log(
     return client
 
 
-def overload_prompt_call(client: PromptsClient) -> PromptsClient:
+def overload_call(client: PromptsClient) -> PromptsClient:
     client._call = client.call
 
     def _overload_call(self, **kwargs) -> PromptCallResponse: