ronantakizawa/codereview-bench huggingface dataset:
Repository: alibaba/lowcode-engine
File: packages/renderer-core/src/renderer/component.tsx (TypeScript)

Code before:
import { IPublicTypeNodeSchema } from '@alilc/lowcode-types';
import adapter from '../adapter';
import baseRendererFactory from './base';
import { IBaseRendererProps, IBaseRenderComponent } from '../types';
import logger from '../utils/logger';
import './component.css';

export default function componentRendererFactory(): IBaseRenderComponent {
  const { PureComponent, createElement } = adapter.getRuntime();
  class FaultComponent extends PureComponent<IPublicTypeNodeSchema | any> {
    render() {
      logger.error(`%c组件 ${this.props.componentName} 渲染异常, 异常原因: ${this.props.error?.message || this.props.error || '未知'}`, 'color: #ff0000;');
      return createElement('div', {
        className: 'lc-component-placeholder',
      }, [
        createElement('img', {
          className: 'placeholder-icon',
          src: 'https://img.alicdn.com/imgextra/i3/O1CN014vaSuf29FSP05MRkB_!!6000000008038-55-tps-200-200.svg',
        }),
        createElement('span', {}, `${this.props.componentName || ''} 低代码组件占位`),
        createElement('span', { className: 'question-wrapper' }, [
          createElement('img', {
            className: 'question-icon',
            src: 'https://img.alicdn.com/imgextra/i3/O1CN01qvltzs1tbPXJTARss_!!6000000005920-55-tps-200-200.svg',
          }),
          createElement('span', {
            className: 'tooltip',
          }, '由于一些原因，导致低代码组件无法正常渲染，因此显示占位。通常这种情况不会影响组件配置可以正常使用，如果希望恢复至正常渲染，可以查看控制台相关报错'),
        ]),
      ]);
    }
  }
  const BaseRenderer = baseRendererFactory();
  return class CompRenderer extends BaseRenderer {

Review comment (suggestion):
这部分感觉用配置更好一些，目前有 faultComponent 的配置，可以考虑加一个 lowcodeFaultComponent 的配置，这样就不需要考虑是在 react-simulator 里面，还是 react-render 内。

![image](https://user-images.githubusercontent.com/11935995/228717758-dcb1f40c-9429-41d3-90d9-683d53686703.png)

对于低代码平台来说，只需要在插件中通过 AliLowcodeEgine.config.set('lowcodeFaultComponent', xxx) 对低代码组件的提示进行定制即可。


Code after:
import baseRendererFactory from './base';
import { IBaseRendererProps, IBaseRenderComponent } from '../types';

export default function componentRendererFactory(): IBaseRenderComponent {
  const BaseRenderer = baseRendererFactory();
  return class CompRenderer extends BaseRenderer {
    static displayName = 'CompRenderer';

    __namespace = 'component';

    __afterInit(props: IBaseRendererProps) {
      this.__generateCtx({
        component: this,
      });
      const schema = props.__schema || {};
      this.state = this.__parseData(schema.state || {});
      this.__initDataSource(props);
      this.__executeLifeCycleMethod('constructor', arguments as any);
    }

    render() {
      const { __schema, __components } = this.props;
      if (this.__checkSchema(__schema)) {
        return '自定义组件 schema 结构异常！';
      }
      this.__debug(`${CompRenderer.displayName} render - ${__schema.fileName}`);

      this.__generateCtx({
        component: this,
      });
      this.__render();

      const noContainer = this.__parseData(__schema.props?.noContainer);
Repository: alibaba/spring-ai-alibaba
File: spring-ai-alibaba-agent-framework/src/main/java/com/alibaba/cloud/ai/graph/agent/flow/node/EnhancedParallelResultAggregator.java (Java)

Code before:
							agentResult.get());
				}
				else {
					logger.warn("No output found for sub-agent: {} (outputKey: {})", subAgent.name(),
							subAgentOutputKey);
				}
			}
		}

		// Apply merge strategy if provided
		Object finalResult;
		if (mergeStrategy instanceof ParallelAgent.MergeStrategy strategy) {
			finalResult = strategy.merge(subAgentResults, state);
		}
		else {
			// Default behavior: return all results as a map
			finalResult = new HashMap<>(subAgentResults);
		}

		result.put(outputKey, finalResult);

		Map<String, KeyStrategy> releaseStrategyResults = subAgentResults.entrySet().stream()
				.collect(HashMap::new, 
						(map, entry) -> map.put(entry.getKey(), KeyStrategy.REPLACE),
						HashMap::putAll);

		state.updateStateWithKeyStrategies(subAgentResults, releaseStrategyResults);

		result.putAll(subAgentResults);

		logger.debug("Enhanced result aggregation completed. Final result stored under key: {}", outputKey);

		return result;
	}

}


