endpoint - 🦑 TruLens

trulens.providers.cortex.endpoint

`trulens.providers.cortex.endpoint`

Classes

`CortexCallback`

Bases: EndpointCallback

Attributes

`endpoint`

endpoint: Endpoint = Field(exclude=True)

The endpoint owning this callback.

`cost`

cost: Cost = Field(default_factory=Cost)

Costs tracked by this callback.

Functions

`handle_generation`

handle_generation(response: dict) -> None

Get the usage information from Cortex LLM function response's usage field.

`repr`

__repr__() -> str

Safe repr that handles circular references.

`__rich_repr__`

__rich_repr__() -> Result

Requirement for pretty printing using the rich package.

`handle`

handle(response: Any) -> None

Called after each request.

`handle_chunk`

handle_chunk(response: Any) -> None

Called after receiving a chunk from a request.

`handle_generation_chunk`

handle_generation_chunk(response: Any) -> None

Called after receiving a chunk from a completion request.

`handle_classification`

handle_classification(response: Any) -> None

Called after each classification response.

`handle_embedding`

handle_embedding(response: Any) -> None

Called after each embedding response.

`CortexEndpoint`

Bases: Endpoint

Snowflake Cortex endpoint.

Attributes

`tru_class_info`

tru_class_info: Class

Class information of this pydantic object for use in deserialization.

`instrumented_methods`

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

Mapping of classes/module-methods that have been instrumented for cost tracking.

`name`

name: str

API/endpoint name.

`rpm`

rpm: float = DEFAULT_RPM

Requests per minute.

`retries`

retries: int = 3

Retries (if performing requests using this class).

`post_headers`

post_headers: Dict[str, str] = Field(
    default_factory=dict, exclude=True
)

Optional post headers for post requests if done by this class.

`pace`

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`

global_callback: EndpointCallback = Field(exclude=True)

Track costs not run inside "track_cost" here.

`callback_class`

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

Callback class to use for usage tracking.

`callback_name`

callback_name: str = Field(exclude=True)

Name of variable that stores the callback noted above.

Classes

`EndpointSetup`

Class for storing supported endpoint information.

Functions

`get_instances`

get_instances() -> Generator[InstanceRefMixin]

Get all instances of the class.

`delete_instances`

delete_instances()

Delete all instances of the class.

`__rich_repr__`

__rich_repr__() -> Result

Requirement for pretty printing using the rich package.

`load`

load(obj, *args, **kwargs)

Deserialize/load this object using the class information in tru_class_info.

`model_validate`

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

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

`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.

`run_in_pace`

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

Run the given func with the provided 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 its output, on pace with the api.

`print_instrumented`

print_instrumented()

Print out all of the methods that have been instrumented for cost tracking.

`track_all_costs`

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`

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.

`track_cost`

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

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

`wrap_function`

wrap_function(func)

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

Functions

`register_otel_cost_tracking`

register_otel_cost_tracking() -> None