跳至内容

NVIDIA Nemotron 3 Super 教程:使用 Gradio 构建仓库分析器

学习如何通过 OpenRouter 使用 NVIDIA 的 Nemotron 3 Super,在不使用 RAG 的情况下分析 GitHub 仓库,利用提示打包和 Gradio 应用实现。
已更新 2026年10月6日  · 15分钟 阅读

使用 AI 探索

ChatGPTClaudePerplexity

长上下文模型为代码库分析打开了另一条路。与其把仓库拆成块、嵌入并检索相关片段,不如将一个中小型仓库整体打包进单个提示里,让模型端到端进行推理。

在本教程中,我将向您展示如何使用带有 Gradio 界面的 NVIDIA Nemotron 3 Super 构建 GitHub 仓库分析器。该应用接收一个公共仓库 URL,完成克隆、过滤非关键文件、估算 token 用量,并将剩余内容打包成一个结构化提示。随后通过 OpenRouter 的托管端点将提示发送给 Nemotron 3 Super,返回用于体系结构分析、上手指导、测试缺口、改进机会等任务的结构化 JSON 输出。

无需嵌入、无需向量数据库、也没有检索循环。我们完全依赖提示打包,只要仓库能放进模型的实际上下文窗口,这种方式就效果良好。读完本教程,您将得到一个 Gradio 应用,可用于:

  • 克隆并预处理公共 GitHub 仓库
  • 估算其是否符合 token 预算
  • 运行一个或多个仓库分析任务
  • 通过任务感知的 UI 呈现结构化结果

下面的视频展示了项目的实际效果(为演示起见已缩短加载时间):

您可以在此处获取本演示的完整代码。

什么是 NVIDIA Nemotron 3 Super?

Nemotron 3 Super 是 NVIDIA 面向智能体与长上下文工作流的开源推理模型。它采用混合Mamba-Transformer 潜在MoE 架构,总参数量 120B,激活参数 12B,支持多 token 预测(MTP),面向编码、工具调用和长上下文分析等重推理任务设计。

NVIDIA 将其定位于复杂的多智能体应用与整库级代码推理。

Nemotron 3 Super Architecture

图 1:Nemotron 3 Super 架构(NVIDIA 博客)

Nemotron 3 Super 背后的核心理念包括:

  • 混合 Mamba-Transformer 主干(图 1),兼顾长序列效率与精确检索;
  • 潜在 MoE 以在相同有效成本下将 token 路由至更多专家;
  • MTP 提升生成速度与推理质量;
  • 在 NVIDIA 支持的部署栈中原生提供 1M-token 上下文窗口。

在本教程中,我们将使用 OpenRouter 的托管免费端点,因为该模型的全量与量化变体在实际推理中通常至少需要一块 H100 GPU。该托管端点目前支持 262K token 的上下文窗口(非完整 1M),对于本演示中的中型仓库分析已足够。

Nemotron 3 家族

NVIDIA 的Nemotron 3 家族被推出为面向智能体 AI 系统的开放模型家族,不同规模面向不同的部署与推理需求,分为 Nano、Super 与 Ultra 等级。

从实用角度看,可以这样理解该家族:

  • Nemotron 3 Nano:轻量级系列,面向高效的目标任务与更低成本的推理,聚焦智能体工作负载与大型流程中的特定步骤。
  • Nemotron 3 Super:中量级的“严肃推理”层级,擅长更密集的技术推理、复杂编程任务、长上下文分析与更高吞吐的智能体系统。
  • Nemotron 3 Ultra:面向更重推理/部署场景的更大规模系列。

因此,这里选择 Nemotron 3 Super 更合适,因为我们需要一个能够一次性对整个仓库进行推理的模型,而不仅是解决孤立的编码子任务。

Nemotron 3 Super 教程:使用 Gradio 构建仓库分析器

本节我们将用包裹在 Gradio 界面中的 Nemotron 3 Super 构建一个仓库分析器。总体而言,最终应用将:

  • 接收公共 GitHub 仓库 URL 及可选分支名。
  • 本地克隆仓库,过滤非关键文件,并估算剩余代码是否可装入可用的上下文预算。
  • 可选地排除测试、文档或笔记本,以缩小打包后的仓库,便于分析。
  • 将过滤后的仓库打包进单个长上下文提示,并通过 OpenRouter 发送给 Nemotron 3 Super。
  • 最后返回与仓库相关任务的结构化结果。

我们一步步搭建。

步骤 1:安装依赖

在构建仓库分析器之前,我们需要一个最小环境来处理三项核心职责:UI 渲染、仓库处理与模型交互。由于不在本地运行 Nemotron,设置保持轻量与聚焦。

通过运行以下命令安装所有内容:

!pip -q install gradio openai tiktoken gitpython orjson pandas

在本项目中,我们将使用:

  • openai 与 OpenRouter 的 API(Nemotron 端点)交互
  • gradio 构建交互式 UI
  • tiktoken 在发送提示前估算 token 用量
  • orjson 快速解析模型输出的 JSON
  • pandas 将结构化结果渲染为表格
  • gitpython 克隆并操作仓库

不同于需要 CUDA、vLLM 或量化的本地 LLM 设置,这种方法完全规避了 GPU 依赖,且对 Colab 友好。

步骤 2:初始化 OpenRouter 客户端

