Repository: crewAIInc/crewAI
File: lib/crewai-tools/src/crewai_tools/tools/qdrant_vector_search_tool/qdrant_search_tool.py (Python)

Code before:
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        if QDRANT_AVAILABLE:
            self.client = QdrantClient(
                url=self.qdrant_url,
                api_key=self.qdrant_api_key if self.qdrant_api_key else None,
            )
        else:
            import click

            if click.confirm(
                "The 'qdrant-client' package is required to use the QdrantVectorSearchTool. "
                "Would you like to install it?"
            ):
                import subprocess

                subprocess.run(["uv", "add", "qdrant-client"], check=True)  # noqa: S607
            else:
                raise ImportError(
                    "The 'qdrant-client' package is required to use the QdrantVectorSearchTool. "
                    "Please install it with: uv add qdrant-client"
                )
        if self.filter_conditions:
            must_conditions = []
            for filter_condition in self.filter_conditions:
                if (
                    not isinstance(filter_condition, (tuple, list))
                    or len(filter_condition) < 2
                ):
                    raise ValueError(
                        f"Each filter condition must be a tuple or list with at least 2 elements (key, value). "
                        f"Got: {filter_condition}"
                    )
                must_conditions.append(
                    FieldCondition(
                        key=filter_condition[0],
                        match=MatchValue(value=filter_condition[1]),
                    )
                )
            self.search_filter = Filter(must=must_conditions)
        else:
            self.search_filter = None

    def _run(
        self,
        query: str,
        filter_by: str | None = None,
        filter_value: str | None = None,
    ) -> str:
        """Execute vector similarity search on Qdrant.


Review comment (refactor):
You can move the validation logic into `Field` on line 61, pydantic will also handle `not isinstance(filter_condition, (tuple, list))` for you automatically since you typed it as such. So the only thing you'll need to refactor is `len(filter_condition) < 2`, which is `max_length=2` per [pydantic](https://docs.pydantic.dev/latest/api/fields/#pydantic.fields.Field)

Code after:
            filter_(
                must=[
                    field_condition(key=k, match=match_value(value=v))
                    for k, v in conditions
                ]
            )
            if conditions
            else None
        )
        query_vector = (
            self.custom_embedding_fn(query)
            if self.custom_embedding_fn
            else (
                lambda: __import__("openai")
                .Client(api_key=os.getenv("OPENAI_API_KEY"))
                .embeddings.create(input=[query], model="text-embedding-3-large")
                .data[0]
                .embedding
            )()
        )
        results = self.client.query_points(
            collection_name=self.qdrant_config.collection_name,
            query=query_vector,
            query_filter=search_filter,
            limit=self.qdrant_config.limit,
            score_threshold=self.qdrant_config.score_threshold,
        )

        return json.dumps(
            [
                {
                    "distance": p.score,
                    "metadata": p.payload.get("metadata", {}) if p.payload else {},
                    "context": p.payload.get("text", "") if p.payload else {},
                }
                for p in results.points
            ],
            indent=2,
        )