Review comment (bug):
While fixing the CI issue in the `test` function, I discovered that `EnhancedParallelResultAggregator` was still executing `result.put(outputKey, finalResult)` even when `outputKey` was null, causing JSON serialization to fail.

I added a null check to ensure correct serialization.

Code after:
							agentResult.get());
				}
				else {
					logger.warn("No output found for sub-agent: {} (outputKey: {})", subAgent.name(),
							subAgentOutputKey);
				}
			}
		}

		// Apply merge strategy if provided
		Object finalResult;
		if (mergeStrategy instanceof ParallelAgent.MergeStrategy strategy) {
			finalResult = strategy.merge(subAgentResults, state);
		}
		else {
			// Default behavior: return all results as a map
			finalResult = new HashMap<>(subAgentResults);
		}

		// Only add the merged result if outputKey is not null
		if (outputKey != null && !outputKey.trim().isEmpty()) {
			result.put(outputKey, finalResult);
			logger.debug("Enhanced result aggregation completed. Final result stored under key: {}", outputKey);
		} else {
			logger.debug("Enhanced result aggregation completed. No outputKey specified, skipping merged result storage.");
		}

		Map<String, KeyStrategy> releaseStrategyResults = subAgentResults.entrySet().stream()
				.collect(HashMap::new, 
						(map, entry) -> map.put(entry.getKey(), KeyStrategy.REPLACE),
						HashMap::putAll);

		state.updateStateWithKeyStrategies(subAgentResults, releaseStrategyResults);

		result.putAll(subAgentResults);

		return result;
	}

}
Repository: allenai/OLMo
File: pretrain_data/c4ify/c4-cleaning.py (Python)

Code before:
    for data in all_data:
        page = convert_original_data_to_page(data)

        if not page_filter_by_language_and_score(page, 0.99):
            continue

        if not page_filter_by_valid_length(page):
            continue

        if not page_filter_by_paragraphs(page):
            continue

        converted_page = page_processing_by_lines(page)
        if converted_page is None:
            continue

        if not page_filter_by_badwords(converted_page, badwords):
            continue

        all_pages_to_save.append(page)

    with open(save_path, "w") as fp:
        for page in all_pages_to_save:
            fp.write(json.dumps(page) + "\n")


if __name__ == "__main__":
    fire.Fire(load_and_parse_file)


Review comment (suggestion):
Can u change this bit so that, instead of removing documents, it writes at `save_path` a jsonl with the output of each of filter. Reason is that we might want to use only a subset of filters. Something like :