此步我们通过 OpenRouter 配置 Nemotron 3 Super,并定义一些控制仓库处理方式的核心常量。由于不在本地运行模型,该客户端就是发送大型打包提示并接收结构化分析输出的核心接口。

OPENROUTER_API_KEY = os.environ.get("OPENROUTER_API_KEY", "")
MODEL = "nvidia/nemotron-3-super-120b-a12b:free"
WORKDIR = "/content/repo_ui"
if not OPENROUTER_API_KEY:
    raise ValueError("Set OPENROUTER_API_KEY in environment before running.")
client = OpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key=OPENROUTER_API_KEY,
)
enc = tiktoken.get_encoding("cl100k_base")

此时,应用通过配置了自定义 base_url 且在 OPENROUTER_API_KEY 中存储 API key 的 openai Python 客户端与 Nemotron 3 Super 经由 OpenRouter 通信。

所选模型指向 OpenRouter 的托管端点,让我们无需本地 GPU 基础设施即可运行超大模型。

为管理上下文大小,我们初始化一个分词器(tiktoken 的 cl100k_base),以便在发给模型前估算打包仓库将消耗多少 token,因为模型的上下文窗口是有限的。

这里有几个实践注意点:

  • OpenRouter 的免费端点会记录提示与输出,因此该设置仅应用于公共仓库。
  • 可用的上下文窗口(约 262K token)小于模型的理论上限,因此大型仓库可能需要过滤(例如排除测试、文档或笔记本)。同时出于安全原因请避免发送私有或敏感代码。
  • 由于一切经由远程 API 路由,设置保持轻量,无需依赖 H100 等高显存 GPU。

步骤 3:仓库过滤与提示规则

接下来需要两道“护栏”,即文件过滤与提示锚定。

第一道防止我们在构建产物、笔记本、图像或二进制文件等无关工件上浪费上下文。

第二道确保一旦仓库被打包,模型能始终锚定在真实代码上,而不偏离到泛泛的软件建议。

这两类规则结合起来,让仓库分析更高效。

BASE_IGNORE_DIRS = {
    ".git", ".github", "__pycache__", "node_modules", ".venv", "venv",
    "dist", "build", ".mypy_cache", ".pytest_cache", ".idea", ".vscode"
}
IGNORE_SUFFIXES = {
    ".png", ".jpg", ".jpeg", ".gif", ".webp", ".pdf", ".zip", ".gz",
    ".tar", ".mp4", ".mov", ".onnx", ".pt", ".bin", ".parquet", ".feather",
    ".ico", ".svg", ".lock"
}
MAX_FILE_BYTES = 250_000
SYSTEM_PROMPT = """You are a codebase analysis model.
You are given the complete repository contents in one prompt.
Rules:
1. Use only the provided repository content.
2. Do not invent files, modules, libraries, or dependencies.
3. Every substantial claim must cite one or more file paths.
4. Clearly separate direct observations from engineering inferences or recommendations.
5. For modernization or improvement suggestions, explicitly distinguish:
   - repo-grounded findings
   - general engineering recommendations
6. Return valid JSON only.
7. If uncertain, say so explicitly.
8. For anything involving security, dead code, dependency replacement, or testing gaps,
   present findings as likely candidates unless directly proven by repository content.
"""

以下三个过滤变量控制仓库的预处理:

  • BASE_IGNORE_DIRS 排除与代码分析无关或噪声过大的目录,如 .git、node_modules、构建目录与编辑器缓存等。
  • IGNORE_SUFFIXES 过滤对仓库推理通常无益的文件类型,包括图像、归档、视频、模型检查点及其他工件。
  • MAX_FILE_BYTES 提供额外的安全检查,跳过过大的文件,帮助将打包仓库维持在可行的上下文预算内。

SYSTEM_PROMPT 与文件过滤同样重要。

由于模型将整个仓库视为一个长提示,我们需要提供清晰的行为约束,避免编造模块、误读抽象或给出过于泛化的建议。

这些指令要求模型严格锚定在提供的仓库上,对重要论断引用文件路径,并将直接证据与一般性判断区分开来。

步骤 4:任务模式与安全防护

在具备文件过滤与系统提示后,下一步是定义一组仓库级任务,并为每个任务配对一个结构化 JSON 模式。

这带来两大好处:模型明确知道要产出哪类结果;应用则能稳定解析并在任务特定的表格中渲染结果。

TASK_DEFS: Dict[str, Dict[str, Any]] = {
    "Architecture Overview": {
        "description": "Explain the architecture, major modules, likely entrypoints, dependencies, and architectural issues.",
        "schema": {
            "repo_overview": {
                "summary": "",
                "primary_languages": [],
                "likely_entrypoints": []
            },
            "modules": [
                {
                    "name": "",
                    "responsibility": "",
                    "key_files": [],
                    "evidence_paths": []
                }
            ],
            "dependencies": [
                {
                    "from_module": "",
                    "to_module": "",
                    "relationship": "",
                    "evidence_paths": []
                }
            ],
            "issues": [
                {
                    "type": "",
                    "description": "",
                    "severity": "low|medium|high",
                    "evidence_paths": []
                }
            ],
            "unknowns_and_uncertainties": []
        },
    },
    ...
}

TASK_DEFS 字典是应用支持的所有分析模式的中央注册表。

