Spaces:
Sleeping
Sleeping
File size: 4,692 Bytes
e2f7b9c e5b884f e2f7b9c e5b884f e2f7b9c e5b884f e2f7b9c e5b884f e2f7b9c 46bf894 e2f7b9c 46bf894 e2f7b9c 46bf894 e2f7b9c 46bf894 e2f7b9c 46bf894 e2f7b9c 46bf894 e2f7b9c 46bf894 e2f7b9c 46bf894 e2f7b9c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | from app.schemas.metadata_schema import InsuranceMetadata, CommonMetaData, HealthcareMetadata, HRMetadata, LegalMetadata,FinancialMetadata,ProcurementMetadata
from app.schemas.request_models import DocumentTypeSchema
import numpy as np
import os
import json
class MetadataService:
def __init__(self):
self.metadata_models = {
"Insurance": InsuranceMetadata,
"HR/Employment": HRMetadata,
"Legal/Compliance": LegalMetadata,
"Financial/Regulatory": FinancialMetadata,
"Government/Public Policy": CommonMetaData,
"Technical/IT Policies": CommonMetaData,
"Healthcare/Pharma": HealthcareMetadata,
"Procurement/Vendor Management": ProcurementMetadata
}
@staticmethod
def format_metadata_for_pinecone(metadata: dict) -> dict:
"""
Convert list fields in metadata to Pinecone's valid filter format using $in.
"""
formatted = {}
for key, value in metadata.items():
if isinstance(value, list):
if len(value) > 0:
formatted[key] = {"$in": value}
else:
formatted[key] = value
return formatted
def Return_document_model(self, doc_type_schema: DocumentTypeSchema):
"""
Returns appropriate metadata model based on document type
Args:
doc_type_schema: DocumentTypeSchema object containing document type
Returns:
Appropriate Pydantic model class for the document type
"""
doc_type = doc_type_schema.document_types
return self.metadata_models.get(doc_type, CommonMetaData)
@staticmethod
def normalize_dict_to_lists(metadata: dict) -> dict:
"""Convert dict values to lists if they aren't already"""
normalized = {}
for key, value in metadata.items():
if value is None:
normalized[key] = []
elif isinstance(value, list):
normalized[key] = value
else:
normalized[key] = [value]
return normalized
@staticmethod
def cosine_similarity(vec1, vec2) -> float:
return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
@staticmethod
def keyword_sementic_check(result, data, embedding_model):
for key in result.keys():
print(f"Comparing key: {key}", flush=True)
if result[key] is not None and data.get(key) is not None:
print(f"result[{key}]: {result[key]}", flush=True)
print(f"data[{key}]: {data[key]}", flush=True)
if isinstance(result[key], list) and isinstance(data[key], list):
# Filter to only strings
data_list = [v for v in data[key] if isinstance(v, str)]
val_list = [v for v in result[key] if isinstance(v, str)]
data_set = set(data_list)
if not data_list or not val_list:
print(f"Skipping key '{key}' due to empty valid strings.")
continue
# Precompute embeddings for data_list
data_embeddings = {val: embedding_model.embed_query(val) for val in data_list}
# Precompute embeddings for val_list
val_embeddings_list = embedding_model.embed_documents(val_list)
for idx, val in enumerate(val_list):
print(f"Comparing value: {val}", flush=True)
if val in data_set:
print(f"'{val}' found in data['{key}']", flush=True)
else:
print(f"'{val}' NOT found in data['{key}']", flush=True)
val_vector = val_embeddings_list[idx]
for data_val, data_vector in data_embeddings.items():
similarity = MetadataService.cosine_similarity(val_vector, data_vector)
print(f"Cosine similarity between '{val}' and '{data_val}': {similarity}", flush=True)
if similarity > 0.90:
print(f"'{val}' is similar to '{data_val}' with similarity {similarity}", flush=True)
result[key][idx] = data_val
break
else:
print(f"'{val}' is NOT similar to '{data_val}' with similarity {similarity}", flush=True)
return result |