```
{
  "id": "...",
  "page_filter_by_language_and_score", 0.xx,
  "page_filter_by_valid_length" true:false",
  ...
}

reason is that we might want to leave filtering to the mixing code.


Code after:
            badwords_found = badwords_regex[lang].search(text.lower())  # type: ignore
            if badwords_found is not None:
                if keep_badword_page(page):
                    return True
                return False
    return True


## Load from source data # noqa: E266


def convert_original_data_to_page(data: Dict[str, Any]) -> Page:
    return {
        "id": data["id"],
        "url": data["metadata"]["url"],
        "normalized_url": normalize_url(data["metadata"]["url"]),
        "text": data["text"],
        "timestamp": data["created"],
        "content_length": data["metadata"]["length"],
        "content_type": "",
        "language": data["metadata"]["language"],
        "language_score": data["metadata"]["language_score"],
    }  # type: ignore


def load_and_parse_file(
    file_path: str,
    save_path: str,
    max_lines: Optional[int] = None,
):
    all_data = []
    with open(file_path, "r") as fp:
        for line in fp.readlines(max_lines):  # type: ignore
            all_data.append(json.loads(line))

    badwords = load_badwords_regex()
    all_pages_to_save = []

    for data in all_data:
        page = convert_original_data_to_page(data)

        page = execute_filter(page, page_filter_by_language_and_score, min_probability=0.99)

        page = execute_filter(page, page_filter_by_valid_length)

        page = execute_filter(page, page_filter_by_paragraphs)

        page = page_processing_by_lines(page)  # type: ignore
        if page is None:
            continue
Repository: allenai/OLMo
File: olmo/config.py (Python)

Code before:
    """

    attention_layer_norm: bool = False
    """
    Apply layer norm to the keys and queries within the attention mechanism.
    This can help stabilize training.
    """

    residual_dropout: float = 0.1
    """
    The dropout probability for the MLP and attention output within each block.
    """

    embedding_dropout: float = 0.1
    """
    The dropout probability for embeddings.
    """

    layer_norm_type: LayerNormType = LayerNormType.default
    """
    The layernorm implementation to use.
    """

    layer_norm_with_affine: bool = True
    """
    Whether to include bias and weight parameters for the layer norms.
    """

    max_sequence_length: int = 1024
    """
    The maximum input sequence length supported by the model.
    """

    include_bias: bool = True
    """
    Whether or not to include bias parameters in linear layers.
    In PaLM, they got rid of all bias terms because they found that large
    models tend to have near 0 bias terms anyway.
    """

    scale_logits: bool = False
    """
    If ``True``, scale the output logits by ``1 / sqrt(d_model)``.
    """

    vocab_size: int = 50257
    """
    Vocabulary size of the model.
    """

    embedding_size: Optional[int] = 50304

Review comment (suggestion):
How about a clarification like this:
```suggestion
    Whether to include bias and weight parameters for the layer norms.
    This only affects layer norms that are immediately followed by a linear layer in the forward pass.
    Other layer norms, such as those applied to attention keys and queries, will always include an elementwise affine transform.
```

Code after:
    """

    attention_layer_norm: bool = False
    """
    Apply layer norm to the keys and queries within the attention mechanism.
    This can help stabilize training.
    """

    residual_dropout: float = 0.1
    """
    The dropout probability for the MLP and attention output within each block.
    """

    embedding_dropout: float = 0.1
    """
    The dropout probability for embeddings.
    """

    layer_norm_type: LayerNormType = LayerNormType.default
    """
    The layernorm implementation to use.
    """

    layer_norm_with_affine: bool = True
    """
    Whether to include bias and weight parameters for the layer norms.
    This only affects layer norms that are immediately followed by a linear layer in the forward pass.
    Other layer norms, such as those applied to attention keys and queries, will always include an elementwise affine transform.
    """

    max_sequence_length: int = 1024
    """
    The maximum input sequence length supported by the model.
    """

    include_bias: bool = True
    """
    Whether or not to include bias parameters in linear layers.
    In PaLM, they got rid of all bias terms because they found that large
    models tend to have near 0 bias terms anyway.
    """

    scale_logits: bool = False
    """
    If ``True``, scale the output logits by ``1 / sqrt(d_model)``.
    """

    vocab_size: int = 50257
    """
    Vocabulary size of the model.
    """
Repository: allenai/OLMo
File: olmo/torch_util.py (Python)

Code before:
import torch.distributed as dist

T = TypeVar("T")


def seed_all(seed: int):
    """Seed all rng objects."""
    import random

    import numpy as np

    if seed < 0 or seed > 2**32 - 1:
        raise ValueError(f"Seed {seed} is invalid. It must be on [0; 2^32 - 1]")
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    # torch.manual_seed may call manual_seed_all but calling it again here
    # to make sure it gets called at least once
    torch.cuda.manual_seed_all(seed)


def is_distributed() -> bool:
    if "LOCAL_RANK" in os.environ:
        return True
    else:
        return False


def get_node_rank() -> int:
    return int(os.environ.get("NODE_RANK") or (get_global_rank() - get_local_rank()) // get_local_world_size())


def get_world_size() -> int:
    if dist.is_available() and dist.is_initialized():
        return dist.get_world_size()
    else:
        return 1


def get_local_world_size() -> int:
    return int(os.environ.get("LOCAL_WORLD_SIZE") or 1)


def get_global_rank() -> int:
    return int(os.environ.get("RANK") or dist.get_rank())


def get_local_rank() -> int:
    return int(os.environ.get("LOCAL_RANK") or 0)



Review comment (suggestion):
Although this isn't necessarily in the scope of the PR, I just realized this could be implemented way better because I don't like the idea of relying on a non-standard env var.

```suggestion
    return dist.is_available() and dist.is_initialized()