每个任务包含一个自然语言描述,告诉模型要做什么;以及一个严格的模式,定义期望的 JSON 输出结构。

这使得同一个打包仓库可以支持非常不同的推理形式,例如架构映射、重复分析、上手指导或测试缺口检测。

例如,Architecture Overview 要求模型识别:

  • 高层次的仓库概述,
  • 主要模块,
  • 依赖关系,
  • 以及架构问题。

完整代码中的其他任务还包括:

  • Code Duplication 专注重复逻辑与相关重构机会,
  • Improvement Opportunities 识别可维护性与模块化改进点,
  • Refactor Roadmap 将发现转化为优先级工程计划,
  • Testing Gaps 标出测试薄弱区域,
  • Onboarding Guide 说明新工程师如何理解代码库,
  • Library Modernization 揭示过时或重叠的库,
  • Code Quality Hotspots 指出脆弱或易出错区域,
  • 以及 Security 或 Config Hygiene 标记风险默认配置或运维问题。
FULL_REVIEW_TASKS = [
    "Architecture Overview",
    "Improvement Opportunities",
    "Code Quality Hotspots",
    "Testing Gaps",
    "Refactor Roadmap",
]
TASK_CHOICES = [
    "Architecture Overview",
    "Code Duplication",
    "Improvement Opportunities",
    "Refactor Roadmap",
    "Testing Gaps",
    "Onboarding Guide",
    "Library Modernization",
    "Code Quality Hotspots",
    "Security / Config Hygiene",
    "Full Review",
]
def safe_str(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, str):
        return value.strip()
    return str(value).strip()

FULL_REVIEW_TASKS 列表定义了一个打包的多步骤评审。

它并非运行所有任务,而是执行一个经筛选的子集,提供对仓库较全面的工程视角。随后 TASK_CHOICES 直接用于 Gradio 下拉菜单,让用户选择单个任务或打包评审。

最后的 safe_str() 解决了一个看似“细枝末节”却很重要的稳定性问题。在开发中,一个常见的失败模式是:

'NoneType' object has no attribute 'strip'

出现该问题的情况包括:

  • Gradio 输入字段返回 None;
  • 某个任务值缺失;
  • 模型响应内容为空;
  • 或解析器接收到意外的空值。

通过用 safe_str() 归一化此类值,应用会更健壮。

步骤 5:仓库克隆与打包

在向模型发送任何内容前,我们需要把一个 GitHub 仓库转为单个结构化提示。

此步骤处理完整的摄取流程:克隆仓库、过滤无关文件、仅提取可用文本,并将一切打包为模型可一次性消费的格式。

不同于传统RAG 流水线,我们不做切块或嵌入;而是将整个仓库压缩成一个长上下文输入。因此在此环节保持选择性与结构化,对于性能与质量都至关重要。

def clone_repo(repo_url: str, branch: str = "main") -> str:
    repo_url = safe_str(repo_url)
    branch = safe_str(branch)
    if not repo_url:
        raise ValueError("Repository URL is empty.")
    if os.path.exists(WORKDIR):
        shutil.rmtree(WORKDIR)
    cmd = ["git", "clone", "--depth", "1"]
    if branch:
        cmd += ["--branch", branch]
    cmd += [repo_url, WORKDIR]
    subprocess.run(cmd, check=True, capture_output=True, text=True)
    return WORKDIR
def is_probably_text(path: Path) -> bool:
    try:
        with open(path, "rb") as f:
            chunk = f.read(2048)
        if b"\x00" in chunk:
            return False
        chunk.decode("utf-8")
        return True
    except Exception:
        return False
def iter_repo_files(root: str, exclude_tests: bool, exclude_docs: bool, exclude_notebooks: bool):
    root = Path(root)
    for path in root.rglob("*"):
        if not path.is_file():
            continue
        rel = path.relative_to(root)
        parts = set(rel.parts)
        rel_str = str(rel).lower()
        if any(part in BASE_IGNORE_DIRS for part in rel.parts):
            continue
        if exclude_tests and ("test" in parts or "tests" in parts or "/tests/" in f"/{rel_str}/"):
            continue
        if exclude_docs and ("docs" in parts or rel_str.endswith(".md") or rel_str.endswith(".rst")):
            continue
        if exclude_notebooks and rel_str.endswith(".ipynb"):
            continue
        if path.suffix.lower() in IGNORE_SUFFIXES:
            continue
        try:
            if path.stat().st_size > MAX_FILE_BYTES:
                continue
        except Exception:
            continue
        if not is_probably_text(path):
            continue
        yield path
def repo_tree_string(root: str, exclude_tests: bool, exclude_docs: bool, exclude_notebooks: bool) -> str:
    root = Path(root)
    lines: List[str] = []
    for path in sorted(root.rglob("*")):
        rel = path.relative_to(root)
        rel_str = str(rel).lower()
        parts = set(rel.parts)
        if any(part in BASE_IGNORE_DIRS for part in rel.parts):
            continue
        if path.is_file() and path.suffix.lower() in IGNORE_SUFFIXES:
            continue
        if exclude_tests and ("test" in parts or "tests" in parts or "/tests/" in f"/{rel_str}/"):
            continue
        if exclude_docs and ("docs" in parts or rel_str.endswith(".md") or rel_str.endswith(".rst")):
            continue
        if exclude_notebooks and rel_str.endswith(".ipynb"):
            continue
        lines.append(f"{rel}/" if path.is_dir() else str(rel))
    return "\n".join(lines)
