Clio support async functions

pyproject.toml

@@ -3,7 +3,7 @@ name = "humanloop"
 
 [tool.poetry]
 name = "humanloop"
-version = "0.8.28"
+version = "0.8.29"
 description = ""
 readme = "README.md"
 authors = []
@@ -39,17 +39,17 @@ deepdiff = "^8.2.0"
 httpx = ">=0.21.2"
 httpx-sse = "0.4.0"
 mmh3 = "^5.1.0"
-opentelemetry-api = "<=1.27.0"
+opentelemetry-api = ">=1.28.0"
 opentelemetry-instrumentation-anthropic = ">=0.20"
 opentelemetry-instrumentation-bedrock = ">=0.15"
 opentelemetry-instrumentation-cohere = ">=0.20"
 opentelemetry-instrumentation-groq = ">=0.29"
 opentelemetry-instrumentation-openai = ">=0.20"
 opentelemetry-instrumentation-replicate = ">=0.20"
-opentelemetry-proto = "^1.30.0"
-opentelemetry-sdk = "<=1.27.0"
+opentelemetry-proto = ">=1.30.0"
+opentelemetry-sdk = ">=1.28.0"
 parse = ">=1"
-protobuf = "^5.29.3"
+protobuf = ">=5.29.3"
 pydantic = ">= 1.9.2"
 pydantic-core = "^2.18.2"
 typing_extensions = ">= 4.0.0"

requirements.txt

@@ -2,17 +2,17 @@ deepdiff==^8.2.0
 httpx>=0.21.2
 httpx-sse==0.4.0
 mmh3==^5.1.0
-opentelemetry-api<=1.27.0
+opentelemetry-api>=1.28.0
 opentelemetry-instrumentation-anthropic>=0.20
 opentelemetry-instrumentation-bedrock>=0.15
 opentelemetry-instrumentation-cohere>=0.20
 opentelemetry-instrumentation-groq>=0.29
 opentelemetry-instrumentation-openai>=0.20
 opentelemetry-instrumentation-replicate>=0.20
-opentelemetry-proto==^1.30.0
-opentelemetry-sdk<=1.27.0
+opentelemetry-proto>=1.30.0
+opentelemetry-sdk>=1.28.0
 parse>=1
-protobuf==^5.29.3
+protobuf>=5.29.3
 pydantic>= 1.9.2
 pydantic-core==^2.18.2
 typing_extensions>= 4.0.0

src/humanloop/client.py

@@ -1,3 +1,4 @@
+from importlib import metadata
 import os
 import typing
 from typing import Any, List, Optional, Sequence
@@ -9,6 +10,7 @@
 
 from humanloop.core.client_wrapper import SyncClientWrapper
 
+from humanloop.error import HumanloopRuntimeError
 from humanloop.evals import run_eval
 from humanloop.evals.types import Dataset, Evaluator, EvaluatorCheck, File
 
@@ -68,6 +70,41 @@ def run(
         )
 
 
