Group chunks by document_id for document-level batch processing
(chunks: List[Dict[str, Any]])
| 187 | |
| 188 | # Utility functions for common batch operations |
| 189 | def batch_chunks_by_document(chunks: List[Dict[str, Any]]) -> Dict[str, List[Dict[str, Any]]]: |
| 190 | """Group chunks by document_id for document-level batch processing""" |
| 191 | document_batches = {} |
| 192 | for chunk in chunks: |
| 193 | doc_id = chunk.get('metadata', {}).get('document_id', 'unknown') |
| 194 | if doc_id not in document_batches: |
| 195 | document_batches[doc_id] = [] |
| 196 | document_batches[doc_id].append(chunk) |
| 197 | return document_batches |
| 198 | |
| 199 | def estimate_memory_usage(chunks: List[Dict[str, Any]]) -> float: |
| 200 | """Estimate memory usage of chunks in MB""" |
nothing calls this directly
no outgoing calls
no test coverage detected