trulens.providers.huggingface.endpoint¶

trulens.providers.huggingface.endpoint ¶

Classes¶

HuggingfaceEndpoint ¶

Bases: _WithPost, Endpoint

HuggingFace endpoint.

Instruments the requests.post method for requests to "https://api-inference.huggingface.co".

Attributes¶

tru_class_info `instance-attribute` ¶

tru_class_info: Class

Class information of this pydantic object for use in deserialization.

Using this odd key to not pollute attribute names in whatever class we mix this into. Should be the same as CLASS_INFO.

instrumented_methods `class-attribute` ¶

instrumented_methods: Dict[
    Any, List[Tuple[Callable, Callable, Type[Endpoint]]]
] = defaultdict(list)

Mapping of classes/module-methods that have been instrumented for cost tracking along with the wrapper methods and the class that instrumented them.

Key is the class or module owning the instrumented method. Tuple value has:

original function,
wrapped version,
endpoint that did the wrapping.

name `instance-attribute` ¶

name: str

API/endpoint name.

rpm `class-attribute` `instance-attribute` ¶

rpm: float = DEFAULT_RPM

Requests per minute.

retries `class-attribute` `instance-attribute` ¶

retries: int = 3

Retries (if performing requests using this class).

pace `class-attribute` `instance-attribute` ¶

pace: Pace = Field(
    default_factory=lambda: Pace(
        marks_per_second=DEFAULT_RPM / 60.0,
        seconds_per_period=60.0,
    ),
    exclude=True,
)

Pacing instance to maintain a desired rpm.

global_callback `class-attribute` `instance-attribute` ¶

global_callback: EndpointCallback = Field(exclude=True)

Track costs not run inside "track_cost" here.

Also note that Endpoints are singletons (one for each unique name argument) hence this global callback will track all requests for the named api even if you try to create multiple endpoints (with the same name).

callback_class `class-attribute` `instance-attribute` ¶

callback_class: Type[EndpointCallback] = Field(exclude=True)

Callback class to use for usage tracking.

callback_name `class-attribute` `instance-attribute` ¶

callback_name: str = Field(exclude=True)

Name of variable that stores the callback noted above.

Classes¶

EndpointSetup `dataclass` ¶

Class for storing supported endpoint information.

See track_all_costs for usage.

Functions¶

get_instances `classmethod` ¶

get_instances() -> Generator[InstanceRefMixin]

Get all instances of the class.

delete_instances `classmethod` ¶

delete_instances()

Delete all instances of the class.

__rich_repr__ ¶

__rich_repr__() -> Result

Requirement for pretty printing using the rich package.

load `staticmethod` ¶

load(obj, *args, **kwargs)

Deserialize/load this object using the class information in tru_class_info to lookup the actual class that will do the deserialization.

model_validate `classmethod` ¶

model_validate(*args, **kwargs) -> Any

Deserialized a jsonized version of the app into the instance of the class it was serialized from.

Note

This process uses extra information stored in the jsonized object and handled by WithClassInfo.

pace_me ¶

pace_me() -> float

Block until we can make a request to this endpoint to keep pace with maximum rpm. Returns time in seconds since last call to this method returned.

_can_retry ¶

_can_retry(e: Exception) -> bool

Determine whether a request that raised the given exception can be retried.

Things like authorization errors should not be retried.

run_in_pace ¶

run_in_pace(
    func: Callable[[A], B], *args, **kwargs
) -> B

Run the given func on the given args and kwargs at pace with the endpoint-specified rpm. Failures will be retried self.retries times.

run_me ¶

run_me(thunk: Thunk[T]) -> T

DEPRECATED: Run the given thunk, returning itse output, on pace with the api. Retries request multiple times if self.retries > 0.

DEPRECATED: Use run_in_pace instead.

print_instrumented `classmethod` ¶

print_instrumented()

Print out all of the methods that have been instrumented for cost tracking. This is organized by the classes/modules containing them.

_instrument_class_wrapper ¶

_instrument_class_wrapper(
    cls,
    wrapper_method_name: str,
    wrapped_method_filter: Callable[[Callable], bool],
) -> None

Instrument a method wrapper_method_name which produces a method so that the produced method gets instrumented. Only instruments the produced methods if they are matched by named wrapped_method_filter.

track_all_costs `staticmethod` ¶

track_all_costs(
    __func: CallableMaybeAwaitable[A, T],
    *args,
    with_openai: bool = True,
    with_hugs: bool = True,
    with_litellm: bool = True,
    with_bedrock: bool = True,
    with_cortex: bool = True,
    with_dummy: bool = True,
    **kwargs
) -> Tuple[T, Sequence[EndpointCallback]]

Track costs of all of the apis we can currently track, over the execution of thunk.

track_all_costs_tally `staticmethod` ¶

track_all_costs_tally(
    __func: CallableMaybeAwaitable[A, T],
    *args,
    with_openai: bool = True,
    with_hugs: bool = True,
    with_litellm: bool = True,
    with_bedrock: bool = True,
    with_cortex: bool = True,
    with_dummy: bool = True,
    **kwargs
) -> Tuple[T, Thunk[Cost]]

Track costs of all of the apis we can currently track, over the execution of thunk.

RETURNS	DESCRIPTION
`T`	Result of evaluating the thunk. TYPE: `T`
`Thunk[Cost]`	Thunk[Cost]: A thunk that returns the total cost of all callbacks that tracked costs. This is a thunk as the costs might change after this method returns in case of Awaitable results.

_track_costs `staticmethod` ¶

_track_costs(
    __func: CallableMaybeAwaitable[A, T],
    *args,
    with_endpoints: Optional[List[Endpoint]] = None,
    **kwargs
) -> Tuple[T, Sequence[EndpointCallback]]

Root of all cost tracking methods.

Runs the given thunk, tracking costs using each of the provided endpoints' callbacks.

track_cost ¶

track_cost(
    __func: CallableMaybeAwaitable[..., T], *args, **kwargs
) -> Tuple[T, EndpointCallback]

Tally only the usage performed within the execution of the given thunk.

Returns the thunk's result alongside the EndpointCallback object that includes the usage information.

_have_context `staticmethod` ¶

_have_context() -> bool

Determine whether we can access the context vars needed for cost tracking.

wrap_function ¶

wrap_function(func)

Create a wrapper of the given function to perform cost tracking.

apost `async` ¶

apost(
    url: str,
    json: JSON,
    timeout: Optional[float] = DEFAULT_NETWORK_TIMEOUT,
) -> Response

Make an http post request.

Subclasses can include additional logic to handle endpoint-specific responses.

post_json_first ¶

post_json_first(
    url: str,
    json: JSON,
    timeout: float = DEFAULT_NETWORK_TIMEOUT,
) -> Dict

Wraps post with json()[0].

apost_json_first `async` ¶

apost_json_first(
    url: str,
    json: JSON,
    timeout: float = DEFAULT_NETWORK_TIMEOUT,
) -> Dict

Wraps apost with json()[0].

post ¶

post(
    url: str,
    json: JSON,
    timeout: float = DEFAULT_NETWORK_TIMEOUT,
) -> Response

Make an http post request to the huggingface api.

This adds some additional logic beyond WithPost.post to handle huggingface-specific responses:

Model loading delay.
Overloaded API.
API error.

trulens.providers.huggingface.endpoint¶