input_body["textGenerationConfig"] = {**model_kwargs}
        else:
            input_body["inputText"] = prompt
        if provider == "anthropic" and "max_tokens_to_sample" not in input_body:
            input_body["max_tokens_to_sample"] = 256
        return input_body
[docs]    @classmethod
    def prepare_output(cls, provider: str, response: Any) -> str:
        if provider == "anthropic":
            response_body = json.loads(response.get("body").read().decode())
            return response_body.get("completion")
        else:
            response_body = json.loads(response.get("body").read())
        if provider == "ai21":
            return response_body.get("completions")[0].get("data").get("text")
        else:
            return response_body.get("results")[0].get("outputText")
[docs]    @classmethod
    def prepare_output_stream(
        cls, provider: str, response: Any, stop: Optional[List[str]] = None
    ) -> Iterator[GenerationChunk]:
        stream = response.get("body")
        if not stream:
            return
        if provider not in cls.provider_to_output_key_map:
            raise ValueError(
                f"Unknown streaming response output key for provider: {provider}"
            )
        for event in stream:
            chunk = event.get("chunk")
            if chunk:
                chunk_obj = json.loads(chunk.get("bytes").decode())
                # chunk obj format varies with provider
                yield GenerationChunk(
                    text=chunk_obj[cls.provider_to_output_key_map[provider]]
                )
[docs]class BedrockBase(BaseModel, ABC):
    client: Any  #: :meta private:
    region_name: Optional[str] = None