def pack_repo(root: str, exclude_tests: bool, exclude_docs: bool, exclude_notebooks: bool):
    root = Path(root)
    tree = repo_tree_string(root, exclude_tests, exclude_docs, exclude_notebooks)
    files: List[Tuple[str, int, str]] = []
    total_tokens = len(enc.encode(tree))
    for path in iter_repo_files(root, exclude_tests, exclude_docs, exclude_notebooks):
        rel = path.relative_to(root)
        try:
            content = path.read_text(encoding="utf-8", errors="ignore")
        except Exception:
            continue
        block = f"\n=== FILE START: {rel} ===\n{content}\n=== FILE END: {rel} ===\n"
        tok = len(enc.encode(block))
        files.append((str(rel), tok, block))
        total_tokens += tok
    return tree, files, total_tokens
def build_repo_context(tree: str, files: List[Tuple[str, int, str]]) -> str:
    joined_files = "".join(block for _, _, block in files)
    return f"""=== REPO TREE START ===
{tree}
=== REPO TREE END ===
=== REPOSITORY FILES START ===
{joined_files}
=== REPOSITORY FILES END ===
"""

此步骤定义了完整的仓库摄取与打包流水线,但每个函数各司其职:

  • clone_repo() 函数:使用浅克隆(--depth 1)将 GitHub 仓库克隆至临时工作目录,避免拉取完整历史。诸如仓库 URL 和分支等输入会通过 safe_str() 归一化,防止运行时错误。输出为后续流程操作的本地路径(WORKDIR)。
  • iter_repo_files() 函数:主过滤层。它递归扫描仓库,只产出信号强的文件。排除系统目录、不支持的文件类型、超大文件,并可选地过滤测试、文档与笔记本。
  • is_probably_text() 函数:过滤过程中的轻量安全检查。读取每个文件一小段,验证无空字节(通常指示二进制内容)且可成功以 UTF-8 解码。仅通过检查的文件才作为文本纳入最终数据集。
  • repo_tree_string() 函数:构建仓库结构的字符串表示。它在应用相同过滤规则的基础上,输出目录与文件列表,供提示中使用,以便模型在阅读文件内容前先获得代码库布局的高层理解。
  • pack_repo() 函数:核心打包函数。遍历所有过滤后的文件,读取内容,并用 === FILE START === 与 === FILE END === 等显式标记包裹每个文件。同时使用 tiktoken 估算 token 用量,并累加总数。
  • build_repo_context() 函数:将一切合并为单个结构化提示。先放仓库树,再放全部文件内容。模型先理解结构,再读取实现细节。

净效果是:将原始 GitHub 仓库转化为单个、结构化的长上下文输入。

步骤 6:任务提示与 JSON 解析

仓库过滤与打包为单个长上下文输入后,下一步是告诉模型应执行哪类分析,以及如何格式化结果。

我们不为每种用例发送相同的通用指令,而是生成任务特定的提示,其中包含所选任务名、其描述,以及期望响应的精确 JSON 模式。

def build_task_prompt(task_name: str, repo_context: str) -> str:
    task_name = safe_str(task_name)
    if not task_name:
        raise ValueError("Task name is empty.")
    if task_name not in TASK_DEFS:
        raise ValueError(f"Unknown task: {task_name}")
    task_def = TASK_DEFS[task_name]
    return f"""Analyze this full repository.
Task Name:
{task_name}
Task Description:
{task_def["description"]}
Return JSON matching this schema exactly:
{json.dumps(task_def["schema"], indent=2)}
{repo_context}
"""
def try_parse_json(text: str) -> Dict[str, Any]:
    text = safe_str(text)
    if not text:
        raise ValueError("Model returned empty text.")
    try:
        return orjson.loads(text)
    except Exception:
        start = text.find("{")
        end = text.rfind("}")
        if start != -1 and end != -1 and end > start:
            return orjson.loads(text[start:end+1])
        raise ValueError("Could not parse model output as JSON.")

上述代码定义了两个密切相关的函数,连接了打包仓库与任务特定分析输出。

  • build_task_prompt() 函数:创建发送给模型的最终提示。先用 safe_str() 归一化任务名,避免空输入导致流水线中断;再校验任务是否存在于 TASK_DEFS 中,防止发送不支持或拼错的任务名。校验通过后,取出该任务的描述与 JSON 模式,并与完整的打包仓库上下文一起嵌入最终提示。
  • try_parse_json() 函数:处理推理后的模型输出。实践中,长上下文模型有时会在 JSON 外包裹额外文本、格式片段或解释性内容。为增强鲁棒性,函数先直接用 orjson.loads() 解析;若失败,则尝试截取首个“{”到最后一个“}”之间的子串再次解析。严格提示与宽容解析的结合显著提升了稳定性。

下一步我们将把该提示逻辑与模型调用相连,开始生成结构化的仓库分析结果。

步骤 7:模型推理

此阶段,我们通过 OpenRouter 将提示发送给 Nemotron 3 Super 来运行分析。

我们不进行拆分或检索仓库部分,而是将整个过滤后的代码库在一次请求中发送,让模型跨文件、模块与依赖进行全局推理。