```

Code after:
import torch.distributed as dist

T = TypeVar("T")


def seed_all(seed: int):
    """Seed all rng objects."""
    import random

    import numpy as np

    if seed < 0 or seed > 2**32 - 1:
        raise ValueError(f"Seed {seed} is invalid. It must be on [0; 2^32 - 1]")
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    # torch.manual_seed may call manual_seed_all but calling it again here
    # to make sure it gets called at least once
    torch.cuda.manual_seed_all(seed)


def is_distributed() -> bool:
    return dist.is_available() and dist.is_initialized()


def get_node_rank() -> int:
    return int(os.environ.get("NODE_RANK") or (get_global_rank() - get_local_rank()) // get_local_world_size())


def get_world_size() -> int:
    if is_distributed():
        return dist.get_world_size()
    else:
        return 1


def get_local_world_size() -> int:
    return int(os.environ.get("LOCAL_WORLD_SIZE") or 1)


def get_global_rank() -> int:
    return int(os.environ.get("RANK") or dist.get_rank())


def get_local_rank() -> int:
    return int(os.environ.get("LOCAL_RANK") or 0)


def get_fs_local_rank() -> int:
    """Get the local rank per filesystem, meaning that, regardless of the number of nodes,
    if all ranks share the same filesystem then `get_fs_local_rank()` will be equivalent to `get_global_rank()`,
Repository: allenai/OLMo
File: olmo/torch_util.py (Python)

Code before:


def peak_gpu_memory(reset: bool = False) -> Optional[float]:
    """
    Get the peak GPU memory usage in MB across all ranks.
    Only rank 0 will get the final result.
    """
    if not torch.cuda.is_available():
        return None

    device = torch.device("cuda")
    peak_mb = torch.cuda.max_memory_allocated(device) / 1000000
    if dist.is_available() and dist.is_initialized():
        peak_mb_tensor = torch.tensor(peak_mb, device=device)
        dist.reduce(peak_mb_tensor, 0, dist.ReduceOp.MAX)
        peak_mb = peak_mb_tensor.item()

    if reset:
        # Reset peak stats.
        torch.cuda.reset_max_memory_allocated(device)

    return peak_mb


def synchronize_flag(flag: bool, device: torch.device) -> bool:
    if dist.is_available() and dist.is_initialized():
        flag_tensor = torch.tensor(flag, device=device)
        dist.broadcast(flag_tensor, 0)
        return flag_tensor.item()  # type: ignore
    else:
        return flag


Review comment (suggestion):
Same thing here:
```suggestion
    if is_distributed():
