Method parse_embedding_response

api/bedrock_client.py:276–297

Parse Bedrock embedding response to EmbedderOutput format.

(self, response: Any)

            return str(response)

    def parse_embedding_response(self, response: Any) -> EmbedderOutput:
        """Parse Bedrock embedding response to EmbedderOutput format."""
        from adalflow.core.types import Embedding

        try:
            embedding_data: List[Embedding] = []

            if isinstance(response, dict) and "embeddings" in response:
                embeddings = response.get("embeddings") or []
                embedding_data = [
                    Embedding(embedding=emb, index=i) for i, emb in enumerate(embeddings)
                ]
            elif isinstance(response, dict) and "embedding" in response:
                emb = response.get("embedding") or []
                embedding_data = [Embedding(embedding=emb, index=0)]
            else:
                raise ValueError(f"Unexpected embedding response type: {type(response)}")

            return EmbedderOutput(data=embedding_data, error=None, raw_response=response)
        except Exception as e:
            log.error(f"Error parsing Bedrock embedding response: {e}")
            return EmbedderOutput(data=[], error=str(e), raw_response=response)

    @backoff.on_exception(
        backoff.expo,