Dropbox Nova:让 AI 编码 Agent 在工程体系里真正跑起来

2026-06-05 23 预计阅读时间:1 分钟
来源:infoq.com AI 摘要 原文链接

免责声明:本文为 AI 摘要整理,建议结合原文阅读。摘要可能省略上下文、版本差异或边界条件,不作为官方说明。

预计阅读时间:14 分钟

Dropbox 公开了内部平台 Nova——专门用来编排和规模化运行 AI 编码 Agent 的基础设施。这不是又一个 IDE 插件或单点工具,而是一整套让 Agent 从"实验玩具"变成"工程流水线一环"的操作系统。

为什么需要专门的 Agent 平台

单个 AI 编码 Agent 在本地跑 demo 很简单:打开 Copilot、写一段 prompt、拿到补全结果。但当你想让 Agent 在数百名工程师的工作流中稳定运转,问题就来了——

  • 权限与安全:Agent 需要读代码仓库、写分支、跑测试,每一步都涉及访问控制。
  • 任务编排:一个"重构任务"可能拆成代码分析→生成修改→跑 CI→创建 PR 四个子步骤,Agent 之间要衔接。
  • 资源调度:LLM 调用有速率限制和成本上限,几十个 Agent 同时跑不能把 quota 打爆。
  • 结果审计:Agent 生成的代码必须可追溯、可回滚,不能悄悄合进主分支。

Nova 的核心定位就是解决这些"从 1 个 Agent 到 100 个 Agent"的运维问题。

Nova 的关键设计思路

从 Dropbox 公开的信息来看,Nova 的架构围绕几个核心能力展开:

统一的 Agent 注册与调度

所有编码 Agent 在 Nova 上注册为"任务单元",平台负责分配任务、排队、限流。这意味着工程师不需要手动选择用哪个 Agent——Nova 根据任务类型(代码审查、bug 修复、依赖升级等)自动路由。

工作流编排而非单次调用

Nova 把一次编码任务建模为多步骤工作流,而非一次性 prompt-response。每个步骤有明确的输入输出契约,步骤之间可以串行或并行。这和传统 CI/CD pipeline 的思路一致,只是执行者从 shell 脚本变成了 AI Agent。

与现有工程基础设施的深度集成

Agent 不是在沙箱里自娱自乐——Nova 让 Agent 直接对接 Dropbox 内部的 Git 仓库、CI 系统、代码审查工具。Agent 的产出(代码变更、测试结果、审查意见)以工程师熟悉的形式呈现,比如一个标准格式的 Pull Request。

可观测性与成本控制

每个 Agent 运行都有日志、trace 和成本记录。平台层面设了 budget 和 rate limit,防止某个团队的重构任务把整个公司的 LLM 调用额度吃光。

自己搭一套迷你 Agent 编排系统

Nova 是 Dropbox 内部平台,外部无法直接使用。但它的设计思路可以落地到你自己的团队。下面是一个最小可运行的 Agent 编排框架示例,用 Python 实现,展示核心的注册、调度和工作流编排逻辑。

"""
mini_nova.py — 最小 AI 编码 Agent 编排框架
依赖:pip install openai pyyaml
运行前设置:export OPENAI_API_KEY=sk-xxx
"""

import os
import yaml
import time
import json
import uuid
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional
from openai import OpenAI

# ── Agent 注册表 ──────────────────────────────────────────

@dataclass
class AgentSpec:
    name: str
    task_type: str            # 如 "refactor", "review", "bugfix"
    steps: List[str]          # 工作流步骤名列表
    max_concurrent: int = 5   # 并发上限
    cost_budget: float = 1.0  # 单次任务美元预算上限

class AgentRegistry:
    def __init__(self):
        self._agents: Dict[str, AgentSpec] = {}

    def register(self, spec: AgentSpec):
        self._agents[spec.name] = spec
        print(f"[registry] Agent '{spec.name}' registered — type={spec.task_type}, steps={spec.steps}")

    def lookup_by_type(self, task_type: str) -> Optional[AgentSpec]:
        for spec in self._agents.values():
            if spec.task_type == task_type:
                return spec
        return None

# ── 工作流步骤执行器 ──────────────────────────────────────

client = OpenAI()  # 从环境变量读 API key