def call_model(prompt: str, max_tokens: int = 5000) -> str:
    prompt = safe_str(prompt)
    if not prompt:
        raise ValueError("Prompt is empty.")
    resp = client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": prompt},
        ],
        temperature=0.2,
        max_tokens=max_tokens,
        extra_headers={
            "HTTP-Referer": "https://colab.research.google.com",
            "X-Title": "Whole Repo Analysis Demo"
        }
    )
    if not getattr(resp, "choices", None):
        raise ValueError("Model response has no choices.")
    message = getattr(resp.choices[0], "message", None)
    if message is None:
        raise ValueError("Model response has no message.")
    content = getattr(message, "content", None)
    content = safe_str(content)
    if not content:
        raise ValueError("Model returned empty content.")
    return content

call_model() 负责执行实际的模型调用并提取响应。完整流程如下:

  • 输入校验:函数先归一化提示,防止向 API 发送无效请求。
  • 构造模型请求:使用兼容 OpenAI 的客户端,以 OpenRouter 为后端。负载包含强制锚定、JSON 输出与基于证据推理的系统提示,以及包含完整任务提示与仓库上下文的用户消息。温度设为 0.2,以保持输出更确定;max_tokens 限制响应长度,促使模型聚焦于结构化输出。
  • 响应校验:检查响应中是否存在 choices,并验证 message 对象是否存在;同时提取并归一化 content。若缺失,抛出错误。
  • 空响应处理:最后一道防护确保即使 API 调用成功,空字符串也不会向下游传播,从而避免“模型返回空内容”这类问题与后续 JSON 解析失败。

最后,我们可以进入模型推理部分,了解如何通过 OpenRouter 托管模型运行分析。

步骤 8:辅助函数

接下来我们需要一些小工具函数,方便 UI 与输出处理。

def to_df(items, columns=None):
    if items is None:
        return pd.DataFrame(columns=columns or [])
    if isinstance(items, dict):
        return pd.DataFrame([items])
    if not items:
        return pd.DataFrame(columns=columns or [])
    return pd.DataFrame(items)
def get_tasks_to_run(task: str) -> List[str]:
    task = safe_str(task)
    if not task:
        raise ValueError("Task is empty.")
    return FULL_REVIEW_TASKS if task == "Full Review" else [task]
def visible_for_task(task: str) -> Dict[str, bool]:
    task = safe_str(task)
    if task == "Full Review":
        return {
            "modules": True,
            "deps": True,
            "issues": True,
            "dups": False,
            "refactor": False,
            "improvements": True,
            "roadmap": True,
            "testing": True,
            "onboarding": False,
            "libs": False,
            "hotspots": True,
            "security": False,
        }
    return {
        "modules": task == "Architecture Overview",
        "deps": task == "Architecture Overview",
        "issues": task == "Architecture Overview",
        "dups": task == "Code Duplication",
        "refactor": task == "Code Duplication",
        "improvements": task == "Improvement Opportunities",
        "roadmap": task == "Refactor Roadmap",
        "testing": task == "Testing Gaps",
        "onboarding": task == "Onboarding Guide",
        "libs": task == "Library Modernization",
        "hotspots": task == "Code Quality Hotspots",
        "security": task == "Security / Config Hygiene",
    }
def render_markdown_summary(repo_url: str, total_tokens: int, files, results: Dict[str, Dict[str, Any]]) -> str:
    repo_url = safe_str(repo_url)
    lines = [
        "# Whole-Repo Analysis Report",
        f"**Repository:** {repo_url}",
        f"**Files included:** {len(files)}",
        f"**Estimated input tokens:** {total_tokens}",
        "",
    ]
    if "Architecture Overview" in results:
        overview = results["Architecture Overview"].get("repo_overview", {})
        lines += [
            "## Architecture",
            overview.get("summary", "No summary returned."),
            "",
            f"**Primary languages:** {', '.join(overview.get('primary_languages', [])) or 'N/A'}",
            f"**Likely entrypoints:** {', '.join(overview.get('likely_entrypoints', [])) or 'N/A'}",
            "",
        ]
    if "Improvement Opportunities" in results:
        lines += [
            "## Improvement Opportunities",
            f"- Identified: {len(results['Improvement Opportunities'].get('improvement_opportunities', []))}",
            "",
        ]
    if "Code Quality Hotspots" in results:
        lines += [
            "## Code Quality Hotspots",
            f"- Identified: {len(results['Code Quality Hotspots'].get('quality_hotspots', []))}",
            "",
        ]
    if "Testing Gaps" in results:
        lines += [
            "## Testing Gaps",
            f"- Identified: {len(results['Testing Gaps'].get('testing_gaps', []))}",
            "",
        ]
    if "Code Duplication" in results:
        dup = results["Code Duplication"]
        lines += [
            "## Duplication & Refactor Summary",
            f"- Duplication candidates: {len(dup.get('duplication_candidates', []))}",
            f"- Design smells: {len(dup.get('design_smells', []))}",
            f"- High risk files: {len(dup.get('high_risk_files', []))}",
            f"- Refactor recommendations: {len(dup.get('refactor_recommendations', []))}",
            "",
        ]
    if "Library Modernization" in results:
        lines += [
            "## Library Modernization",
            f"- Recommendations: {len(results['Library Modernization'].get('library_recommendations', []))}",
            "",
        ]
    if "Security / Config Hygiene" in results:
        lines += [
            "## Security / Config Hygiene",
            f"- Findings: {len(results['Security / Config Hygiene'].get('security_and_config_findings', []))}",
            "",
        ]
    if "Refactor Roadmap" in results:
        lines += [
            "## Refactor Roadmap",
            f"- Planned items: {len(results['Refactor Roadmap'].get('refactor_roadmap', []))}",
            "",
        ]
    if "Onboarding Guide" in results:
        onboarding = results["Onboarding Guide"].get("onboarding_guide", {})
        lines += [
            "## Onboarding Guide",
            onboarding.get("summary", "No summary returned."),
            "",
            f"**Recommended reading order:** {', '.join(onboarding.get('recommended_reading_order', [])) or 'N/A'}",
            f"**Entrypoints:** {', '.join(onboarding.get('entrypoints', [])) or 'N/A'}",
            "",
        ]
    return "\n".join(lines)