+class HumanloopTracerSingleton:
+    _instance = None
+
+    def __init__(self, hl_client_headers: dict[str, str], hl_client_base_url: str):
+        if HumanloopTracerSingleton._instance is not None:
+            raise HumanloopRuntimeError("Internal error: HumanloopTracerSingleton already initialized")
+
+        self.tracer_provider = TracerProvider(
+            resource=Resource(
+                attributes={
+                    "service.name": "humanloop-python-sdk",
+                    "service.version": metadata.version("humanloop"),
+                }
+            )
+        )
+        self.tracer_provider.add_span_processor(
+            HumanloopSpanProcessor(
+                exporter=HumanloopSpanExporter(
+                    hl_client_headers=hl_client_headers,
+                    hl_client_base_url=hl_client_base_url,
+                )
+            )
+        )
+
+        instrument_provider(provider=self.tracer_provider)
+
+        self.tracer = self.tracer_provider.get_tracer("humanloop.sdk")
+
+    @classmethod
+    def get_instance(cls, hl_client_headers: dict[str, str], hl_client_base_url: str):
+        if cls._instance is None:
+            cls._instance = cls(hl_client_headers, hl_client_base_url)
+        return cls._instance
+
+
 class Humanloop(BaseHumanloop):
     """
     See docstring of :class:`BaseHumanloop`.
@@ -117,27 +154,12 @@ def __init__(
         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
-        else:
-            self._tracer_provider = TracerProvider(
-                resource=Resource(
-                    attributes={
-                        "instrumentor": "humanloop.sdk",
-                    }
-                ),
-            )
-        instrument_provider(provider=self._tracer_provider)
-        self._tracer_provider.add_span_processor(
-            HumanloopSpanProcessor(exporter=HumanloopSpanExporter(client=self)),
+        # Initialize the tracer singleton
+        self._tracer_singleton = HumanloopTracerSingleton.get_instance(
+            hl_client_headers=self._client_wrapper.get_headers(),
+            hl_client_base_url=self._client_wrapper._base_url,
         )
-
-        if opentelemetry_tracer is None:
-            self._opentelemetry_tracer = self._tracer_provider.get_tracer(
-                "humanloop.sdk"
-            )
-        else:
-            self._opentelemetry_tracer = opentelemetry_tracer
+        self._opentelemetry_tracer = self._tracer_singleton.tracer
 
     def prompt(
         self,

src/humanloop/core/client_wrapper.py

@@ -16,7 +16,7 @@ def get_headers(self) -> typing.Dict[str, str]:
         headers: typing.Dict[str, str] = {
             "X-Fern-Language": "Python",
             "X-Fern-SDK-Name": "humanloop",
-            "X-Fern-SDK-Version": "0.8.28",
+            "X-Fern-SDK-Version": "0.8.29",
         }
         headers["X-API-KEY"] = self.api_key
         return headers

src/humanloop/decorators/flow.py

@@ -1,6 +1,7 @@
+import asyncio
 import logging
 from functools import wraps
-from typing import Any, Callable, Optional, TypeVar
+from typing import Any, Awaitable, Callable, Coroutine, Optional, TypeVar
 from typing_extensions import ParamSpec
 
 from opentelemetry.trace import Span, Tracer
@@ -121,13 +122,101 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> Optional[R]:
                         # Return the output of the decorated function
                         return func_output  # type: ignore [return-value]
 
+        @wraps(func)
+        async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> Awaitable[Optional[R]]:
+            span: Span
+            with set_decorator_context(
+                DecoratorContext(
+                    path=decorator_path,
+                    type="flow",
+                    version=flow_kernel,
+                )
+            ) as decorator_context:
+                with opentelemetry_tracer.start_as_current_span(HUMANLOOP_FLOW_SPAN_NAME) as span:  # type: ignore
+                    span.set_attribute(HUMANLOOP_FILE_PATH_KEY, decorator_path)
+                    span.set_attribute(HUMANLOOP_FILE_TYPE_KEY, file_type)
+                    trace_id = get_trace_id()
+                    func_args = bind_args(func, args, kwargs)
+
+                    # Create the trace ahead so we have a parent ID to reference
+                    init_log_inputs = {
+                        "inputs": {k: v for k, v in func_args.items() if k != "messages"},
+                        "messages": func_args.get("messages"),
+                        "trace_parent_id": trace_id,
+                    }
+                    this_flow_log: FlowLogResponse = client.flows._log(  # type: ignore [attr-defined]
+                        path=decorator_context.path,
+                        flow=decorator_context.version,
+                        log_status="incomplete",
+                        **init_log_inputs,
+                    )
+
+                    with set_trace_id(this_flow_log.id):
+                        func_output: Optional[R]
+                        log_output: Optional[str]
+                        log_error: Optional[str]
+                        log_output_message: Optional[ChatMessage]
+                        try:
+                            # Polymorphic decorator does not recognize the function is a coroutine
+                            func_output = await func(*args, **kwargs)  # type: ignore [misc]
+                            if (
+                                isinstance(func_output, dict)
+                                and len(func_output.keys()) == 2
+                                and "role" in func_output
+                                and "content" in func_output
+                            ):
+                                log_output_message = func_output  # type: ignore [assignment]
+                                log_output = None
+                            else:
+                                log_output = process_output(func=func, output=func_output)
+                                log_output_message = None
+                            log_error = None
+                        except HumanloopRuntimeError as e:
+                            # Critical error, re-raise
+                            client.logs.delete(id=this_flow_log.id)
+                            span.record_exception(e)
+                            raise e
+                        except Exception as e:
+                            logger.error(f"Error calling {func.__name__}: {e}")
+                            log_output = None
+                            log_output_message = None
+                            log_error = str(e)
+                            func_output = None
+
+                        updated_flow_log = {
+                            "log_status": "complete",
+                            "output": log_output,
+                            "error": log_error,
+                            "output_message": log_output_message,
+                            "id": this_flow_log.id,
+                        }
+                        # Write the Flow Log to the Span on HL_LOG_OT_KEY
+                        write_to_opentelemetry_span(
+                            span=span,  # type: ignore [arg-type]
+                            key=HUMANLOOP_LOG_KEY,
+                            value=updated_flow_log,  # type: ignore
+                        )
+                        # Return the output of the decorated function
+                        return func_output  # type: ignore [return-value]
+
+        # If the decorated function is an async function, return the async wrapper
+        if asyncio.iscoroutinefunction(func):
+            async_wrapper.file = File(  # type: ignore
+                path=decorator_path,
+                type=file_type,  # type: ignore [arg-type, typeddict-item]
+                version=FlowDict(**flow_kernel),  # type: ignore
+                callable=async_wrapper,
+            )
+            # Polymorphic decorator does not recognize the function is a coroutine
+            return async_wrapper  # type: ignore [return-value]
+
+        # If the decorated function is a sync function, return the sync wrapper
         wrapper.file = File(  # type: ignore
             path=decorator_path,
             type=file_type,  # type: ignore [arg-type, typeddict-item]
             version=FlowDict(**flow_kernel),  # type: ignore
             callable=wrapper,
         )
-
         return wrapper
 
     return decorator

src/humanloop/decorators/tool.py

@@ -1,3 +1,4 @@
+import asyncio
 import builtins
 import inspect
 import logging
@@ -7,7 +8,7 @@
 from dataclasses import dataclass
 from functools import wraps
 from inspect import Parameter
-from typing import Any, Callable, Literal, Mapping, Optional, Sequence, TypeVar, TypedDict, Union
+from typing import Any, Awaitable, Callable, Coroutine, Literal, Mapping, Optional, Sequence, TypeVar, TypedDict, Union
 from typing_extensions import ParamSpec
 
 from opentelemetry.trace import Tracer
@@ -112,13 +113,83 @@ def wrapper(*args: P.args, **kwargs: P.kwargs) -> Optional[R]:
                 # Return the output of the decorated function
                 return func_output
 
+        @wraps(func)
+        async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> Awaitable[Optional[R]]:
+            evaluation_context = get_evaluation_context()
+            if evaluation_context is not None:
+                if evaluation_context.path == path:
+                    raise HumanloopRuntimeError("Tools cannot be evaluated with the `evaluations.run()` utility.")
+            with opentelemetry_tracer.start_as_current_span("humanloop.tool") as span:
+                # Write the Tool Kernel to the Span on HL_FILE_OT_KEY
+                write_to_opentelemetry_span(
+                    span=span,  # type: ignore [arg-type]
+                    key=HUMANLOOP_FILE_KEY,
+                    value=tool_kernel,  # type: ignore [arg-type]
+                )
+                span.set_attribute(HUMANLOOP_FILE_PATH_KEY, path)
+                span.set_attribute(HUMANLOOP_FILE_TYPE_KEY, file_type)
+
+                log_inputs: dict[str, Any] = bind_args(func, args, kwargs)
+                log_error: Optional[str]
+                log_output: str
+
+                func_output: Optional[R]
+                try:
+                    # Polymorphic decorator does not recognize the function is a coroutine
+                    func_output = await func(*args, **kwargs)  # type: ignore [misc]
+                    log_output = process_output(
+                        func=func,
+                        output=func_output,
+                    )
+                    log_error = None
+                except HumanloopRuntimeError as e:
+                    # Critical error, re-raise
+                    raise e
+                except Exception as e:
+                    logger.error(f"Error calling {func.__name__}: {e}")
+                    output = None
+                    log_output = process_output(
+                        func=func,
+                        output=output,
+                    )
+                    log_error = str(e)
+
+                # Populate Tool Log attributes
+                tool_log = {
+                    "inputs": log_inputs,
+                    "output": log_output,
+                    "error": log_error,
+                    "trace_parent_id": get_trace_id(),
+                }
+                # Write the Tool Log to the Span on HL_LOG_OT_KEY
+                write_to_opentelemetry_span(
+                    span=span,  # type: ignore [arg-type]
+                    key=HUMANLOOP_LOG_KEY,
+                    value=tool_log,  # type: ignore [arg-type]
+                )
+
+                # Return the output of the decorated function
+                # Polymorphic decorator does not recognize the function is a coroutine
+                return func_output  # type: ignore [return-value]
+
+        # If the decorated function is an async function, return the async wrapper
+        if asyncio.iscoroutinefunction(func):
+            async_wrapper.file = File(  # type: ignore
+                path=path,
+                type=file_type,  # type: ignore [arg-type, typeddict-item]
+                version=tool_kernel,
+                callable=async_wrapper,
+            )
+            # Polymorphic decorator does not recognize the function is a coroutine
+            return async_wrapper  # type: ignore [return-value]
+
+        # If the decorated function is a sync function, return the sync wrapper
         wrapper.file = File(  # type: ignore
             path=path,
             type=file_type,  # type: ignore [arg-type, typeddict-item]
             version=tool_kernel,
             callable=wrapper,
         )
-
         return wrapper
 
     return decorator