```

Code after:
    """
    Get the peak GPU memory usage in MB across all ranks.
    Only rank 0 will get the final result.
    """
    if not torch.cuda.is_available():
        return None

    device = torch.device("cuda")
    peak_mb = torch.cuda.max_memory_allocated(device) / 1000000
    if is_distributed():
        peak_mb_tensor = torch.tensor(peak_mb, device=device)
        dist.reduce(peak_mb_tensor, 0, dist.ReduceOp.MAX)
        peak_mb = peak_mb_tensor.item()

    if reset:
        # Reset peak stats.
        torch.cuda.reset_max_memory_allocated(device)

    return peak_mb


def synchronize_flag(flag: bool, device: torch.device) -> bool:
    if is_distributed():
        flag_tensor = torch.tensor(flag, device=device)
        dist.broadcast(flag_tensor, 0)
        return flag_tensor.item()  # type: ignore
    else:
        return flag
Repository: allenai/OLMo
File: olmo/eval/downstream.py (Python)

Code before:
import abc
import logging
import re
from typing import Any, ClassVar, Dict, List, Optional, Sequence, Union

import datasets
import torch
import torch.nn.functional as F
from sklearn.metrics import f1_score
from torchmetrics import Metric

from olmo.torch_util import barrier, get_fs_local_rank
from olmo.util import load_dataset_from_disk, save_dataset_to_disk

from ..tokenizer import Tokenizer

log = logging.getLogger(__name__)


def load_dataset(path, name, split, datasets_cache_dir: Optional[str] = None):
    # Try get dataset from disk.
    if datasets_cache_dir is not None:
        try:
            return load_dataset_from_disk(path, name, split, datasets_cache_dir)
        except FileNotFoundError:
            log.info(
                "Path %s name %s split %s not present in local dir %s, loading from online",
                path,
                name,
                split,
                datasets_cache_dir,
            )
        # Barrier here to stop the case where FS rank 0 saves the dataset to disk before some non-zero
        # ranks try getting the dataset from disk. This would cause those non-zero ranks to bypass
        # the next barrier and cause rank 0 to be stuck at that barrier.
        barrier()

    dataset = None
    # Download and save the dataset, only once per FS. When a local datasets dir is not set,
    # this will hopefully cache the datasets for use in other FS ranks.
    if get_fs_local_rank() == 0:
        dataset = datasets.load_dataset(
            path=path,
            name=name,
            split=split,

Review comment (performance):
nit: missing type hints
```suggestion
def load_dataset(path, name, split, datasets_cache_dir: Optional[str] = None):
```

Code after:
import abc
import logging
import re
from typing import Any, ClassVar, Dict, List, Optional, Sequence, Union

import datasets
import torch
import torch.nn.functional as F
from sklearn.metrics import f1_score
from torchmetrics import Metric

from olmo.util import load_hf_dataset

from ..tokenizer import Tokenizer

log = logging.getLogger(__name__)


class ICLMetric(Metric):
    # update method does not require access to global metric state
    full_state_update: bool = False

    def __init__(self, metric_type="acc") -> None:
        """metric_type: f1, acc, len_norm, pmi_dc, ce_loss"""
        super().__init__(sync_on_compute=True)

        self.metric_type = metric_type

        self.add_state("loglikelihoods", default=[], dist_reduce_fx=None)
        self.add_state("labels", default=[], dist_reduce_fx=None)

    def reset(
        self,
    ):
        self.loglikelihoods = []
        self.labels = []

    def update(self, batch: Dict[str, Any], lm_logits: torch.Tensor, dc_lm_logits=None):
        lm_logits = F.log_softmax(lm_logits, dim=-1)

        if self.metric_type == "pmi_dc":
            assert dc_lm_logits is not None, "PMI_DC acc type selected but no domain conditional logits provided"

        for idx, (doc_id, cont_id) in enumerate(zip(batch["doc_id"], batch["cont_id"])):
            # [cont_len]: continuation is padded for batching
Repository: allenai/OLMo
File: olmo_data/data.py (Python)

Code before:
from contextlib import contextmanager
from pathlib import Path
from typing import Iterator

import importlib_resources
from importlib_resources.abc import Traversable


def _get_data_traversable(data_rel_path: str) -> Traversable:
    return importlib_resources.files("olmo_data").joinpath(data_rel_path)


def is_data_dir(data_rel_path: str) -> bool:
    return _get_data_traversable(data_rel_path).is_dir()


def is_data_file(data_rel_path: str) -> bool:
    return _get_data_traversable(data_rel_path).is_file()


@contextmanager
def get_data_path(data_rel_path: str) -> Iterator[Path]:
    try:
        with importlib_resources.as_file(_get_data_traversable(data_rel_path)) as path:
            yield path
    finally:
        pass


Review comment (suggestion):
I think the correct return type is:
```suggestion
def get_data_path(data_rel_path: str) -> Generator[Path, None, None]:
```

Code after:
from contextlib import contextmanager
from pathlib import Path
from typing import Generator

import importlib_resources
from importlib_resources.abc import Traversable


def _get_data_traversable(data_rel_path: str) -> Traversable:
    return importlib_resources.files("olmo_data").joinpath(data_rel_path)


def is_data_dir(data_rel_path: str) -> bool:
    return _get_data_traversable(data_rel_path).is_dir()


def is_data_file(data_rel_path: str) -> bool:
    return _get_data_traversable(data_rel_path).is_file()


@contextmanager
def get_data_path(data_rel_path: str) -> Generator[Path, None, None]:
    try:
        with importlib_resources.as_file(_get_data_traversable(data_rel_path)) as path:
            yield path
    finally:
        pass
Repository: allenai/OLMo
File: scripts/train.py (Python)

Code before:
        if not os.getenv("MASTER_ADDR"):
            os.environ["MASTER_ADDR"] = "0.0.0.0"
        if not os.getenv("MASTER_PORT"):
            os.environ["MASTER_PORT"] = "24501"
        dist.init_process_group(backend="gloo", timeout=timedelta(minutes=30))

    else:
        dist.init_process_group(backend="gloo", timeout=timedelta(minutes=30))

    log.info("Process group initialized")

    prepare_cli_environment()
    log.info("CLI environment prepared")

    add_cached_path_clients()

    try:
        yaml_path, args_list = sys.argv[1], sys.argv[2:]
    except IndexError:
        raise OLMoCliError(f"Usage: {sys.argv[0]} [CONFIG_PATH] [OPTIONS]")

    cfg = TrainConfig.load(yaml_path, [clean_opt(s) for s in args_list])
    if mps_device:
        log.info("Device is MPS. Updating config...")
        cfg.model.init_device = "mps"
        cfg.distributed_strategy = "single"  # type: ignore
    main(cfg)


Review comment (suggestion):
We also need these kinds of overrides for the case of the CPU. I think it will be easier if instead of having a flag like `mps_device`, you check what's in the torch default device.

Code after:
        if not os.getenv("MASTER_ADDR"):
            os.environ["MASTER_ADDR"] = "0.0.0.0"
        if not os.getenv("MASTER_PORT"):
            os.environ["MASTER_PORT"] = "24501"
        dist.init_process_group(backend="gloo", timeout=timedelta(minutes=30))

    else:
        dist.init_process_group(backend="gloo", timeout=timedelta(minutes=30))

    log.info("Process group initialized")

    prepare_cli_environment()
    log.info("CLI environment prepared")

    add_cached_path_clients()

    try:
        yaml_path, args_list = sys.argv[1], sys.argv[2:]
    except IndexError:
        raise OLMoCliError(f"Usage: {sys.argv[0]} [CONFIG_PATH] [OPTIONS]")

    cfg = TrainConfig.load(yaml_path, [clean_opt(s) for s in args_list])
    if torch.device("mps"):
        log.info("Device is MPS. Updating config...")
        cfg.model.init_device = "mps"
        cfg.distributed_strategy = "single"  # type: ignore

    if torch.device("cpu"):
        log.info("Device is CPU. Updating config...")
        cfg.model.init_device = "cpu"
        cfg.distributed_strategy = "single"  # type: ignore
    main(cfg)