Retrieve similar tools using FAISS similarity search. Args: query: Query string to search for k: Number of results to return (default: 4) Returns: List of dictionaries containing tool information with similarity scores
(self, query: str, k: int = 4)
| 1037 | return contract_text |
| 1038 | |
| 1039 | async def retrieve(self, query: str, k: int = 4) -> List[Dict[str, Any]]: |
| 1040 | """Retrieve similar tools using FAISS similarity search. |
| 1041 | |
| 1042 | Args: |
| 1043 | query: Query string to search for |
| 1044 | k: Number of results to return (default: 4) |
| 1045 | |
| 1046 | Returns: |
| 1047 | List of dictionaries containing tool information with similarity scores |
| 1048 | """ |
| 1049 | if self._faiss_service is None: |
| 1050 | logger.warning("| ⚠️ FAISS service not initialized, cannot retrieve tools") |
| 1051 | return [] |
| 1052 | |
| 1053 | try: |
| 1054 | from src.environment.faiss.types import FaissSearchRequest |
| 1055 | |
| 1056 | request = FaissSearchRequest( |
| 1057 | query=query, |
| 1058 | k=k, |
| 1059 | fetch_k=k * 5 # Fetch more candidates before filtering |
| 1060 | ) |
| 1061 | |
| 1062 | result = await self._faiss_service.search_similar(request) |
| 1063 | |
| 1064 | if not result.success: |
| 1065 | logger.warning(f"| ⚠️ FAISS search failed: {result.message}") |
| 1066 | return [] |
| 1067 | |
| 1068 | # Extract documents and scores from result |
| 1069 | documents = [] |
| 1070 | if result.extra and "documents" in result.extra: |
| 1071 | docs = result.extra["documents"] |
| 1072 | scores = result.extra.get("scores", []) |
| 1073 | |
| 1074 | for doc, score in zip(docs, scores): |
| 1075 | # Extract tool name from metadata |
| 1076 | metadata = doc.get("metadata", {}) if isinstance(doc, dict) else {} |
| 1077 | tool_name = metadata.get("name", "") |
| 1078 | |
| 1079 | # Get tool config if available |
| 1080 | tool_config = None |
| 1081 | if tool_name and tool_name in self._tool_configs: |
| 1082 | tool_config = self._tool_configs[tool_name] |
| 1083 | |
| 1084 | documents.append({ |
| 1085 | "name": tool_name, |
| 1086 | "description": metadata.get("description", ""), |
| 1087 | "score": float(score), |
| 1088 | "content": doc.get("page_content", "") if isinstance(doc, dict) else str(doc), |
| 1089 | "config": tool_config.model_dump() if tool_config else None |
| 1090 | }) |
| 1091 | |
| 1092 | return documents |
| 1093 | |
| 1094 | except Exception as e: |
| 1095 | logger.error(f"| ❌ Error retrieving tools: {e}") |
| 1096 | return [] |
nothing calls this directly
no test coverage detected