def search_similar()

in backend-apis/app/routers/p7_return_agent.py [0:0]

• 33 lines of code
• 7 McCabe index (conditional complexity)

def search_similar(data: SearchSimilarRequest) -> SearchSimilarResponse:
    """
    Searches for the similar product options for the returning product
    based on image and category

    ### Representation of an Search Request
    **image**: *string*
    - Image URL - bucket/path/to/img

    **query**: *string*
    - Search Query

    Returns:
    ### Representation of an Search Response
    **results**: *list*
    - List of Similar Products

    """
    # Multimodal search
    try:
        image_path = data.image.split("/", 1)
        bucket = storage_client.bucket(image_path[0])
        blob = bucket.blob(f"{image_path[1]}")
        image_contents = blob.download_as_string()

        feature_vector = embeddings_client.get_embedding(
            text=data.query, image_bytes=image_contents
        )

        reduced_vector = utils_palm.reduce_embedding_dimension(
            vector_image=feature_vector.image_embedding,
            vector_text=feature_vector.text_embedding,
        )

        neighbors = utils_vertex_vector.find_neighbor(
            feature_vector=reduced_vector,
        )

        num_responses = len(neighbors.nearest_neighbors[0].neighbors)
        results = []
        for i, n in enumerate(neighbors.nearest_neighbors[0].neighbors):
            product = utils_cloud_sql.get_product(
                int(n.datapoint.datapoint_id)
            )
            snapshot = {}

            if product:
                snapshot = utils_cloud_sql.convert_product_to_dict(product)
            results.append(
                {"id": n.datapoint.datapoint_id, "snapshot": snapshot}
            )

            if i == 9 or (num_responses < 10 and i == num_responses - 1):
                break
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e)) from e

    return SearchSimilarResponse(results=results)