def hidden_update():
    return gr.update(value=pd.DataFrame(), visible=False)
def visible_update(df):
    return gr.update(value=df, visible=True)
def empty_outputs(stats, status_message):
    return (
        status_message,
        "",
        *all_hidden(),
        "{}",
        json.dumps(stats, indent=2),
    )

这一组辅助函数数量不多,却共同承担了应用的大部分呈现逻辑。

  • to_df() 函数:将模型输出转为 pandas DataFrame,以便在 Gradio 表格中整洁呈现。
  • get_tasks_to_run() 函数:决定运行单个任务还是打包的 Full Review。当选择“Full Review”时,会展开为预定义的评审任务列表;否则返回仅包含所选任务的一项列表。
  • visible_for_task() 函数:控制 UI 中应显示哪些报告表。比如选择 Onboarding Guide 时,无需显示重复或安全表。
  • render_markdown_summary() 函数:非直接输出原始 JSON,而是生成可读性更好的摘要,包含仓库元数据、token 计数与任务亮点。
  • hidden_update() 与 visible_update() 函数:用于控制 Gradio 组件状态。hidden_update() 清空并隐藏表格,visible_update() 填充数据并显示。由于 Gradio 期望为每个输出组件返回值,这一显式重置模式尤为重要。
  • empty_outputs() 函数:处理提前退出与错误状态。若仓库过大、任务无效或其他失败,应用仍需为每个 UI 组件返回有效输出。

这些辅助函数不改变核心模型逻辑,却让应用可用可看。

步骤 9:端到端仓库分析

此函数编排完整工作流。它校验输入、准备仓库、运行一个或多个分析任务,并将结构化输出返回至 Gradio UI。

