trulens.feedback.dummy.endpoint¶

trulens.feedback.dummy.endpoint ¶

Dummy API and Endpoint.

These are are meant to resemble (make similar sequences of calls) real APIs and Endpoints but not they do not actually make any network requests. Some randomness is introduced to simulate the behavior of real APIs.

Classes¶

NonDeterminism ¶

Bases: BaseModel

Hold random number generators and seeds for controlling non-deterministic behavior.

Attributes¶

seed `class-attribute` `instance-attribute` ¶

seed: int = 3735928559

Control randomness.

random `class-attribute` `instance-attribute` ¶

random: Any = Random(seed)

Random number generator.

np_random `class-attribute` `instance-attribute` ¶

np_random: Any = RandomState(seed)

Numpy Random number generator.

Functions¶

discrete_choice ¶

discrete_choice(
    seq: Sequence[A], probs: Sequence[float]
) -> A

Sample a random element from a sequence with the given probabilities.

DummyAPI ¶

Bases: BaseModel

A dummy model evaluation API used by DummyEndpoint.

This is meant to stand in for classes such as OpenAI.completion . Methods in this class are instrumented for cost tracking testing.

Attributes¶

loading_time_uniform_params `class-attribute` `instance-attribute` ¶

loading_time_uniform_params: Tuple[
    NonNegativeFloat, NonNegativeFloat
] = (0.7, 3.7)

How much time to indicate as needed to load the model.

Parameters of a uniform distribution.

loading_prob `class-attribute` `instance-attribute` ¶

loading_prob: NonNegativeFloat = 0.0

How often to produce the "model loading" response that huggingface api sometimes produces.

error_prob `class-attribute` `instance-attribute` ¶

error_prob: NonNegativeFloat = 0.0

How often to produce an error response.

freeze_prob `class-attribute` `instance-attribute` ¶

freeze_prob: NonNegativeFloat = 0.0

How often to freeze instead of producing a response.

overloaded_prob `class-attribute` `instance-attribute` ¶

overloaded_prob: NonNegativeFloat = 0.0

How often to produce the overloaded message that huggingface sometimes produces.

alloc `class-attribute` `instance-attribute` ¶

alloc: NonNegativeInt = 1024

How much data in bytes to allocate when making requests.

delay `class-attribute` `instance-attribute` ¶

delay: NonNegativeFloat = 0.0

How long to delay each request.

Delay is normally distributed with this mean and half this standard deviation, in seconds. Any delay sample below 0 is replaced with 0.

Functions¶

apost `async` ¶

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

Pretend to make an http post request to some model execution API.

post ¶

post(
    url: str,
    json: JSON,
    headers: Optional[Dict] = None,
    timeout: Optional[float] = None,
) -> Response

Pretend to make an http post request to some model execution API.

completion ¶

completion(
    *args, model: str, temperature: float = 0.0, prompt: str
) -> JSON

Fake text completion request.

acompletion `async` ¶

acompletion(
    *args, model: str, temperature: float = 0.0, prompt: str
) -> JSON

Fake text completion request.

classification ¶

classification(
    *args, model: str = "fakeclassier", text: str
) -> List[JSON]

Fake classification request.

aclassification `async` ¶

aclassification(
    *args, model: str = "fakeclassier", text: str
) -> List[JSON]

Fake classification request.

DummyAPICreator ¶

Creator of DummyAPI methods.

This is used for testing instrumentation of classes like boto3.ClientCreator.

Functions¶

create_method ¶

create_method(method_name: str) -> DummyAPI

Dynamically create a method that behaves like a DummyAPI method.

This method should be instrumented by DummyEndpoint for testing method creation like that of boto3.ClientCreator._create_api_method.

DummyEndpointCallback ¶

Bases: EndpointCallback

Callbacks for instrumented methods in DummyAPI to recover costs from those calls.

Attributes¶

endpoint `class-attribute` `instance-attribute` ¶

endpoint: Endpoint = Field(exclude=True)

The endpoint owning this callback.

cost `class-attribute` `instance-attribute` ¶

cost: Cost = Field(default_factory=Cost)

Costs tracked by this callback.

Functions¶

__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_embedding ¶

handle_embedding(response: Any) -> None

Called after each embedding response.

DummyEndpoint ¶

Bases: _WithPost, Endpoint

Endpoint for testing purposes.

Does not make any network calls and just pretends to.

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.

api `class-attribute` `instance-attribute` ¶

api: DummyAPI = Field(default_factory=DummyAPI)

Fake API to use for making fake requests.

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.

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.

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

wrap_function ¶

wrap_function(func)

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

post_json_first ¶

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

Wraps post with json()[0].

apost_json_first `async` ¶

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

Wraps apost with json()[0].

trulens.feedback.dummy.endpoint¶