diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS new file mode 100644 index 0000000..e22b7de --- /dev/null +++ b/.github/CODEOWNERS @@ -0,0 +1,3 @@ +.github/ @NVIDIA/nvflow-dev +pyproject.toml @NVIDIA/nvflow-dev +uv.lock @NVIDIA/nvflow-dev diff --git a/.github/ISSUE_TEMPLATE/bug.md b/.github/ISSUE_TEMPLATE/bug.md new file mode 100644 index 0000000..2ffc657 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/bug.md @@ -0,0 +1,38 @@ +--- +name: Bug +about: Bug report +title: '' +labels: '' +assignees: '' + +--- + +**Describe the bug** + +A clear and concise description of what the bug is. + +**Steps/Code to reproduce bug** + +Please list *minimal* steps or code snippet for us to be able to reproduce the bug. + +A helpful guide on how to craft a minimal bug report: http://matthewrocklin.com/blog/work/2018/02/28/minimal-bug-reports. + +**Expected behavior** + +A clear and concise description of what you expected to happen. + +**Configs** + +NVFlow workflow or stage configuration files (YAML). + +**Environment details** + +Please provide: +- OS version +- Python version +- `uv pip list` output + +**Additional context** + +Add any other context about the problem here. +Example: GPU model, cluster configuration diff --git a/.github/ISSUE_TEMPLATE/feature.md b/.github/ISSUE_TEMPLATE/feature.md new file mode 100644 index 0000000..58a6427 --- /dev/null +++ b/.github/ISSUE_TEMPLATE/feature.md @@ -0,0 +1,23 @@ +--- +name: Feature +about: Feature template +title: '' +labels: '' +assignees: '' + +--- + +**Use cases, pain points, and background** +Why should we do this? Why is this needed or wanted? + +**Description**: +What should we do? + +**Design**: +What files should be touched? What logic should be written? + +**Out of scope**: +What are some items that this issue could be mistaken to cover that this issue should explicitly NOT cover? + +**Acceptance Criteria**: +- [ ] Individual items that need to be finished in order for this issue to be considered completed diff --git a/.github/workflows/close-inactive-issue-pr.yml b/.github/workflows/close-inactive-issue-pr.yml new file mode 100644 index 0000000..7c10121 --- /dev/null +++ b/.github/workflows/close-inactive-issue-pr.yml @@ -0,0 +1,38 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +name: Stale-Close-Inactive-Issues-PRs +on: + schedule: + - cron: "30 1 * * *" + +jobs: + close-issues: + runs-on: ubuntu-latest + permissions: + issues: write + pull-requests: write + steps: + - uses: actions/stale@v6 + with: + operations-per-run: 100 + days-before-issue-stale: 30 + days-before-issue-close: 7 + stale-issue-label: "stale" + stale-issue-message: "This issue is stale because it has been open for 30 days with no activity. Remove stale label or comment or this will be closed in 7 days." + close-issue-message: "This issue was closed because it has been inactive for 7 days since being marked as stale." + days-before-pr-stale: 14 + days-before-pr-close: 7 + stale-pr-message: "This PR is stale because it has been open for 14 days with no activity. Remove stale label or comment or update or this will be closed in 7 days." + close-pr-message: "This PR was closed because it has been inactive for 7 days since being marked as stale." + repo-token: ${{ secrets.GITHUB_TOKEN }} diff --git a/.github/workflows/code-linting.yml b/.github/workflows/code-linting.yml new file mode 100644 index 0000000..f921c56 --- /dev/null +++ b/.github/workflows/code-linting.yml @@ -0,0 +1,31 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +name: Code linting + +on: + pull_request: + types: [opened, synchronize, reopened] + +jobs: + lint-check: + name: Lint check + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Check lint + run: | + pip install pre-commit==3.6.0 + pre-commit run --all-files --show-diff-on-failure --color=always diff --git a/.github/workflows/copyright-check.yml b/.github/workflows/copyright-check.yml new file mode 100644 index 0000000..3a2772e --- /dev/null +++ b/.github/workflows/copyright-check.yml @@ -0,0 +1,28 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +name: Copyright check + +on: + pull_request: + +jobs: + copyright-check: + name: Copyright check + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Check license headers + run: python scripts/check_license_headers.py diff --git a/.github/workflows/secrets-detector.yml b/.github/workflows/secrets-detector.yml new file mode 100644 index 0000000..dcb2ca9 --- /dev/null +++ b/.github/workflows/secrets-detector.yml @@ -0,0 +1,35 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +name: Secrets detector + +on: + pull_request: + +jobs: + secrets-detector: + name: Secrets detector + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Install trufflehog + run: | + curl -sSfL https://raw.githubusercontent.com/trufflesecurity/trufflehog/main/scripts/install.sh | sh -s -- -b /usr/local/bin + + - name: Scan for secrets + run: | + trufflehog git file://. --since-commit ${{ github.event.pull_request.base.sha }} --fail --no-update diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml new file mode 100644 index 0000000..23c3137 --- /dev/null +++ b/.github/workflows/unit-tests.yml @@ -0,0 +1,43 @@ +# Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +name: Unit tests + +on: + pull_request: + types: [opened, synchronize, reopened] + +jobs: + test: + name: Test + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Fail on draft PRs + if: github.event.pull_request.draft == true + run: | + echo "Failing on draft PR to enforce this check to run" + exit 1 + + - name: Setup for test + run: | + sudo apt-get update + sudo apt-get install -y --no-install-recommends git curl ca-certificates + curl -LsSf https://astral.sh/uv/install.sh | sh + uv sync --python 3.12 --no-install-project --extra dev + + - name: Test + run: | + uv run pytest tests/ -v --tb=short diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 2a8a51d..483ff73 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -6,6 +6,7 @@ repos: - id: end-of-file-fixer - id: check-yaml - id: check-added-large-files + exclude: '^uv\.lock$' - id: check-merge-conflict - id: detect-private-key @@ -30,10 +31,12 @@ repos: pass_filenames: true - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.1.9 + rev: v0.9.9 hooks: - id: ruff args: [--fix, --exit-non-zero-on-fix] + - id: ruff + args: [check, --select, I, --fix] - id: ruff-format - repo: https://github.com/pre-commit/mirrors-mypy diff --git a/CODE_OF_CONDUCT.md b/CODE_OF_CONDUCT.md new file mode 100644 index 0000000..ffe6fd5 --- /dev/null +++ b/CODE_OF_CONDUCT.md @@ -0,0 +1,133 @@ +# Code of Conduct + +## Our Pledge + +We as members, contributors, and leaders pledge to make participation in our +community a harassment-free experience for everyone, regardless of age, body +size, visible or invisible disability, ethnicity, sex characteristics, gender +identity and expression, level of experience, education, socio-economic status, +nationality, personal appearance, race, caste, color, religion, or sexual +identity and orientation. + +We pledge to act and interact in ways that contribute to an open, welcoming, +diverse, inclusive, and healthy community. + +## Our Standards + +Examples of behavior that contributes to a positive environment for our +community include: + +* Demonstrating empathy and kindness toward other people +* Being respectful of differing opinions, viewpoints, and experiences +* Giving and gracefully accepting constructive feedback +* Accepting responsibility and apologizing to those affected by our mistakes, + and learning from the experience +* Focusing on what is best not just for us as individuals, but for the overall + community + +Examples of unacceptable behavior include: + +* The use of sexualized language or imagery, and sexual attention or advances of + any kind +* Trolling, insulting or derogatory comments, and personal or political attacks +* Public or private harassment +* Publishing others' private information, such as a physical or email address, + without their explicit permission +* Other conduct which could reasonably be considered inappropriate in a + professional setting + +## Enforcement Responsibilities + +Community leaders are responsible for clarifying and enforcing our standards of +acceptable behavior and will take appropriate and fair corrective action in +response to any behavior that they deem inappropriate, threatening, offensive, +or harmful. + +Community leaders have the right and responsibility to remove, edit, or reject +comments, commits, code, wiki edits, issues, and other contributions that are +not aligned to this Code of Conduct, and will communicate reasons for moderation +decisions when appropriate. + +## Scope + +This Code of Conduct applies within all community spaces, and also applies when +an individual is officially representing the community in public spaces. +Examples of representing our community include using an official e-mail address, +posting via an official social media account, or acting as an appointed +representative at an online or offline event. + +## Enforcement + +Instances of abusive, harassing, or otherwise unacceptable behavior may be +reported through Github's report functionality. + +All complaints will be reviewed and investigated promptly and fairly by the +administrators of this repository. + +All community leaders are obligated to respect the privacy and security of the +reporter of any incident. + +## Enforcement Guidelines + +Community leaders will follow these Community Impact Guidelines in determining +the consequences for any action they deem in violation of this Code of Conduct: + +### 1. Correction + +**Community Impact**: Use of inappropriate language or other behavior deemed +unprofessional or unwelcome in the community. + +**Consequence**: A private, written warning from community leaders, providing +clarity around the nature of the violation and an explanation of why the +behavior was inappropriate. A public apology may be requested. + +### 2. Warning + +**Community Impact**: A violation through a single incident or series of +actions. + +**Consequence**: A warning with consequences for continued behavior. No +interaction with the people involved, including unsolicited interaction with +those enforcing the Code of Conduct, for a specified period of time. This +includes avoiding interactions in community spaces as well as external channels +like social media. Violating these terms may lead to a temporary or permanent +ban. + +### 3. Temporary Ban + +**Community Impact**: A serious violation of community standards, including +sustained inappropriate behavior. + +**Consequence**: A temporary ban from any sort of interaction or public +communication with the community for a specified period of time. No public or +private interaction with the people involved, including unsolicited interaction +with those enforcing the Code of Conduct, is allowed during this period. +Violating these terms may lead to a permanent ban. + +### 4. Permanent Ban + +**Community Impact**: Demonstrating a pattern of violation of community +standards, including sustained inappropriate behavior, harassment of an +individual, or aggression toward or disparagement of classes of individuals. + +**Consequence**: A permanent ban from any sort of public interaction within the +community. + +## Attribution + +This Code of Conduct is adapted from the [Contributor Covenant][homepage], +version 2.1, available at +[https://www.contributor-covenant.org/version/2/1/code_of_conduct.html][v2.1]. + +Community Impact Guidelines were inspired by +[Mozilla's code of conduct enforcement ladder][Mozilla CoC]. + +For answers to common questions about this code of conduct, see the FAQ at +[https://www.contributor-covenant.org/faq][FAQ]. Translations are available at +[https://www.contributor-covenant.org/translations][translations]. + +[homepage]: https://www.contributor-covenant.org +[v2.1]: https://www.contributor-covenant.org/version/2/1/code_of_conduct.html +[Mozilla CoC]: https://github.com/mozilla/diversity +[FAQ]: https://www.contributor-covenant.org/faq +[translations]: https://www.contributor-covenant.org/translations diff --git a/nvflow/core/console.py b/nvflow/core/console.py index 6604e29..de0f792 100644 --- a/nvflow/core/console.py +++ b/nvflow/core/console.py @@ -45,9 +45,9 @@ def header(message: str) -> None: Starting SDG Pipeline ================================================================================ """ - console.print(f"\n{'='*80}") + console.print(f"\n{'=' * 80}") console.print(f"[bold]{message}[/bold]") - console.print(f"{'='*80}\n") + console.print(f"{'=' * 80}\n") def section(message: str) -> None: @@ -64,9 +64,9 @@ def section(message: str) -> None: ▶ Running Stage: sdg.generate_qas ──────────────────────────────────────────────────────────────────────────────── """ - console.print(f"\n{'─'*80}") + console.print(f"\n{'─' * 80}") console.print(f"[cyan]▶[/cyan] {message}") - console.print(f"{'─'*80}") + console.print(f"{'─' * 80}") def success(message: str) -> None: diff --git a/nvflow/core/stage_registry.py b/nvflow/core/stage_registry.py index eae5525..c731ea4 100644 --- a/nvflow/core/stage_registry.py +++ b/nvflow/core/stage_registry.py @@ -161,7 +161,7 @@ def get(cls, recipe: str, workflow: str, stage: str) -> type[BaseStage]: else: available_recipes = list(cls._stages.keys()) raise KeyError( - f"Recipe '{recipe}' not found. " f"Available recipes: {available_recipes}" + f"Recipe '{recipe}' not found. Available recipes: {available_recipes}" ) from None @classmethod @@ -209,9 +209,7 @@ def list_workflows(cls, recipe: str) -> list[str]: KeyError: If recipe is not found """ if recipe not in cls._stages: - raise KeyError( - f"Recipe '{recipe}' not found. " f"Available recipes: {cls.list_recipes()}" - ) + raise KeyError(f"Recipe '{recipe}' not found. Available recipes: {cls.list_recipes()}") registered_workflows = set(cls._stages[recipe].keys()) @@ -243,9 +241,7 @@ def list_stages(cls, recipe: str, workflow: str) -> list[str]: KeyError: If recipe or workflow is not found """ if recipe not in cls._stages: - raise KeyError( - f"Recipe '{recipe}' not found. " f"Available recipes: {cls.list_recipes()}" - ) + raise KeyError(f"Recipe '{recipe}' not found. Available recipes: {cls.list_recipes()}") if workflow not in cls._stages[recipe]: raise KeyError( f"Workflow '{workflow}' not found in recipe '{recipe}'. " diff --git a/nvflow/core/workflow_runner.py b/nvflow/core/workflow_runner.py index d3469b4..6ee8076 100644 --- a/nvflow/core/workflow_runner.py +++ b/nvflow/core/workflow_runner.py @@ -86,7 +86,7 @@ def _load_config_with_inheritance(self, config_path: Path) -> OmegaConf: if not base_path.exists(): raise FileNotFoundError( - f"Base config not found: {config['_base_']} " f"(resolved to {base_path})" + f"Base config not found: {config['_base_']} (resolved to {base_path})" ) # Recursively load base config (supports chained inheritance) @@ -338,8 +338,7 @@ def _validate_stages(self, stages_to_run: list[str], all_stages: list[str]) -> N # Check if stage is in the workflow config if stage not in all_stages: raise ValueError( - f"Stage '{stage}' not found in workflow config. " - f"Available stages: {all_stages}" + f"Stage '{stage}' not found in workflow config. Available stages: {all_stages}" ) # Check if stage is registered in hierarchical registry diff --git a/nvflow/lib/rl/resume_filter.py b/nvflow/lib/rl/resume_filter.py index af40ee5..44425f1 100644 --- a/nvflow/lib/rl/resume_filter.py +++ b/nvflow/lib/rl/resume_filter.py @@ -35,6 +35,7 @@ 0 -- success (remaining_file written; may be empty if ALL_DONE) 1 -- error """ + import hashlib import json import os @@ -94,8 +95,7 @@ def resume_filter( f.write(json.dumps(r) + "\n") print( - f"RESUME_STATUS: remaining={len(remaining)} " - f"completed={len(completed)} total={len(inputs)}" + f"RESUME_STATUS: remaining={len(remaining)} completed={len(completed)} total={len(inputs)}" ) if not remaining: print("ALL_DONE") diff --git a/nvflow/lib/rl/verify_worker.py b/nvflow/lib/rl/verify_worker.py index 8c7b915..4cfb9dc 100644 --- a/nvflow/lib/rl/verify_worker.py +++ b/nvflow/lib/rl/verify_worker.py @@ -26,6 +26,7 @@ \\ """ + import asyncio import json import sys @@ -54,8 +55,7 @@ def _wait_for_server_client( last_err = e time.sleep(poll_interval) raise RuntimeError( - f"Could not connect to head server at {head_host}:{head_port} " - f"after {timeout}s: {last_err}" + f"Could not connect to head server at {head_host}:{head_port} after {timeout}s: {last_err}" ) diff --git a/nvflow/recipes/finance/datasets/finance_agent/tools.py b/nvflow/recipes/finance/datasets/finance_agent/tools.py index 857e981..2a4eddc 100644 --- a/nvflow/recipes/finance/datasets/finance_agent/tools.py +++ b/nvflow/recipes/finance/datasets/finance_agent/tools.py @@ -18,14 +18,13 @@ import re import traceback from abc import ABC, abstractmethod -from typing import Any +from typing import Any, override import aiohttp import backoff from bs4 import BeautifulSoup from model_library.base import LLM, ToolBody, ToolDefinition from tavily import AsyncTavilyClient -from typing_extensions import override from nvflow.utils import setup_logger @@ -97,8 +96,7 @@ async def call_tool( data_storage: dict[str, Any], llm: LLM, logger: logging.Logger | None = None, - ) -> dict[str, Any]: - ... + ) -> dict[str, Any]: ... async def __call__( self, diff --git a/nvflow/recipes/finance/stages/evaluation/evaluate.py b/nvflow/recipes/finance/stages/evaluation/evaluate.py index 500e946..d9fb4ac 100644 --- a/nvflow/recipes/finance/stages/evaluation/evaluate.py +++ b/nvflow/recipes/finance/stages/evaluation/evaluate.py @@ -557,7 +557,7 @@ def execute( ) for step in eval_steps: - console.info(f"Evaluating checkpoint step {step} " f"(format: {checkpoint_format})") + console.info(f"Evaluating checkpoint step {step} (format: {checkpoint_format})") if checkpoint_format == "hf": model_path = str(Path(checkpoint_path) / f"step_{step}" / "policy") diff --git a/nvflow/recipes/finance/stages/rl/prepare_data.py b/nvflow/recipes/finance/stages/rl/prepare_data.py index ff75d51..b1e6de9 100644 --- a/nvflow/recipes/finance/stages/rl/prepare_data.py +++ b/nvflow/recipes/finance/stages/rl/prepare_data.py @@ -79,7 +79,7 @@ def _write_overlay(self, output_dir: str, agents: list[dict[str, Any]]) -> str: overlay_path = host_dir / "agent_config_overlay.yaml" agent_names = [a["name"] for a in agents] - header = "# Auto-generated by PrepareDataForGRPOStage.\n" f"# Agents: {agent_names}\n" + header = f"# Auto-generated by PrepareDataForGRPOStage.\n# Agents: {agent_names}\n" overlay_path.write_text( header + yaml.dump(overlay, default_flow_style=False, sort_keys=False) ) diff --git a/nvflow/recipes/finance/stages/rl/training.py b/nvflow/recipes/finance/stages/rl/training.py index c3fb48f..44f4ce1 100644 --- a/nvflow/recipes/finance/stages/rl/training.py +++ b/nvflow/recipes/finance/stages/rl/training.py @@ -200,7 +200,7 @@ def _validate_parallelism_config( f"Parallelism validation failed for MoE model on MEGATRON backend:\n" f" world_size ({world_size}) must be divisible by TP×PP×CP = {tp}×{pp}×{cp} = {regular_model_size}\n" f" Regular layer DP would be: {world_size}/{regular_model_size} = " - f"{world_size/regular_model_size:.2f} (must be integer)" + f"{world_size / regular_model_size:.2f} (must be integer)" ) if world_size % expert_model_size != 0: @@ -208,7 +208,7 @@ def _validate_parallelism_config( f"Parallelism validation failed for MoE model on MEGATRON backend:\n" f" world_size ({world_size}) must be divisible by ETP×EP×PP = {etp}×{ep}×{pp} = {expert_model_size}\n" f" Expert layer DP would be: {world_size}/{expert_model_size} = " - f"{world_size/expert_model_size:.2f} (must be integer)" + f"{world_size / expert_model_size:.2f} (must be integer)" ) regular_dp = world_size // regular_model_size @@ -248,7 +248,7 @@ def _validate_parallelism_config( f"Parallelism validation failed for Dense model on {backend.upper()} backend:\n" f" world_size ({world_size}) must be divisible by {formula} = {parallelism_product}\n" f" Data parallel size would be: {world_size}/{parallelism_product} = " - f"{world_size/parallelism_product:.2f} (must be integer)" + f"{world_size / parallelism_product:.2f} (must be integer)" ) def _validate_sequence_packing_for_cp(self, nemo_rl_config: dict, backend: str) -> None: diff --git a/nvflow/recipes/finance/stages/sft/training.py b/nvflow/recipes/finance/stages/sft/training.py index e6927da..d3de70b 100644 --- a/nvflow/recipes/finance/stages/sft/training.py +++ b/nvflow/recipes/finance/stages/sft/training.py @@ -307,7 +307,7 @@ def _validate_parallelism_config( f"Parallelism validation failed for MoE model on MEGATRON backend:\n" f" world_size ({world_size}) must be divisible by TP×PP×CP = {tp}×{pp}×{cp} = {regular_model_size}\n" f" Regular layer DP would be: {world_size}/{regular_model_size} = " - f"{world_size/regular_model_size:.2f} (must be integer)" + f"{world_size / regular_model_size:.2f} (must be integer)" ) if world_size % expert_model_size != 0: @@ -315,7 +315,7 @@ def _validate_parallelism_config( f"Parallelism validation failed for MoE model on MEGATRON backend:\n" f" world_size ({world_size}) must be divisible by ETP×EP×PP = {etp}×{ep}×{pp} = {expert_model_size}\n" f" Expert layer DP would be: {world_size}/{expert_model_size} = " - f"{world_size/expert_model_size:.2f} (must be integer)" + f"{world_size / expert_model_size:.2f} (must be integer)" ) # Calculate actual DP sizes @@ -361,7 +361,7 @@ def _validate_parallelism_config( f"Parallelism validation failed for Dense model on {backend.upper()} backend:\n" f" world_size ({world_size}) must be divisible by {formula} = {parallelism_product}\n" f" Data parallel size would be: {world_size}/{parallelism_product} = " - f"{world_size/parallelism_product:.2f} (must be integer)" + f"{world_size / parallelism_product:.2f} (must be integer)" ) def _validate_sequence_packing_for_cp(self, nemo_rl_config: dict, backend: str) -> None: diff --git a/nvflow/recipes/finance/tests/test_prepare_question_gen_data.py b/nvflow/recipes/finance/tests/test_prepare_question_gen_data.py index 1daca2d..74d8516 100644 --- a/nvflow/recipes/finance/tests/test_prepare_question_gen_data.py +++ b/nvflow/recipes/finance/tests/test_prepare_question_gen_data.py @@ -183,9 +183,9 @@ def test_generate_combinations_multiple_years(self, tmp_path): output_rows.append(row) # Expected: 2 companies × 3 years = 6 entries - assert ( - len(output_rows) == 6 - ), f"Expected 6 rows (2 companies × 3 years), got {len(output_rows)}" + assert len(output_rows) == 6, ( + f"Expected 6 rows (2 companies × 3 years), got {len(output_rows)}" + ) # Check years years = sorted({row["year"] for row in output_rows}) diff --git a/nvflow/recipes/finance/utils/download/constants.py b/nvflow/recipes/finance/utils/download/constants.py index 7f49594..8178449 100644 --- a/nvflow/recipes/finance/utils/download/constants.py +++ b/nvflow/recipes/finance/utils/download/constants.py @@ -17,7 +17,6 @@ Section definitions, part mappings, and other constants used by the extractors. """ - # 10-K sections with their official numbering SECTIONS_10K: list[str] = [ "1", # Business diff --git a/nvflow/recipes/finance/utils/evaluation/checkpoint_converter.py b/nvflow/recipes/finance/utils/evaluation/checkpoint_converter.py index d556ef6..59a74ad 100644 --- a/nvflow/recipes/finance/utils/evaluation/checkpoint_converter.py +++ b/nvflow/recipes/finance/utils/evaluation/checkpoint_converter.py @@ -86,7 +86,7 @@ def convert_checkpoint( weights_path = megatron_path / "policy" / "weights" if not weights_path.exists(): raise FileNotFoundError( - f"Megatron checkpoint not found at {megatron_path}\n" f"Expected: {weights_path}/" + f"Megatron checkpoint not found at {megatron_path}\nExpected: {weights_path}/" ) # Convert Megatron → HuggingFace diff --git a/nvflow/recipes/finance/utils/rl/aggregate_seeds.py b/nvflow/recipes/finance/utils/rl/aggregate_seeds.py index 9b56c3f..528ec8e 100644 --- a/nvflow/recipes/finance/utils/rl/aggregate_seeds.py +++ b/nvflow/recipes/finance/utils/rl/aggregate_seeds.py @@ -29,6 +29,7 @@ /metrics.json -- machine-readable metrics /difficulty.jsonl -- per-question pass rates and pass@k """ + import json import math import sys @@ -182,14 +183,14 @@ def aggregate(rollout_dir: str, output_dir: str) -> None: f"Questions (uuid): {num_questions}", f"Total rows: {total_rows}", f"Avg pass rate: {avg_pass_rate:.1%}", - f"Mixed (06}" for k in k_values: - header += f" {'pass@'+str(k):>8}" + header += f" {'pass@' + str(k):>8}" lines.append("pass@k (macro average across questions):") lines.append(header) lines.append(" " + "-" * (28 + 10 * len(k_values))) @@ -199,7 +200,7 @@ def aggregate(rollout_dir: str, output_dir: str) -> None: for k in k_values: key = f"pass@{k}" if key in tm: - row += f" {tm[key]*100:>7.1f}%" + row += f" {tm[key] * 100:>7.1f}%" else: row += f" {'N/A':>8}" lines.append(row) @@ -207,7 +208,7 @@ def aggregate(rollout_dir: str, output_dir: str) -> None: for k in k_values: key = f"pass@{k}" if key in metrics: - overall += f" {metrics[key]*100:>7.1f}%" + overall += f" {metrics[key] * 100:>7.1f}%" lines.append(" " + "-" * (28 + 10 * len(k_values))) lines.append(overall) lines.append("") diff --git a/nvflow/recipes/finance/utils/rl/analyze_rollouts.py b/nvflow/recipes/finance/utils/rl/analyze_rollouts.py index 689db74..13c7dcb 100644 --- a/nvflow/recipes/finance/utils/rl/analyze_rollouts.py +++ b/nvflow/recipes/finance/utils/rl/analyze_rollouts.py @@ -30,6 +30,7 @@ Cross-seed difficulty analysis (pass@k, per-question pass rates) is handled separately by aggregate_seeds.py. """ + import json import sys from collections import Counter @@ -79,12 +80,12 @@ def analyze( title, "=" * 60, f"Total samples: {total}", - f"Correct (1.0): {len(correct):5d} ({len(correct)/total*100:5.1f}%)", - f"Incorrect (0.0): {len(incorrect):5d} ({len(incorrect)/total*100:5.1f}%)", - f"Partial (0 """ + import hashlib import json import sys diff --git a/nvflow/recipes/finance/utils/rl/filter_training_data.py b/nvflow/recipes/finance/utils/rl/filter_training_data.py index 62ffb62..43a492a 100644 --- a/nvflow/recipes/finance/utils/rl/filter_training_data.py +++ b/nvflow/recipes/finance/utils/rl/filter_training_data.py @@ -37,6 +37,7 @@ /validation.jsonl -- validation data (copied unchanged) /filter/filter_report.json -- filtering statistics """ + import argparse import json import shutil diff --git a/nvflow/recipes/finance/utils/sdg/apply_answer_filter.py b/nvflow/recipes/finance/utils/sdg/apply_answer_filter.py index 368abc5..cbd4359 100644 --- a/nvflow/recipes/finance/utils/sdg/apply_answer_filter.py +++ b/nvflow/recipes/finance/utils/sdg/apply_answer_filter.py @@ -42,9 +42,11 @@ def apply_answer_filter(input_file, output_file, keep_tag="ANSWERABLE"): buffer = [] - with open(input_file, "rb") as reader, open(output_file, "wb") as writer, open( - log_file, "w" - ) as log_writer: + with ( + open(input_file, "rb") as reader, + open(output_file, "wb") as writer, + open(log_file, "w") as log_writer, + ): for line in reader: line = line.strip() if not line: @@ -91,28 +93,28 @@ def apply_answer_filter(input_file, output_file, keep_tag="ANSWERABLE"): writer.write(b"\n".join(buffer) + b"\n") # Write summary to log - log_writer.write(f"\n{'='*60}\n") + log_writer.write(f"\n{'=' * 60}\n") log_writer.write("FILTER SUMMARY\n") - log_writer.write(f"{'='*60}\n") + log_writer.write(f"{'=' * 60}\n") log_writer.write(f"Total entries: {num_total_entries}\n") log_writer.write(f"Kept ({keep_tag}): {num_kept}\n") log_writer.write(f"Filtered out: {num_filtered}\n") log_writer.write(f"Missing tag: {num_missing_tag}\n") if num_total_entries > 0: - log_writer.write(f"Keep rate: {num_kept/num_total_entries*100:.2f}%\n") - log_writer.write(f"Filter rate: {num_filtered/num_total_entries*100:.2f}%\n") + log_writer.write(f"Keep rate: {num_kept / num_total_entries * 100:.2f}%\n") + log_writer.write(f"Filter rate: {num_filtered / num_total_entries * 100:.2f}%\n") # Log summary - logger.info(f"\n{'='*60}") + logger.info(f"\n{'=' * 60}") logger.info("FILTER SUMMARY") - logger.info(f"{'='*60}") + logger.info(f"{'=' * 60}") logger.info(f"Total entries: {num_total_entries}") logger.info(f"Kept ({keep_tag}): {num_kept}") logger.info(f"Filtered out: {num_filtered}") logger.info(f"Missing tag: {num_missing_tag}") if num_total_entries > 0: - logger.info(f"Keep rate: {num_kept/num_total_entries*100:.2f}%") - logger.info(f"Filter rate: {num_filtered/num_total_entries*100:.2f}%") + logger.info(f"Keep rate: {num_kept / num_total_entries * 100:.2f}%") + logger.info(f"Filter rate: {num_filtered / num_total_entries * 100:.2f}%") if __name__ == "__main__": diff --git a/nvflow/recipes/finance/utils/sdg/dg_sdg_data_preprocess.py b/nvflow/recipes/finance/utils/sdg/dg_sdg_data_preprocess.py index 8734670..1ad3f49 100644 --- a/nvflow/recipes/finance/utils/sdg/dg_sdg_data_preprocess.py +++ b/nvflow/recipes/finance/utils/sdg/dg_sdg_data_preprocess.py @@ -349,7 +349,7 @@ def write_chunk_outputs( f'' if base_href else "", head_assets or "", ] - orig_doc = f"""{''.join(head_content)}{orig_body}""" + orig_doc = f"""{"".join(head_content)}{orig_body}""" orig_path = output_dir / f"{base_name}_{chunk_index}_orig.html" with open(orig_path, "w", encoding="utf-8") as f: diff --git a/nvflow/recipes/finance/utils/sdg/dgsdg_post_process.py b/nvflow/recipes/finance/utils/sdg/dgsdg_post_process.py index 18f59f5..a732863 100644 --- a/nvflow/recipes/finance/utils/sdg/dgsdg_post_process.py +++ b/nvflow/recipes/finance/utils/sdg/dgsdg_post_process.py @@ -134,9 +134,10 @@ def dgsdg_post_process( logger.info("Pass 1: Cleaning records and writing full data...") - with open(input_file, encoding="utf-8") as f_in, open( - full_output_file, "w", encoding="utf-8" - ) as f_out: + with ( + open(input_file, encoding="utf-8") as f_in, + open(full_output_file, "w", encoding="utf-8") as f_out, + ): for line_num, line in enumerate(f_in, 1): line = line.strip() if not line: @@ -168,9 +169,10 @@ def dgsdg_post_process( medium_sft_count = 0 hard_rl_count = 0 - with open(medium_sft_file, "w", encoding="utf-8") as f_medium, open( - hard_rl_file, "w", encoding="utf-8" - ) as f_hard: + with ( + open(medium_sft_file, "w", encoding="utf-8") as f_medium, + open(hard_rl_file, "w", encoding="utf-8") as f_hard, + ): for record in all_records: # Check for final_result.jsonl if is_medium_sft_eligible(record): diff --git a/nvflow/recipes/finance/utils/sdg/difficulty_estimation.py b/nvflow/recipes/finance/utils/sdg/difficulty_estimation.py index 7de8d85..3413416 100644 --- a/nvflow/recipes/finance/utils/sdg/difficulty_estimation.py +++ b/nvflow/recipes/finance/utils/sdg/difficulty_estimation.py @@ -42,9 +42,10 @@ def prepare_difficulty_input(input_file: str, output_file: str): os.makedirs(os.path.dirname(output_file), exist_ok=True) count = 0 - with open(input_file, encoding="utf-8") as f_in, open( - output_file, "w", encoding="utf-8" - ) as f_out: + with ( + open(input_file, encoding="utf-8") as f_in, + open(output_file, "w", encoding="utf-8") as f_out, + ): for line in f_in: if not line.strip(): continue diff --git a/nvflow/recipes/finance/utils/sdg/parse_filter_responses.py b/nvflow/recipes/finance/utils/sdg/parse_filter_responses.py index a0f2f34..f1405a7 100644 --- a/nvflow/recipes/finance/utils/sdg/parse_filter_responses.py +++ b/nvflow/recipes/finance/utils/sdg/parse_filter_responses.py @@ -86,9 +86,11 @@ def parse_filter_responses(input_file, output_file): buffer = [] - with open(input_file, "rb") as reader, open(output_file, "wb") as writer, open( - log_file, "w" - ) as log_writer: + with ( + open(input_file, "rb") as reader, + open(output_file, "wb") as writer, + open(log_file, "w") as log_writer, + ): for line in reader: line = line.strip() if not line: @@ -146,32 +148,36 @@ def parse_filter_responses(input_file, output_file): writer.write(b"\n".join(buffer) + b"\n") # Write summary to log - log_writer.write(f"\n{'='*60}\n") + log_writer.write(f"\n{'=' * 60}\n") log_writer.write("PARSE SUMMARY\n") - log_writer.write(f"{'='*60}\n") + log_writer.write(f"{'=' * 60}\n") log_writer.write(f"Total entries: {num_total_entries}\n") log_writer.write(f"Successfully parsed: {num_successfully_parsed}\n") log_writer.write(f"Failed to parse: {num_failed_to_parse}\n") - log_writer.write(f"Success rate: {num_successfully_parsed/num_total_entries*100:.2f}%\n") + log_writer.write( + f"Success rate: {num_successfully_parsed / num_total_entries * 100:.2f}%\n" + ) log_writer.write(f"\nAnswerable: {num_answerable}\n") log_writer.write(f"Unanswerable: {num_unanswerable}\n") if num_successfully_parsed > 0: log_writer.write( - f"Unanswerable rate: {num_unanswerable/num_successfully_parsed*100:.2f}%\n" + f"Unanswerable rate: {num_unanswerable / num_successfully_parsed * 100:.2f}%\n" ) # Log summary - logger.info(f"\n{'='*60}") + logger.info(f"\n{'=' * 60}") logger.info("PARSE SUMMARY") - logger.info(f"{'='*60}") + logger.info(f"{'=' * 60}") logger.info(f"Total entries: {num_total_entries}") logger.info(f"Successfully parsed: {num_successfully_parsed}") logger.info(f"Failed to parse: {num_failed_to_parse}") - logger.info(f"Success rate: {num_successfully_parsed/num_total_entries*100:.2f}%") + logger.info(f"Success rate: {num_successfully_parsed / num_total_entries * 100:.2f}%") logger.info(f"\nAnswerable: {num_answerable}") logger.info(f"Unanswerable: {num_unanswerable}") if num_successfully_parsed > 0: - logger.info(f"Unanswerable rate: {num_unanswerable/num_successfully_parsed*100:.2f}%") + logger.info( + f"Unanswerable rate: {num_unanswerable / num_successfully_parsed * 100:.2f}%" + ) if __name__ == "__main__": diff --git a/nvflow/recipes/finance/utils/sdg/parse_generated_questions.py b/nvflow/recipes/finance/utils/sdg/parse_generated_questions.py index 3f8c876..91dd045 100644 --- a/nvflow/recipes/finance/utils/sdg/parse_generated_questions.py +++ b/nvflow/recipes/finance/utils/sdg/parse_generated_questions.py @@ -80,9 +80,11 @@ def parse_generations(input_file, output_file): log_file = output_file.replace(".jsonl", "_log.txt") num_original_questions, num_failed_to_parse = 0, 0 num_generated_questions, num_generated_questions_after_deduplication = 0, 0 - with jsonlines.open(input_file) as reader, jsonlines.open(output_file, "w") as writer, open( - log_file, "w" - ) as log_writer: + with ( + jsonlines.open(input_file) as reader, + jsonlines.open(output_file, "w") as writer, + open(log_file, "w") as log_writer, + ): deduplicated_questions = set() for row in reader: num_original_questions += 1 diff --git a/nvflow/recipes/finance/utils/sdg/parse_judge_responses.py b/nvflow/recipes/finance/utils/sdg/parse_judge_responses.py index 87b2517..e5fd11e 100644 --- a/nvflow/recipes/finance/utils/sdg/parse_judge_responses.py +++ b/nvflow/recipes/finance/utils/sdg/parse_judge_responses.py @@ -78,9 +78,11 @@ def parse_judge_responses(input_file, output_file): num_failed_to_parse = 0 num_successfully_parsed = 0 - with open(input_file) as reader, open(output_file, "w") as writer, open( - log_file, "w" - ) as log_writer: + with ( + open(input_file) as reader, + open(output_file, "w") as writer, + open(log_file, "w") as log_writer, + ): for line in reader: row = json.loads(line) num_total_entries += 1 @@ -112,13 +114,15 @@ def parse_judge_responses(input_file, output_file): log_writer.write(msg + "\n") # Write summary to log - log_writer.write(f"\n{'='*60}\n") + log_writer.write(f"\n{'=' * 60}\n") log_writer.write("SUMMARY\n") - log_writer.write(f"{'='*60}\n") + log_writer.write(f"{'=' * 60}\n") log_writer.write(f"Total entries: {num_total_entries}\n") log_writer.write(f"Successfully parsed: {num_successfully_parsed}\n") log_writer.write(f"Failed to parse: {num_failed_to_parse}\n") - log_writer.write(f"Success rate: {num_successfully_parsed/num_total_entries*100:.2f}%\n") + log_writer.write( + f"Success rate: {num_successfully_parsed / num_total_entries * 100:.2f}%\n" + ) # Print summary to console as well logger.info("=" * 60) diff --git a/nvflow/recipes/finance/utils/sdg/postprocess_genselect.py b/nvflow/recipes/finance/utils/sdg/postprocess_genselect.py index 5fb6579..c07d5ba 100644 --- a/nvflow/recipes/finance/utils/sdg/postprocess_genselect.py +++ b/nvflow/recipes/finance/utils/sdg/postprocess_genselect.py @@ -66,9 +66,10 @@ def postprocess_genselect(input_file, output_file): records_processed = 0 records_skipped = 0 - with open(input_file, encoding="utf-8") as f_in, open( - output_file, "w", encoding="utf-8" - ) as f_out: + with ( + open(input_file, encoding="utf-8") as f_in, + open(output_file, "w", encoding="utf-8") as f_out, + ): for line_num, line in enumerate(f_in, 1): try: record = orjson.loads(line.strip()) diff --git a/nvflow/recipes/finance/utils/sft/add_token_counts.py b/nvflow/recipes/finance/utils/sft/add_token_counts.py index 7926678..dddfec0 100644 --- a/nvflow/recipes/finance/utils/sft/add_token_counts.py +++ b/nvflow/recipes/finance/utils/sft/add_token_counts.py @@ -101,9 +101,10 @@ def add_token_counts( try: with Pool(num_workers, initializer=_init_worker, initargs=(tokenizer_path,)) as pool: - with open(input_path, encoding="utf-8") as f_in, open( - actual_output, "w", encoding="utf-8" - ) as f_out: + with ( + open(input_path, encoding="utf-8") as f_in, + open(actual_output, "w", encoding="utf-8") as f_out, + ): batch: list[str] = [] processed = 0 diff --git a/nvflow/recipes/finance/utils/sft/messages_converter.py b/nvflow/recipes/finance/utils/sft/messages_converter.py index 9c5633e..1a1bf0f 100644 --- a/nvflow/recipes/finance/utils/sft/messages_converter.py +++ b/nvflow/recipes/finance/utils/sft/messages_converter.py @@ -135,9 +135,10 @@ def convert_file( output_path.parent.mkdir(parents=True, exist_ok=True) - with open(input_path, encoding="utf-8") as f_in, open( - output_path, "w", encoding="utf-8" - ) as f_out: + with ( + open(input_path, encoding="utf-8") as f_in, + open(output_path, "w", encoding="utf-8") as f_out, + ): for line_num, line in enumerate(f_in, 1): try: record = json.loads(line.strip()) diff --git a/nvflow/recipes/finance/utils/shared/dataset_splitter.py b/nvflow/recipes/finance/utils/shared/dataset_splitter.py index a6cc256..a876392 100644 --- a/nvflow/recipes/finance/utils/shared/dataset_splitter.py +++ b/nvflow/recipes/finance/utils/shared/dataset_splitter.py @@ -108,9 +108,11 @@ def keep(rec: dict) -> bool: train_count = val_count = 0 output_fields = None if keep_all_fields else _SFT_OUTPUT_FIELDS - with open(input_file, encoding="utf-8") as f_in, open( - train_path, "w", encoding="utf-8" - ) as f_train, open(val_path, "w", encoding="utf-8") as f_val: + with ( + open(input_file, encoding="utf-8") as f_in, + open(train_path, "w", encoding="utf-8") as f_train, + open(val_path, "w", encoding="utf-8") as f_val, + ): for line in f_in: if not line.strip(): continue @@ -182,9 +184,9 @@ def main(): logger.info("SUMMARY") logger.info("=" * 60) logger.info(f"Total: {total:,}") - logger.info(f"Train: {train_n:,} ({train_n/total:.1%})") + logger.info(f"Train: {train_n:,} ({train_n / total:.1%})") if val_n: - logger.info(f"Val: {val_n:,} ({val_n/total:.1%})") + logger.info(f"Val: {val_n:,} ({val_n / total:.1%})") if filtered_n: logger.info(f"Filtered: {filtered_n:,}") logger.info(f"✅ Output: {args.output_dir}") diff --git a/tests/conftest.py b/tests/conftest.py index ebf1b0b..8278c6b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -14,7 +14,6 @@ # """Pytest configuration and fixtures.""" - import pytest