def analyze_repo(repo_url, branch, task, token_budget, exclude_tests, exclude_docs, exclude_notebooks):
    repo_url = safe_str(repo_url)
    branch = safe_str(branch)
    task = safe_str(task)
    try:
        if not repo_url:
            stats = {"repo_url": "", "branch": branch}
            return empty_outputs(stats, "Error: Repository URL is empty.")
        if not task:
            stats = {"repo_url": repo_url, "branch": branch}
            return empty_outputs(stats, "Error: Task is empty.")
        repo_path = clone_repo(repo_url, branch)
        tree, files, total_tokens = pack_repo(repo_path, exclude_tests, exclude_docs, exclude_notebooks)
        stats = {
            "repo_url": repo_url,
            "branch": branch,
            "files_included": len(files),
            "estimated_input_tokens": total_tokens,
            "token_budget": token_budget,
            "fits_budget": total_tokens <= token_budget,
            "sample_files": [f[0] for f in files[:30]],
            "tasks_requested": get_tasks_to_run(task),
        }
        if total_tokens > token_budget:
            msg = (
                f"Repo too large for current budget.\n\n"
                f"Estimated tokens: {total_tokens}\n"
                f"Budget: {token_budget}\n\n"
                f"Try excluding tests/docs/notebooks or use a smaller repo."
            )
            return empty_outputs(stats, msg)
        repo_context = build_repo_context(tree, files)
        tasks_to_run = get_tasks_to_run(task)
        results: Dict[str, Dict[str, Any]] = {}
        for task_name in tasks_to_run:
            prompt = build_task_prompt(task_name, repo_context)
            raw_text = call_model(prompt)
            results[task_name] = try_parse_json(raw_text)
        report_md = render_markdown_summary(repo_url, total_tokens, files, results)
        vis = visible_for_task(task)
        arch_json = results.get("Architecture Overview", {})
        dup_json = results.get("Code Duplication", {})
        imp_json = results.get("Improvement Opportunities", {})
        roadmap_json = results.get("Refactor Roadmap", {})
        testing_json = results.get("Testing Gaps", {})
        onboarding_json = results.get("Onboarding Guide", {})
        lib_json = results.get("Library Modernization", {})
        hotspot_json = results.get("Code Quality Hotspots", {})
        security_json = results.get("Security / Config Hygiene", {})
        modules_df = to_df(arch_json.get("modules", []), ["name", "responsibility", "key_files", "evidence_paths"])
        deps_df = to_df(arch_json.get("dependencies", []), ["from_module", "to_module", "relationship", "evidence_paths"])
        issues_df = to_df(arch_json.get("issues", []), ["type", "description", "severity", "evidence_paths"])
        dups_df = to_df(dup_json.get("duplication_candidates", []), ["summary", "files_involved", "why_it_looks_duplicate", "confidence"])
        refactor_df = to_df(dup_json.get("refactor_recommendations", []), ["title", "description", "target_files", "expected_benefit"])
        improvements_df = to_df(imp_json.get("improvement_opportunities", []), ["title", "description", "why_it_matters", "affected_files", "suggested_change", "expected_impact", "effort", "evidence_paths"])
        roadmap_df = to_df(roadmap_json.get("refactor_roadmap", []), ["priority", "title", "description", "target_files", "impact", "effort", "why_now"])
        testing_df = to_df(testing_json.get("testing_gaps", []), ["module_or_file", "why_it_needs_tests", "likely_missing_cases", "recommended_test_type", "evidence_paths"])
        onboarding_df = to_df(onboarding_json.get("onboarding_guide", {}), ["summary", "recommended_reading_order", "entrypoints", "important_directories", "debugging_start_points", "common_confusion_points"])
        libs_df = to_df(lib_json.get("library_recommendations", []), ["current_library_or_pattern", "where_used", "suggested_alternative", "why_better", "migration_difficulty", "confidence", "evidence_paths", "recommendation_type"])
        hotspots_df = to_df(hotspot_json.get("quality_hotspots", []), ["file_or_module", "risk_level", "why_it_is_a_hotspot", "evidence_paths", "recommended_cleanup"])
        security_df = to_df(security_json.get("security_and_config_findings", []), ["title", "severity", "description", "affected_files", "evidence_paths", "recommended_fix", "confidence"])
        status = (
            f"Done.\n\n"
            f"Files included: {len(files)}\n"
            f"Estimated input tokens: {total_tokens}\n"
            f"Tasks run: {', '.join(tasks_to_run)}"
        )
        return (
            status,
            report_md,
            visible_update(modules_df) if vis["modules"] else hidden_update(),
            visible_update(deps_df) if vis["deps"] else hidden_update(),
            visible_update(issues_df) if vis["issues"] else hidden_update(),
            visible_update(dups_df) if vis["dups"] else hidden_update(),
            visible_update(refactor_df) if vis["refactor"] else hidden_update(),
            visible_update(improvements_df) if vis["improvements"] else hidden_update(),
            visible_update(roadmap_df) if vis["roadmap"] else hidden_update(),
            visible_update(testing_df) if vis["testing"] else hidden_update(),
            visible_update(onboarding_df) if vis["onboarding"] else hidden_update(),
            visible_update(libs_df) if vis["libs"] else hidden_update(),
            visible_update(hotspots_df) if vis["hotspots"] else hidden_update(),
            visible_update(security_df) if vis["security"] else hidden_update(),
            json.dumps(results, indent=2),
            json.dumps(stats, indent=2),
        )
    except subprocess.CalledProcessError as e:
        stats = {"repo_url": repo_url, "branch": branch}
        stderr = safe_str(getattr(e, "stderr", "")) or safe_str(e)
        return empty_outputs(stats, f"Git clone failed:\n{stderr}")
    except Exception as e:
        stats = {"repo_url": repo_url, "branch": branch}
        return empty_outputs(stats, f"Error: {safe_str(e)}")

该函数编排了从原始仓库到结构化输出的全流程。

  • 输入处理与仓库准备:克隆仓库并将其打包为结构化提示,同时估算 token。若超出预算则提前退出。
  • 任务执行循环(核心逻辑):针对每个请求的任务构建任务特定的提示,调用模型并解析结构化 JSON 输出。所有结果统一存储在 results 字典中。
  • 为 UI 结构化输出:通过 to_df() 将原始 JSON 输出转为 DataFrame,并映射到 UI 组件。可见性由 visible_for_task() 控制,确保仅显示与所选任务相关的表格。
  • 报告与错误处理:Markdown 报告汇总关键信息,同时保留完整 JSON 输出便于调试。所有失败(克隆问题、空响应、解析错误)均通过 empty_outputs() 安全处理,保持 UI 稳定。

步骤 10:构建 Gradio UI

最后,我们将一切封装进一个简洁的 Gradio 界面。

