(
self,
text,
source_type: SourceType,
knowledge_id: str,
document_id: str,
paragraph_id: str,
source_id: str,
is_active: bool,
embedding: Embeddings,
)
| 43 | return True |
| 44 | |
| 45 | def _save( |
| 46 | self, |
| 47 | text, |
| 48 | source_type: SourceType, |
| 49 | knowledge_id: str, |
| 50 | document_id: str, |
| 51 | paragraph_id: str, |
| 52 | source_id: str, |
| 53 | is_active: bool, |
| 54 | embedding: Embeddings, |
| 55 | ): |
| 56 | text = normalize_for_embedding(text) |
| 57 | text_embedding = [float(x) for x in embedding.embed_query(text)] |
| 58 | terms = list( |
| 59 | QuerySet(Termbase) |
| 60 | .filter( |
| 61 | knowledge_id=knowledge_id, |
| 62 | ) |
| 63 | .values_list("content", flat=True) |
| 64 | ) |
| 65 | embedding = Embedding( |
| 66 | id=uuid.uuid7(), |
| 67 | knowledge_id=knowledge_id, |
| 68 | document_id=document_id, |
| 69 | is_active=is_active, |
| 70 | paragraph_id=paragraph_id, |
| 71 | source_id=source_id, |
| 72 | embedding=text_embedding, |
| 73 | source_type=source_type, |
| 74 | search_vector=SearchVector(Value(to_ts_vector(text, user_words=terms)), config='simple'), |
| 75 | ) |
| 76 | embedding.save() |
| 77 | return True |
| 78 | |
| 79 | def _batch_save(self, text_list: List[Dict], embedding: Embeddings, is_the_task_interrupted): |
| 80 | texts = [normalize_for_embedding(row.get("text")) for row in text_list] |
nothing calls this directly
no test coverage detected