def search_response_to_document()

in python/src/tablestore_for_agent_memory/util/tablestore_helper.py [0:0]

• 31 lines of code
• 8 McCabe index (conditional complexity)

    def search_response_to_document(search_response: tablestore.SearchResponse, text_field: str, embedding_field: str) -> (List[DocumentHit], Optional[str]):
        hits = []
        next_token = search_response.next_token
        if next_token:
            next_token = base64.b64encode(next_token).decode('utf-8')
        else:
            next_token = None
        for hit in search_response.search_hits:
            row = hit.row
            score = hit.score
            document_id = row[0][0][1]
            tenant_id = row[0][1][1]
            meta_data = {}
            text_content = None
            vector_content = None
            for col in row[1]:
                key = col[0]
                val = col[1]
                if key == text_field:
                    text_content = val
                    continue
                if key == embedding_field:
                    vector_content = json.loads(val)
                    continue
                meta_data[key] = val
            document = Document(document_id=document_id, tenant_id=tenant_id, text=text_content, embedding=vector_content, metadata=meta_data)
            doc_hit = DocumentHit(document=document)
            if score is not None and math.isnan(score) == False:
                doc_hit.score = score
            hits.append(doc_hit)
        return hits, next_token