adaptive_sdk.external

class RewardClient:

RewardClient(base_url, max_connections: int = 32, timeout: float | None = None)

async def score( self, req: Request) -> Response:

async def batch_score( self, requests: list[Request]) -> list[Response]:

async def validate_metadata(self, metadata: dict[typing.Any, typing.Any]):

async def batch_validate_metadata( self, list_of_metadata: list[dict[typing.Any, typing.Any]]) -> adaptive_sdk.external.reward_types.BatchedMetadataValidationResponse:

async def info(self) -> ServerInfo:

def blocking_info(self) -> ServerInfo:

def blocking_batch_score( self, requests: list[Request]) -> list[Response]:

class RewardServer(abc.ABC, typing.Generic[~META]):

Helper class that provides a standard way to create an ABC using inheritance.

requests_journal

request_timeout_s

metadata_cls

verbose

def cleanup(self):

@abstractmethod

async def score( self, request: adaptive_sdk.external.reward_types.ValidatedRequest[~META]) -> Response:

async def batch_score( self, requests: adaptive_sdk.external.reward_types.ValidatedBatchedRequest[~META]) -> BatchedResponse:

def get_medata_schema(self) -> starlette.responses.JSONResponse:

@abstractmethod

async def info(self) -> ServerInfo:

class Request(adaptive_sdk.external.reward_types.BaseModel):

turns: list[Turn]

metadata: dict[str, typing.Any] | None

model_config = {'extra': 'forbid'}