def run_llm_step(prompt: str, model: str = "gpt-4o-mini", budget: float = 1.0) -> str:
    """调用 LLM 执行单步,带简单成本估算"""
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=2048,
    )
    content = resp.choices[0].message.content
    # 粗略成本估算:gpt-4o-mini ~$0.15/1M input, $0.60/1M output
    input_tokens = resp.usage.prompt_tokens
    output_tokens = resp.usage.completion_tokens
    cost = (input_tokens * 0.15 + output_tokens * 0.60) / 1_000_000
    print(f"  [step cost] ${cost:.4f} (in={input_tokens}, out={output_tokens})")
    if cost > budget:
        raise RuntimeError(f"Step cost ${cost:.2f} exceeded budget ${budget:.2f}")
    return content

# ── 具体步骤实现(可替换为真实工具调用)───────────────────

def step_analyze(context: dict) -> dict:
    """分析代码仓库,输出变更计划"""
    prompt = f"""你是代码分析 Agent。分析以下代码片段,列出需要重构的点:

文件:{context['file_path']}
代码:

{context['code']}

输出格式:JSON 列表,每项包含 {{"issue": "...", "suggestion": "...", "priority": "high/medium/low"}}
只输出 JSON,不要其他内容。"""
    result = run_llm_step(prompt, budget=0.05)
    context["analysis"] = result
    return context

def step_generate(context: dict) -> dict:
    """根据分析结果生成修改代码"""
    prompt = f"""你是代码生成 Agent。根据以下分析结果,输出修改后的完整代码:

分析结果:
{context['analysis']}

原始代码:

{context['code']}

输出修改后的完整代码 markdown code block 包裹"""
    result = run_llm_step(prompt, budget=0.10)
    context["generated_code"] = result
    return context

def step_review(context: dict) -> dict:
    """对生成代码做自审查"""
    prompt = f"""你是代码审查 Agent检查以下生成代码是否有明显问题

```{context['generated_code']}``

输出PASS  FAIL + 具体原因"""
    result = run_llm_step(prompt, budget=0.03)
    context["review_result"] = result
    return context

STEP_FUNCS: Dict[str, Callable] = {
    "analyze": step_analyze,
    "generate": step_generate,
    "review": step_review,
}

# ── 调度器:串联工作流 ────────────────────────────────────

@dataclass
class TaskRun:
    id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])
    agent: str = ""
    status: str = "pending"
    steps_completed: List[str] = field(default_factory=list)
    context: dict = field(default_factory=dict)
    total_cost: float = 0.0

class Scheduler:
    def __init__(self, registry: AgentRegistry):
        self.registry = registry
        self._runs: Dict[str, TaskRun] = {}

    def submit(self, task_type: str, context: dict) -> TaskRun:
        spec = self.registry.lookup_by_type(task_type)
        if not spec:
            raise ValueError(f"No agent for task type '{task_type}'")
        run = TaskRun(agent=spec.name, context=context)
        self._runs[run.id] = run
        print(f"[scheduler] Task {run.id} assigned to agent '{spec.name}'")
        self._execute(run, spec)
        return run

    def _execute(self, run: TaskRun, spec: AgentSpec):
        for step_name in spec.steps:
            func = STEP_FUNCS.get(step_name)
            if not func:
                raise ValueError(f"Unknown step '{step_name}'")
            print(f"[scheduler] Task {run.id}  executing step '{step_name}'")
            run.context = func(run.context)
            run.steps_completed.append(step_name)
            run.status = "running"
        run.status = "completed"
        print(f"[scheduler] Task {run.id} completed  steps={run.steps_completed}")

# ── 启动示例 ──────────────────────────────────────────────

if __name__ == "__main__":
    registry = AgentRegistry()

    # 注册一个"重构 Agent",三步工作流
    registry.register(AgentSpec(
        name="refactor-bot",
        task_type="refactor",
        steps=["analyze", "generate", "review"],
        max_concurrent=3,
        cost_budget=0.20,
    ))

    scheduler = Scheduler(registry)

    # 提交一个重构任务
    sample_code = """def process_items(items):
    result = []
    for i in range(len(items)):
        if items[i] != None:
            if items[i] > 0:
                result.append(items[i] * 2)
    return result"""

    run = scheduler.submit("refactor", {
        "file_path": "utils/processor.py",
        "code": sample_code,
    })

    print(f"\n=== Task {run.id} Final Status: {run.status} ===")
    print(f"Steps completed: {run.steps_completed}")
    print(f"Review result:\n{run.context.get('review_result', 'N/A')}")

运行方式:

pip install openai pyyaml
export OPENAI_API_KEY=sk-your-key-here
python mini_nova.py

这个示例展示了 Nova 的三个核心概念在代码层面的映射:

Nova 概念 示例对应
Agent 注册与路由 AgentRegistrytask_type 查找
多步骤工作流 AgentSpec.steps 定义步骤序列,Scheduler 串联执行
成本控制 run_llm_step 内的 budget 检查

