Method create

backend/app/operators/retrieval/chunk.py:21–74

(
        self,
        create_dict: Dict,
        **kwargs,
    )

class ChunkModelOperator(PostgresModelOperator):
    async def create(
        self,
        create_dict: Dict,
        **kwargs,
    ) -> ModelEntity:
        from app.services.retrieval.embedding import embed_documents

        # handle kwargs
        self._check_kwargs(object_id_required=None, **kwargs)
        collection_id = kwargs["collection_id"]

        request = ChunkCreateRequest(**create_dict)
        content = request.content
        metadata = request.metadata

        # Get collection
        collection: Collection = await collection_ops.get(collection_id=collection_id)

        # Get model
        embedding_model = await model_ops.get(model_id=collection.embedding_model_id)

        # check if collection has available capacity
        if not collection.has_available_capacity(1):
            raise_http_error(
                ErrorCode.RESOURCE_LIMIT_REACHED,
                message=f"Collection {collection_id} has no available capacity to store new chunks.",
            )

        # embed the document
        embeddings = await embed_documents(
            documents=[content],
            embedding_model=embedding_model,
            embedding_size=collection.embedding_size,
        )
        embedding = embeddings[0]

        # create chunk
        num_tokens = default_tokenizer.count_tokens(content)
        new_chunk_id = Chunk.generate_random_id()
        await db_chunk.create_chunk(
            chunk_id=new_chunk_id,
            collection=collection,
            content=content,
            embedding=embedding,
            metadata=metadata,
            num_tokens=num_tokens,
        )

        # get the created chunk
        chunk = await self.get(collection_id=collection_id, chunk_id=new_chunk_id)

        # pop collection redis
        await collection_ops.redis.pop(collection)
        return chunk

    async def update(
        self,