import hashlib import math from apps.knowledge.models import KnowledgeChunk, KnowledgeDocument, KnowledgeItem DEFAULT_CHUNK_SIZE = 600 def split_text(content: str, chunk_size: int = DEFAULT_CHUNK_SIZE) -> list[str]: normalized = "\n".join(line.strip() for line in content.splitlines() if line.strip()) if not normalized: return [] return [ normalized[index : index + chunk_size] for index in range(0, len(normalized), chunk_size) ] def embed_text(content: str, dimensions: int = 16) -> list[float]: digest = hashlib.sha256(content.encode("utf-8")).digest() values = [digest[index] / 255 for index in range(dimensions)] length = math.sqrt(sum(value * value for value in values)) or 1 return [round(value / length, 6) for value in values] def rebuild_item_chunks(item: KnowledgeItem) -> list[KnowledgeChunk]: item.chunks.all().delete() content = "\n".join( [ item.title, item.problem, item.applicable_scope, item.error_code, item.root_cause, item.troubleshooting_steps, item.solution, item.risk_note, ] ) chunks = [] for index, chunk_text in enumerate(split_text(content)): chunks.append( KnowledgeChunk.objects.create( knowledge_base=item.knowledge_base, item=item, chunk_index=index, content=chunk_text, embedding=embed_text(chunk_text), metadata={ "source_type": "knowledge_item", "title": item.title, "item_id": item.id, }, ) ) return chunks def ingest_document_text( knowledge_base_id: int, file_name: str, file_type: str, content: str, operator=None, ) -> KnowledgeDocument: document = KnowledgeDocument.objects.create( knowledge_base_id=knowledge_base_id, file_name=file_name, file_type=file_type, content_text=content, parse_status="success", created_by=operator, ) for index, chunk_text in enumerate(split_text(content)): KnowledgeChunk.objects.create( knowledge_base=document.knowledge_base, document=document, chunk_index=index, content=chunk_text, embedding=embed_text(chunk_text), metadata={ "source_type": "document", "title": document.file_name, "document_id": document.id, }, ) return document