实际生产中,每一步会对接真实工具(Git 操作、CI 触发、PR 创建),而不是只调 LLM。步骤间的上下文传递也会用持久化存储而非内存 dict。

从 demo 到生产:需要补的几块

上面的代码是骨架,要真正在团队里跑起来,至少还要补这些:

1. 权限边界——Agent 只该操作它被授权的仓库和分支。最简单的做法是用 Git 的 fine-grained token,限制 Agent 只能推到 agent/* 前缀的分支。

2. 结果审计——每次 Agent 运行的完整 trace(prompt、response、token 用量、耗时)必须落盘。出了问题能回溯到具体哪一步、哪个 prompt 导致了错误代码。

3. 人工兜底——Agent 生成的 PR 必须经过人类审查才能合并。Nova 的设计里这应该是默认行为,而不是可选配置。在编排层面,工作流最后一步应该是 create_pr(创建审查请求),而不是 auto_merge

4. 失败重试与回退——LLM 调用会超时、返回格式错误、甚至产生幻觉代码。调度器需要处理步骤失败:重试、换模型、或标记任务为 failed 并通知负责人。

下面是一个生产级工作流定义的 YAML 示例,比 Python dict 更适合团队协作和维护:

# agent_workflows.yaml — Agent 工作流定义文件
agents:
  refactor-bot:
    task_type: refactor
    description: "分析代码质量问题并生成重构建议"
    steps:
      - name: analyze
        executor: llm
        model: gpt-4o-mini
        budget: 0.05
        prompt_template: |
          你是代码分析 Agent。分析以下代码,列出重构点。
          文件:{{file_path}}
          代码:{{code}}
          输出 JSON 列表。

      - name: generate
        executor: llm
        model: gpt-4o-mini
        budget: 0.10
        prompt_template: |
          根据分析结果生成修改代码。
          分析:{{analysis}}
          原始代码:{{code}}

      - name: self_review
        executor: llm
        model: gpt-4o-mini
        budget: 0.03
        prompt_template: |
          审查生成代码是否有明显问题。
          代码:{{generated_code}}
          输出 PASS 或 FAIL + 原因。

      - name: create_pr
        executor: git_tool  # 非 LLM 步骤,调用 Git API
        config:
          branch_prefix: "agent/refactor"
          auto_merge: false   # 关键:必须人工审查
          require_reviewers: 1

    limits:
      max_concurrent: 5
      daily_budget: 50.0     # 每天最多花 $50
      timeout_minutes: 10

  review-bot:
    task_type: code_review
    description: "对 PR 做自动代码审查,输出评论"
    steps:
      - name: diff_analysis
        executor: llm
        model: gpt-4o
        budget: 0.15
        prompt_template: |
          审查以下 diff,关注安全、性能、可维护性。
          diff:{{pr_diff}}
          输出结构化审查意见。

      - name: post_comments
        executor: github_tool
        config:
          comment_format: markdown
          link_to_lines: true

    limits:
      max_concurrent: 10
      daily_budget: 100.0

这种声明式定义的好处:非代码人员也能看懂工作流逻辑;修改步骤顺序或换模型只改 YAML,不动代码;可以接入 CI 系统,在 PR 创建时自动触发对应 Agent。

落地前的取舍清单

在团队内推行 AI 编码 Agent 平台,有几道必答题:

  • 先窄后宽:别一开始就让 Agent 改核心业务代码。从依赖升级、文档生成、测试补充等低风险任务起步,验证工作流稳定性后再扩大范围。
  • 模型选择:分析和审查步骤可以用便宜模型(gpt-4o-mini),代码生成步骤用更强模型。Nova 的多步骤设计天然支持这种混合策略。
  • 成本透明:给每个团队设可见的 LLM 调用 dashboard。当团队看到自己的 Agent 每天花了多少钱、完成了多少任务,才会认真评估 ROI。
  • 不要跳过人类审查:无论 Agent 多聪明,auto_merge 都是高风险行为。Nova 的价值在于让 Agent 的产出以人类熟悉的形式(PR、评论)呈现,而不是绕过人类。
  • Agent 不是银弹:复杂架构决策、跨团队接口设计、需要深度业务理解的修改——这些仍然需要人类工程师。Agent 最擅长的是重复性高、边界清晰的编码任务。

Dropbox 的 Nova 把 AI 编码 Agent 从"个人工具"推到了"团队基础设施"的层面。这个方向是对的——Agent 的规模化价值不在单个补全有多聪明,而在能不能像 CI/CD 一样,成为工程流程中可靠、可控、可观测的一环。


相关推荐