with gr.Blocks(title="Nemotron Repo Analyzer") as demo:
    gr.Markdown("# Nemotron Repo Analyzer")
    # gr.Markdown("Analyze a public GitHub repository without RAG using a single packed prompt.")
    with gr.Tab("Analyze"):
        with gr.Row():
            repo_url = gr.Textbox(
                label="GitHub Repo URL",
                value="https://github.com/AashiDutt/Nemotron-3-Nano.git"
            )
            branch = gr.Textbox(label="Branch", value="main")
        with gr.Row():
            task = gr.Dropdown(
                choices=TASK_CHOICES,
                value="Full Review",
                label="Task"
            )
            token_budget = gr.Slider(
                minimum=40000,
                maximum=180000,
                step=5000,
                value=120000,
                label="Token Budget"
            )
        with gr.Row():
            exclude_tests = gr.Checkbox(value=False, label="Exclude tests")
            exclude_docs = gr.Checkbox(value=False, label="Exclude docs / markdown")
            exclude_notebooks = gr.Checkbox(value=True, label="Exclude notebooks")
        analyze_btn = gr.Button("Analyze Repo", variant="primary")
        status = gr.Textbox(label="Status", lines=7)
    with gr.Tab("Report"):
        report_md = gr.Markdown()
        modules_df = gr.Dataframe(label="Modules", visible=False)
        deps_df = gr.Dataframe(label="Dependencies", visible=False)
        issues_df = gr.Dataframe(label="Architecture Issues", visible=False)
        dups_df = gr.Dataframe(label="Duplication Candidates", visible=False)
        refactor_df = gr.Dataframe(label="Refactor Recommendations", visible=False)
        improvements_df = gr.Dataframe(label="Improvement Opportunities", visible=False)
        roadmap_df = gr.Dataframe(label="Refactor Roadmap", visible=False)
        testing_df = gr.Dataframe(label="Testing Gaps", visible=False)
        onboarding_df = gr.Dataframe(label="Onboarding Guide", visible=False)
        libs_df = gr.Dataframe(label="Library Modernization", visible=False)
        hotspots_df = gr.Dataframe(label="Code Quality Hotspots", visible=False)
        security_df = gr.Dataframe(label="Security / Config Hygiene", visible=False)
    with gr.Tab("Raw JSON"):
        all_results_json_box = gr.Code(label="All Task Results JSON", language="json")
        stats_box = gr.Code(label="Run Stats", language="json")
    analyze_btn.click(
        analyze_repo,
        inputs=[repo_url, branch, task, token_budget, exclude_tests, exclude_docs, exclude_notebooks],
        outputs=[
            status,
            report_md,
            modules_df,
            deps_df,
            issues_df,
            dups_df,
            refactor_df,
            improvements_df,
            roadmap_df,
            testing_df,
            onboarding_df,
            libs_df,
            hotspots_df,
            security_df,
            all_results_json_box,
            stats_box,
        ]
    )
demo.launch(debug=True, share=True)

gr.Blocks() 容器定义了应用的整体布局,顶部 Markdown 设置标题与上下文。界面划分为三个选项卡,保证流程清晰。

  • “Analyze”选项卡作为输入层。它收集仓库 URL 与分支,允许用户选择任务并配置 token 预算;还提供排除测试、文档或笔记本的开关,有助于控制提示大小与噪声。“Analyze Repo”按钮触发完整流水线,状态文本框显示进度、错误或完成信息。
  • “Report”选项卡是主要输出层。它先呈现 Markdown 摘要用于快速浏览,再给出结构化的 DataFrame 表格以便检查。
  • “Raw JSON”选项卡用于调试与透明性。它展示模型返回的完整结构化输出及运行元数据,便于校验、调试或导出到下游系统。

这将用户输入直接连接到 analyze_repo() 函数,并将其输出映射回 UI 组件。这里输出的顺序至关重要,因为每个返回值都会路由到特定 UI 元件。

最后,通过以下命令启动应用:

demo.launch(debug=True, share=True)

debug=True 有助于在开发期间暴露运行时错误,而 share=True 会创建一个公共链接,便于在无需部署的情况下共享应用。

结语

在本教程中,我演示了如何结合 NVIDIA Nemotron 3 Super、OpenRouter 与 Gradio 构建代码库分析器。

应用不依赖切块、嵌入与检索,而是克隆公共 GitHub 仓库,对代码库进行过滤并打包为单个提示,请求模型一次性对整个仓库进行推理。

这种“全上下文”方法适用于需要全局视角的任务,如架构讲解、上手指导、测试缺口分析、改进机会与重构规划。

接下来,您可以从以下方面扩展项目:

  • 添加“预估适配”按钮,在分析前估算 token 数;
  • 为 Full Review 添加部分失败处理;
  • 支持下载 Markdown 报告;
  • 或将仓库提示打包与同一仓库上的 RAG 基线进行对比。

FAQs

本教程是否使用了 Nemotron 3 Super 的完整 1M-token 上下文窗口?

不会。NVIDIA 在官方 Nemotron 3 Super 部署中文档标称最高 1M token,但 OpenRouter 的托管免费端点当前暴露的是 262,144 上下文,本教程已足够使用。

为什么不在本地运行 Nemotron 3 Super?

Nemotron 3 Super 是一个大型推理模型(通过 Ollama 约 85Gb),对于这样的教程,使用托管端点要容易得多。

哪个 Nemotron 任务对这个应用最可靠?

实践中,最稳健的任务包括:

  • Architecture Overview
  • Improvement Opportunities
  • Testing Gaps
  • Onboarding Guide
  • Refactor Roadmap

这些任务最能从一次性查看完整仓库中获益。

我是否应默认使用 Full Review?

对于较小的公共仓库可以。对于较大的仓库,单任务分析通常更稳定。


Aashi Dutt's photo
Author
Aashi Dutt
LinkedIn
Twitter

我是一名 Google Developers 机器学习(生成式 AI)领域的专家、Kaggle 三项专家,以及 Women Techmakers 大使,拥有 3 年以上的技术从业经验。2020 年我共同创办了一家健康科技初创公司,目前在佐治亚理工学院攻读计算机科学硕士,专攻机器学习。

主题
人工智能
大语言模型

Top DataCamp Courses

课程

使用 LangChain 设计 Agentic 系统

3 小时
14.8K
掌握 LangChain agent 的基础组件,构建自定义聊天 agent。
查看详情Right Arrow
开始课程
查看更多Right Arrow