def agg(
        self,
        func: Optional[Callable] = None,
        partition_by: Optional[PartitionByType] = None,
        params: Union[None, str, Sequence[str]] = None,
        output: OutputType = None,
        **signal_map,
    ) -> "Self":
        """Aggregate rows using `partition_by` statement and apply a function to the
        groups of aggregated rows. The function needs to return new objects for each
        group of the new rows. It returns a chain itself with new signals.

        Input-output relationship: N:M

        This method bears similarity to `gen()` and `map()`, employing a comparable set
        of parameters, yet differs in two crucial aspects:
        1. The `partition_by` parameter: This specifies the column name or a list of
           column names that determine the grouping criteria for aggregation.
        2. Group-based UDF function input: Instead of individual rows, the function
           receives a list all rows within each group defined by `partition_by`.

        Examples:
            ```py
            chain = chain.agg(
                total=lambda category, amount: [sum(amount)],
                output=float,
                partition_by="category",
            )
            chain.save("new_dataset")
            ```

            An alternative syntax, when you need to specify a more complex function:

            ```py
            # It automatically resolves which columns to pass to the function
            # by looking at the function signature.
            def agg_sum(
                file: list[File], amount: list[float]
            ) -> Iterator[tuple[File, float]]:
                yield file[0], sum(amount)

            chain = chain.agg(
                agg_sum,
                output={"file": File, "total": float},
                # Alternative syntax is to use `C` (short for Column) to specify
                # a column name or a nested column, e.g. C("file.path").
                partition_by=C("category"),
            )
            chain.save("new_dataset")
            ```
        """
        udf_obj = self._udf_to_obj(Aggregator, func, params, output, signal_map)
        return self._evolve(
            query=self._query.generate(
                udf_obj.to_udf_wrapper(),
                partition_by=partition_by,
                **self._settings.to_dict(),
            ),
            signal_schema=udf_obj.output,
        )
Aggregate rows using `partition_by` statement and apply a function to the
        groups of aggregated rows. The function needs to return new objects for each
        group of the new rows. It returns a chain itself with new signals.

        Input-output relationship: N:M

        This method bears similarity to `gen()` and `map()`, employing a comparable set
        of parameters, yet differs in two crucial aspects:
        1. The `partition_by` parameter: This specifies the column name or a list of
           column names that determine the grouping criteria for aggregation.
        2. Group-based UDF function input: Instead of individual rows, the function
           receives a list all rows within each group defined by `partition_by`.

        Examples:
            ```py
            chain = chain.agg(
                total=lambda category, amount: [sum(amount)],
                output=float,
                partition_by="category",
            )
            chain.save("new_dataset")
            ```

            An alternative syntax, when you need to specify a more complex function:

            ```py
            # It automatically resolves which columns to pass to the function
            # by looking at the function signature.
            def agg_sum(
                file: list[File], amount: list[float]
            ) -> Iterator[tuple[File, float]]:
                yield file[0], sum(amount)

            chain = chain.agg(
                agg_sum,
                output={"file": File, "total": float},
                # Alternative syntax is to use `C` (short for Column) to specify
                # a column name or a nested column, e.g. C("file.path").
                partition_by=C("category"),
            )
            chain.save("new_dataset")
            ```
        
src/datachain/lib/dc/datachain.py