AI 时代七问:从工程能力到多 Agent 协作

AI 时代七问:从工程能力到多 Agent 协作

一个函数能够运行,与一个项目能够交付,中间还有很长一段路。

学习编程时,问题往往已经被整理好了:输入是什么、输出是什么、需要实现什么算法,都写在题目里。真正开始做项目后,开发者才需要自己决定这些事。需求会变化,数据未必干净,依赖可能冲突,多个模块需要衔接,系统上线后还得有人维护。

AI 让实现的速度明显加快,也让这段距离变得更容易被忽视。几轮对话就能得到页面、接口和模型代码,但功能是否满足真实需求、不同模块能否配合、错误能否定位,仍然需要一套完整的工程方法。

这篇文章围绕七个相互关联的问题展开:怎样理解工程能力,怎样开始一个项目,怎样与 Agent 分工,怎样调试,怎样组织团队,怎样安排学习,以及何时引入多 Agent。它们可以连成一条路径:从独立完成一个小项目,走向可靠地组织人、代码和 AI 工具。

一、写出代码之后,还缺哪些工程能力? #

完整项目最先考验的,往往是把模糊想法变成明确问题的能力。

“做一个图片异常检测系统”只是方向。继续追问,工程问题才会逐渐浮现:谁上传图片?用于辅助质检还是自动拦截?漏检与误报哪个代价更高?没有模型结果时怎么办?系统需要处理多少请求?数据有没有使用授权?模型能否在目标机器上运行?

这些问题决定后面的技术选择。允许人工复核的辅助工具,与直接驱动生产线的系统,即使使用同一个模型,也需要不同的验收标准、权限边界和故障处理方式。

把工程过程理解为一个反馈闭环 #

一个可以持续维护的软件项目,通常需要经历:

需求澄清 → 约束分析 → 系统与接口设计 → 最小闭环 → 实现与测试 → 部署 → 观察运行 → 反馈与迭代

这条路径并非严格的单向流水线。一次集成测试可能揭示接口设计的问题,一次数据评估可能迫使团队重新定义需求。工程设计的作用,是让这些反馈能够被理解和处理。

阶段核心问题适合留下的成果
问题定义谁使用,解决什么问题?一页需求说明
范围与约束本轮做什么,资源与风险边界在哪里?范围、非目标、环境约束
成功标准怎样判定功能与结果符合要求?验收条件、评估集
系统设计有哪些模块,数据怎样流动?架构图、模块职责
接口设计输入、输出和失败行为如何表达?API、Schema、数据契约
最小闭环怎样尽早验证关键假设?可运行的端到端演示
验证与发布如何发现回归,如何交付和恢复?测试、发布说明、回滚方案
运行与维护如何发现异常、定位问题和升级?日志、指标、Runbook、更新记录

这些成果可以很短。小项目通常不需要先写几十页设计文档,但需要能回答关键问题。

例如,异常检测应用的第一版需求可以写成:

目标用户:需要人工复核图片的质检人员。
核心流程:上传一张图片,查看分数、判断结果和模型版本。
第一版范围:单张图片、单一产品类型、一个基线模型。
暂不实现:自动控制生产线、在线训练、多租户权限系统。

验收条件:
- 有效图片能够完成上传、推理和结果展示。
- 非图片、空文件和超过大小限制的输入有明确错误。
- 模型异常时,页面显示失败状态,并提供请求编号。
- 在约定设备与测试集上记录漏检率、误报率和推理耗时。
- 新成员能够按照 README 重建环境并运行演示。

注意最后两项:模型指标需要绑定测试集,性能需要绑定硬件与负载。“准确率高”“响应快”无法直接指导验收。

算法指标与产品指标要同时看 #

异常检测项目容易把注意力集中在 AUROC 等模型指标上。实际使用时,还需要看选定阈值下的漏检率、误报率、不同产品类型的表现,以及人工复核的负担。

提高模型分数,并不自动意味着产品更有用。某个阈值如果让几乎所有图片都进入人工复核,系统可能只是把工作换了一个位置。因此,评估中既要回答“模型区分得怎样”,也要回答“用户完成工作是否更可靠、更省力”。

工程能力可以在每个小项目中反复练习:明确问题,完成最小方案,留下验证证据,修复问题,交付给别人使用,再根据反馈迭代。走完一次闭环,常常比增加几千行代码更有价值。

二、新项目从哪里开始,文件、数据与环境怎样组织? #

面对新项目,先识别最值得验证的假设,再搭建最小可运行闭环。

如果最大的不确定性是数据能否支持检测,就先检查样本和基线指标;如果不确定的是模型能否在目标设备运行,就先验证部署;如果已有可用模型,却从未做过 Web 应用,就先打通前后端与推理服务。

“MVP”是最小可行产品。工程实践中,也可以先做一个端到端的最小演示,用来验证系统能否连起来;演示跑通之后,还需要真实用户或使用场景的反馈,才能判断它是否具有产品价值。

先让数据走完全程 #

异常检测应用可以先搭建这样的链路:

图片上传
  → 类型与大小检查
  → 图像解码与预处理
  → 基线模型推理
  → 分数与阈值判断
  → JSON 响应
  → 页面展示

模型尚未就绪时,可以用固定的 Mock 结果验证接口和界面,并在界面上明确标记模拟数据。Mock 跑通能够说明链路连通,模型性能和实际推理行为仍要单独验证。

闭环成立后,再逐层替换与优化:模型、批量推理、缓存、上传交互、部署方式。这样每次修改都有清晰的比较对象。

目录首先表达职责 #

Python / ML 项目可以采用下面的起点,并随实际规模删减:

anomaly-demo/
├── README.md
├── pyproject.toml
├── uv.lock
├── .env.example
├── .gitignore
├── docs/
│   ├── requirements.md
│   ├── architecture.md
│   └── decisions/
├── src/
│   └── anomaly_demo/
│       ├── api/
│       ├── inference/
│       └── contracts.py
├── tests/
├── configs/
├── scripts/
├── notebooks/
├── data/
│   ├── raw/
│   ├── interim/
│   └── processed/
├── models/
└── outputs/

这里假定项目使用 uv;如果采用其他依赖管理工具,应使用相应锁文件,避免维护多份互不一致的依赖清单。

  • src/ 保存正式、可复用的业务代码;scripts/ 提供训练、评估等运行入口,尽量调用正式模块。
  • notebooks/ 用于探索与可视化。成熟逻辑应进入正式模块,避免服务依赖某个 Notebook 的执行顺序。
  • data/raw/ 保留原始数据;后续处理在其他目录完成,记录转换方法与数据来源。
  • models/ 保存模型或模型索引;outputs/ 保存日志、图表、评估报告等产物。
  • configs/ 保存运行参数;docs/ 保存系统认知、接口说明与重要决策。

src 布局还能够减少“在仓库根目录恰好可以导入,安装后却失败”的问题。代价是开发时通常需要先安装项目,例如使用可编辑安装;这应写进 README。Python Packaging 的官方说明解释了它与平铺布局的区别。

目录名称本身没有魔法。只有两个脚本的工具,也可以保持简单;当同类职责开始混杂、代码需要被多处调用时,再拆分模块。

一次结果应该能够追溯到一次运行 #

机器学习项目中,“代码相同”远远不够。数据、权重、预处理、驱动和配置变化,都可能改变结果。

可以在每次训练或评估时保存一份运行清单。下面的 JSON 是格式示例,尖括号内容需要由实际运行信息替换:

{
  "run_id": "baseline-001",
  "code_commit": "<Git commit SHA>",
  "working_tree_dirty": false,
  "dataset_version": "<dataset version>",
  "dataset_manifest_sha256": "<SHA-256>",
  "model_version": "baseline-v1",
  "weights_sha256": "<SHA-256>",
  "config_sha256": "<SHA-256>",
  "python_version": "<Python version>",
  "dependency_lock_sha256": "<SHA-256>",
  "runtime": "<OS / GPU / driver / CUDA>",
  "random_seed": 42
}

如果运行时存在未提交修改,单独记录 commit 无法还原代码,应同时保存必要的差异或源代码快照。随机种子有助于复现,GPU 算子、并行策略和运行环境仍可能带来不确定性。

早期项目可以先用文件记录这些信息;实验规模变大后,再根据需要引入实验追踪或数据版本管理工具。能够还原一次结果,是选择工具时的实际标准。

大型数据、权重、缓存和生成产物通常不适合直接进入 Git,但要有可追溯的存储位置、获取方式和保留策略。仓库中保留 .env.example 说明变量名称,密钥和真实凭证交给环境或密钥管理系统。

三、哪些工作可以交给 Agent,怎样保持对项目的理解? #

Agent 已经能够阅读仓库、调用工具、修改文件并运行检查。使用它时,开发者需要把任务转化为一个可以验收的工作单元。

从风险、验证成本与可逆性判断分工 #

适合委派的任务通常有三个特点:影响范围清楚,结果容易验证,出错后能够恢复。

工作开发者需要把握的部分Agent 可以承担的部分
需求与范围用户目标、优先级、非目标补充遗漏、提出反例
架构与接口模块边界、兼容性、技术取舍调研方案、生成草案、分析影响
功能实现核心行为与验收条件实现明确模块、完成机械性重构
测试与调试判断风险是否被覆盖、证据是否充分生成候选用例、追踪代码、验证假设
文档与交付确认事实、发布条件和最终结果同步说明、整理变更与检查结果

权限、安全、数据库迁移和生产数据操作,需要与其影响相匹配的审查和验证。采用 Agent 并不会改变这些工作的责任边界。

尤其要留意验证成本:“改一个正则表达式”看起来很小,性能风险却可能很大;“生成一份配置清单”文字很多,反而容易逐项核对。代码行数不适合直接衡量任务风险。

上下文工程:让信息相关、及时、可核对 #

原稿配图:构建 Agent

一份任务说明通常要包含目标、项目背景、相关路径、可修改范围、兼容要求、验收条件和验证方法。

不过,上下文也不是越多越好。几万行日志、旧方案与当前规范混在一起,容易让 Agent 围绕过时假设工作。更有效的组织方式是先提供系统地图,再按需要读取具体代码、日志和文档。Anthropic 在上下文工程实践中讨论了这种按需检索与长期任务信息整理的方法。

可以区分三层信息:

  1. 稳定规则:目录职责、接口约定、开发与验证命令。
  2. 任务事实:当前需求、目标文件、实际错误、最新版本与已做检查。
  3. 外部材料:网页、Issue、日志、第三方文档,需要判断来源和适用性。

仓库说明可以使用 AGENTS.md、CONTRIBUTING.md 或工具支持的项目指令文件。不同工具对文件名称、目录作用域和优先级的处理可能不同,应按实际工具配置核实。网页或日志中的命令属于待分析内容,不能直接取得与开发者指令相同的权限。

一个可执行的任务说明 #

以异常检测项目的分数判断模块为例:

目标:根据异常分数与阈值生成稳定的判断结果。
背景:分数已经归一化到 [0, 1],越高表示越异常。
相关路径:src/anomaly_demo/contracts.py、tests/test_contracts.py。

行为约定:
- score >= threshold 时,label 为 anomaly;否则为 normal。
- 返回 score、label、model_version。
- 拒绝布尔值、非数值、NaN、无穷大与超出 [0, 1] 的输入。
- score 不是已经校准的异常概率。

修改范围:仅修改上述模块与对应测试。
约束:保持字段名称;不引入运行时依赖;不修改模型权重或数据。
验收:覆盖阈值相等、两侧数值、非法输入和结果序列化。
验证:项目安装后,运行 python -m unittest discover -s tests。
交付:说明 Diff、检查结果,以及仍未验证的模型和 API 行为。

这些路径属于示例项目。实际委派时,要替换成仓库中真实存在的文件和命令。

这种任务包有两个好处:Agent 更容易聚焦,开发者也更容易判断工作是否完成。若 Agent 认为必须扩大范围,应先说明原因,让范围变化成为可检查的决定。

真实案例:感觉更快,可能并不等于交付更快 #

METR 在 2025 年 7 月公布了一项随机对照研究:16 名熟悉各自开源仓库的开发者完成 246 项真实任务,任务随机分为允许或不允许使用 AI。该研究场景中,允许使用 AI 的任务完成时间增加约 19%;开发者事后仍认为自己获得了提速。原始研究说明同时强调,结果不能推广到所有开发者、项目和工具。

时间边界同样重要。METR 在 2026 年 2 月更新说明:后续实验出现参与者与任务选择偏差,并行使用 Agent 也使耗时更难统计,因而新数据不足以可靠估计当时的提速幅度。后续研究说明

对项目维护者而言,更实用的做法是记录自己的完整工作成本:任务说明、实现、审查、修正、验证与集成分别花了多少时间,交付后是否出现回归。Agent 生成代码的速度,只覆盖其中一部分。

四、调试怎样从一条报错走向一条证据链? #

报错通常告诉开发者在哪里失败,未必告诉开发者为什么失败。

例如页面上传图片后显示“推理失败”,原因可能是文件损坏、接口字段变化、模型权重不兼容、GPU 内存不足,也可能是请求超时。只围绕最后一条报错反复修改,很容易让系统增加更多不确定性。

先区分现象、假设与证据 #

可靠的调试过程可以组织为:

确定预期与现象 → 稳定复现 → 收集现场信息 → 缩小范围 → 提出假设 → 设计区分性实验 → 最小修复 → 回归验证

“依赖版本不兼容”是一个假设。“某个版本下稳定失败,回退后稳定成功,堆栈落在相应接口”才构成支持它的证据。一次只改变一个关键条件,有助于判断修复为何有效。

排查层次在示例项目中可能表现为
输入与数据空文件、解码失败、RGB/BGR 次序错误、尺寸不符
业务逻辑阈值边界写错、失败状态被误判为正常
接口字段名称、类型、错误码或请求格式不一致
依赖与运行时包版本、ABI、驱动与 CUDA 不匹配
环境与配置权重路径、工作目录、权限、配置覆盖错误
基础设施网络、数据库、磁盘或 GPU 不可用
性能与并发OOM、资源泄漏、竞态、请求堆积与超时

沿输入、接口、模型和输出逐层检查,比在整个系统中同时尝试多个修补方案更容易形成结论。

让故障可以被重新观察 #

一份简洁的故障记录可以写成:

预期:有效图片返回判断结果。
实际:相同图片在服务端返回 500,命令行推理正常。
复现:使用固定样本与同一模型版本,重复请求可稳定出现。
范围:只影响 API,尚未发现命令行入口受影响。
版本:记录代码、锁文件、模型、系统与运行时版本。
近期变化:整理相关提交、配置变化与部署时间。
证据:请求编号、完整堆栈、输入摘要、日志时间线。

假设 A:API 与命令行加载了不同配置。
验证:比较实际生效的配置与权重校验值。
假设 B:服务进程的工作目录改变了相对路径解析。
验证:检查解析后的绝对路径,并用同一入口复现。

上面是分析模板,并非真实故障记录。提供给 Agent 时,应要求它说明每个假设的支持证据、反证以及下一步验证方法,避免把一种可能性直接当成结论。

Logs、Metrics、Traces 也各有用途:日志记录具体事件,指标反映延迟和资源使用等趋势,链路追踪帮助连接跨模块的调用。它们可以通过请求编号等信息相互关联,相关概念可参考 OpenTelemetry 官方说明。记录现场时应避免把凭证或完整敏感输入写入日志。

用版本历史缩小回归范围 #

如果已知某个旧版本正常,而当前版本失败,可以使用 git bisect 二分定位。下面是假设仓库中存在 v0.1.0 标签和回归测试的示例:

# 在工作区干净、复现条件稳定的情况下执行。
git bisect start
git bisect bad HEAD
git bisect good v0.1.0
git bisect run python -m unittest discover -s tests -p test_regression.py
git bisect reset

v0.1.0 与测试文件必须替换成实际存在的正常版本和用例。测试要能明确区分正常与异常;历史版本无法运行时,应核实环境问题或跳过该版本,不能把所有运行失败都判为业务回归。具体退出码与操作方式见 Git 官方文档。

真实案例:Cloudflare 的一条规则为什么能引起大范围故障? #

2019 年 7 月 2 日,Cloudflare 发布的一条 WAF 规则包含大量回溯的正则表达式,耗尽处理 HTTP/HTTPS 请求的 CPU,导致大范围 502 错误。官方复盘记录了约 27 分钟的服务中断,并指出问题同时涉及性能保护、测试、发布与恢复流程。官方故障复盘

这一案例给工程验证提供了一个具体提醒:规则在功能测试中能正确匹配,并不说明它在所有输入上具有可接受的执行成本。模拟运行也会消耗资源,需要相应的性能保护。

将这个经验带回自己的项目,可以增加长输入与边界输入的性能检查,限制单次任务的资源消耗,分阶段发布,并实际演练恢复流程。这些是由案例得到的实践建议;修复某一行代码之后,仍要检查它为什么能够影响整个系统。

五、多人协作怎样避免“各自完成,合起来失败”? #

团队协作先明确数据流、模块边界与契约,再分配实现工作。

异常检测应用可以划分为:

用户 → Web UI → API → 输入校验 → 推理服务 → 模型 → 判断结果

前端需要稳定响应,后端需要稳定推理接口,模型模块需要明确的输入格式。只有先把这些约定说明白,成员才能在不过度了解其他模块内部细节的情况下并行工作。

契约要描述字段,也要描述含义 #

“返回一个分数”还不够。这个分数的范围是什么?越大越异常还是越正常?它是概率还是相似度?阈值相等时怎样处理?失败时返回空对象、异常还是错误码?

HTTP API 可以使用 OpenAPI描述请求与响应。内部 Python 模块可以使用类型注解、dataclass 或其他模型工具,再根据需要加入运行时校验。类型注解与数据类本身不会自动验证所有业务约束。

下面是一个独立可运行的契约演示,使用 Python 3.10+ 标准库。它接收已经产生的归一化异常分数,负责验证与阈值判断;模型推理和图片处理留在各自模块中。

# 保存为 decision_demo.py,运行:python decision_demo.py
from dataclasses import asdict, dataclass
import json
import math
import unittest
from typing import Literal


@dataclass(frozen=True)
class PredictionResult:
    label: Literal["normal", "anomaly"]
    score: float
    model_version: str


def validate_unit_interval(value: float, name: str) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{name} must be a number")
    if not 0 <= value <= 1 or not math.isfinite(value):
        raise ValueError(f"{name} must be finite and within [0, 1]")
    return float(value)


def decide(
    score: float, threshold: float, *, model_version: str
) -> PredictionResult:
    score = validate_unit_interval(score, "score")
    threshold = validate_unit_interval(threshold, "threshold")
    if not isinstance(model_version, str) or not model_version.strip():
        raise ValueError("model_version must be a non-empty string")
    return PredictionResult(
        label="anomaly" if score >= threshold else "normal",
        score=score,
        model_version=model_version,
    )


class ContractTests(unittest.TestCase):
    def test_threshold_boundary(self):
        for score, expected in [(0.49, "normal"), (0.5, "anomaly"), (0.51, "anomaly")]:
            with self.subTest(score=score):
                result = decide(score, 0.5, model_version="baseline-v1")
                self.assertEqual(result.label, expected)

    def test_invalid_numbers(self):
        invalid = [True, "0.5", -0.1, 1.1, float("nan"), float("inf")]
        for value in invalid:
            for score, threshold in [(value, 0.5), (0.5, value)]:
                with self.subTest(score=score, threshold=threshold):
                    with self.assertRaises(ValueError):
                        decide(score, threshold, model_version="baseline-v1")

    def test_serialization(self):
        result = decide(0.8, 0.5, model_version="baseline-v1")
        payload = json.loads(json.dumps(asdict(result)))
        self.assertEqual(payload, {
            "label": "anomaly", "score": 0.8, "model_version": "baseline-v1"
        })

    def test_empty_model_version(self):
        with self.assertRaises(ValueError):
            decide(0.8, 0.5, model_version=" ")


if __name__ == "__main__":
    unittest.main()

这里把阈值相等、非法数值与序列化写成测试,让接口语义变成可执行约束。真实项目可以将实现移入 src/anomaly_demo/contracts.py,测试放入 tests/;API 层再把非法输入、模型不可用和超时映射为约定的响应。

分数归一化到 [0, 1],并不意味着它就是异常概率。前端文案、接口说明和模型评估需要保持同一种解释。

团队围绕契约开展并行工作 #

模块负责人需要交付的内容可以依赖的稳定边界
数据与模型数据处理、权重、模型评估与推理实现输入格式、分数含义、模型版本
后端上传校验、调用推理、错误处理与 API推理契约、响应 Schema
前端上传交互、结果与错误展示API Schema 与 Mock 响应
集成与交付端到端验证、环境、部署与恢复固定入口、版本与验收条件

负责人安排可以随团队人数调整。公共接口发生变化时,要让依赖方参与,而不是等各自完成后再发现字段不一致。

日常协作可以依靠几个简单机制:小范围提交,定期集成,自动检查,独立审查,以及重要决策留痕。例如,一份 ADR 可以记录为什么把模型独立为服务、有哪些替代方案、增加了什么成本、何时需要重新评估。

原稿配图:Git 分支协作

契约是需要维护的公共资产。CI 可以帮助检查 Schema 和响应是否符合约定,破坏兼容性的变化则需要版本与迁移计划。

六、AI 越来越会写代码,哪些能力仍然值得深入学习? #

AI 能够承担更多实现工作后,开发者需要理解更多结果:哪些代码可以进入系统,哪些建议适合当前环境,哪些测试只是覆盖了表面行为。

基础知识的价值会体现在这些判断上。看不懂事务,就难以审查数据写入;不理解进程与权限,就难以判断命令的影响;不熟悉测试,就可能把“程序跑过一次”当成可靠性证据。

按承担责任的顺序安排学习 #

原先按照工具名称排列学习清单,容易形成“安装过很多东西,却无法交付”的状态。可以把学习目标改成可观察的能力:

能力一个可以检验的学习目标
问题定义将模糊需求写成范围与验收条件
系统拆解画出数据流,说明模块边界与依赖
编程与代码阅读沿调用链理解输入、状态、输出和失败路径
调试与评估稳定复现问题,设计能区分假设的实验
Git 与环境管理查看差异、处理合并、恢复修改、重建环境
测试为业务边界与真实风险编写有效验证
网络与数据理解 HTTP、事务、缓存、并发与错误处理
交付与运行部署、观察指标、定位问题并完成恢复
Agent 协作组织上下文、限制范围、验证并集成结果

这些能力的优先级随项目变化。涉及账号和数据写入的应用,需要尽早学习权限与事务;GPU 项目则需要更早理解运行时、显存和性能分析。

把测试看作对需求的另一种表达 #

测试能够帮助 Agent 迭代,但测试本身也可能写错。

如果实现和测试都把 score > threshold 当成正确规则,它们会一致通过,却共同违反“阈值相等时判为异常”的需求。验收用例应该从业务规则和风险出发,尽量避免只复述当前实现。

同样,训练集上的高指标不能证明模型对未见数据可靠。划分数据时应考虑重复样本、同一设备或同一批次带来的泄漏,评估集要能够代表目标场景,并记录没有覆盖的分布。

对 Agent 的评估也可以区分三层:

  • 功能结果:任务是否完成,相关回归是否通过。
  • 过程约束:是否超出修改范围,是否引入未授权操作。
  • 交付成本:审查、返工、执行时间和调用成本是否合理。

一次成功只是一个样本。涉及模型判断或自主工具选择时,同一任务可能出现不同执行路径,需要在代表性任务集上观察失败类型和结果稳定性。

一个可实践的学习路线 #

选择一个规模可控的小项目,按下面的顺序扩展:

  1. 用熟悉的语言做出命令行版本,并说明输入输出。
  2. 加入测试、配置和固定运行入口,让别人能够运行。
  3. 用 Git 管理修改,练习通过 Diff 与测试检查一次重构。
  4. 加入 API 或界面,完成端到端集成。
  5. 部署到自己的测试环境,记录日志并演练恢复。
  6. 把一个边界明确的任务交给单 Agent,对照人工结果验收。
  7. 只有出现明确的并行机会时,再引入更多 Agent。

学习的尺度可以很小,责任闭环要尽量完整。工具变化后,理解系统、验证证据与恢复错误的能力仍能继续使用。

七、多 Agent 怎样工作,什么时候值得引入? #

多 Agent 通常通过任务分工、并行执行和上下文隔离,扩大系统能够处理的工作范围。能否获得收益,取决于任务是否适合拆分,以及协调成本是否可控。

先区分工作流与自主 Agent #

一个程序按固定顺序调用三个模型,是多步工作流;一个模型根据环境反馈决定下一步搜索什么、调用什么工具、何时停止,具有更强的 Agent 特征。多 Agent 系统可以把两种方式组合起来。

Anthropic 在 2024 年发布的架构文章区分了这些模式。它适合用来理解结构;具体工具与 SDK 已经演进,选型时仍应核实当前文档。

协调器分配子任务、多个工作单元执行后汇总的流程

常见的使用方式有三种:

方式谁拆分与协调任务适合的起点
人工管理多个 Agent开发者边界清楚的少量任务
主 Agent 与子 Agent主 Agent 动态派发,开发者设置约束开放式研究或较复杂任务
自建调度工作流程序管理状态、依赖、工具与验收重复运行、有明确工程需求的流程

“Builder + Reviewer”属于一种角色组合。Reviewer 应依据需求、差异和必要上下文独立检查;可以知道实现者运行了哪些测试,但还需要核对这些证据,而不是直接复述实现者的结论。两个 Agent 使用相似模型与上下文,也可能犯相似错误。

原稿配图:多 Agent 协作流程

真实案例:Anthropic 的 Research 系统 #

Anthropic 在 2025 年 6 月公开了 Research 系统的实现:主 Agent 规划研究,将不同方向交给子 Agent 搜索,再汇总结果和引用。公司报告,Opus 4 主 Agent 与 Sonnet 4 子 Agent 的组合,在其内部研究评估中相对单个 Opus 4 表现提升 90.2%。工程报告

这个数字是特定内部评估下的相对提升,并不表示准确率增加 90.2 个百分点,也不能直接推导出编程项目会获得同样收益。模型组合与资源投入均有差异,需要结合评估条件理解。

报告还指出,其观察中的多 Agent 系统消耗约为普通聊天的 15 倍 token;比较基准是聊天,并非单 Agent。这个研究案例更适合说明:多个方向可以独立检索、结果能够汇总时,并行与独立上下文可能带来收益。

对共享代码和强依赖任务,则要先处理接口、状态与集成。开更多 Agent 之前,先判断是否存在足够独立的工作。

并行边界由依赖关系决定 #

异常检测项目可以画成:

                 需求与验收
                     ↓
                 接口与数据契约
                  ↙    ↓    ↘
             推理模块  API  前端 Mock
                  ↘    ↓    ↙
                 端到端集成
                     ↓
                 审查与发布

这是一阶段的依赖图;运行后的反馈可以开启下一轮迭代。相对独立的模块能够并行,依赖上游接口的任务则需要等待契约稳定。

如果三个子任务分别耗时 T1、T2、T3,理想并行耗时接近最大值,但实际还要加上分发、等待资源、汇总、冲突处理和验收成本:

并行耗时 ≈ max(T1, T2, T3) + 调度与集成开销

这是帮助估算的模型,不是性能保证。GPU、数据库或外部 API 容量有限时,任务仍可能排队。

分支隔离代码历史,Worktree 隔离工作目录 #

仅创建不同分支,仍让多个进程操作同一个目录,文件修改会彼此干扰。Git Worktree 能够给不同分支提供独立检出目录。下面是假设仓库存在 main 分支、目标分支与目录尚未创建的示例:

git worktree add -b agent/inference ../anomaly-inference main
git worktree add -b agent/api ../anomaly-api main
git worktree list

使用时应替换成仓库的实际基准分支,在任务结束后审查、验证与集成各自提交。命令说明见 Git Worktree 官方文档。

Worktree 也有边界:它们仍属于同一仓库,不会自动隔离数据库、端口、密钥、外部服务与所有 Git 管理操作。测试数据库和服务实例需要另外安排,公共 Schema、锁文件和迁移也需要明确负责人。

对于共享状态较多的维护任务,可以采用单写者策略:多个 Agent 负责只读调查与独立审查,一个 Builder 汇总修改。这样仍能获得分工收益,也能减少并行写入冲突。

一个可运行的并行调度骨架 #

下面的 Python 3.10+ 示例演示限定并发、超时处理和结构化结果收集。demo_agent 返回模拟结果,方便在没有模型账户和 API Key 时运行;接入真实 Agent 时,替换该调用函数即可。

它展示的是调度层。模型调用、工具循环、权限控制与持久化状态,需要在实际 Agent 实现中另外提供。

# 保存为 orchestration_demo.py,运行:python orchestration_demo.py
import asyncio
from collections.abc import Awaitable, Callable
from dataclasses import asdict, dataclass
import json


@dataclass(frozen=True)
class Task:
    name: str
    objective: str


@dataclass(frozen=True)
class Result:
    task: str
    ok: bool
    summary: str
    error: str | None = None


AgentCall = Callable[[Task], Awaitable[str]]


async def run_tasks(
    tasks: list[Task], agent_call: AgentCall,
    *, concurrency: int = 2, timeout_s: float = 15,
) -> list[Result]:
    if concurrency < 1 or timeout_s <= 0:
        raise ValueError("concurrency and timeout must be positive")
    semaphore = asyncio.Semaphore(concurrency)

    async def run_one(task: Task) -> Result:
        async with semaphore:
            try:
                summary = await asyncio.wait_for(
                    agent_call(task), timeout=timeout_s
                )
                if not isinstance(summary, str) or not summary.strip():
                    raise ValueError("empty or invalid result")
                return Result(task.name, True, summary)
            except asyncio.TimeoutError:
                return Result(task.name, False, "", "timeout")
            except Exception as exc:
                return Result(task.name, False, "", type(exc).__name__)

    return list(await asyncio.gather(*(run_one(task) for task in tasks)))


async def demo_agent(task: Task) -> str:
    await asyncio.sleep(0.01)
    return f"模拟结果:{task.objective};尚未读取真实仓库。"


async def main():
    tasks = [
        Task("input", "检查输入校验边界"),
        Task("inference", "检查推理失败路径"),
        Task("tests", "检查验收用例是否完整"),
    ]
    results = await run_tasks(tasks, demo_agent)
    print(json.dumps([asdict(item) for item in results], ensure_ascii=False, indent=2))
    if not all(item.ok for item in results):
        raise SystemExit("存在失败任务,需要处理后才能进入最终验收。")


if __name__ == "__main__":
    asyncio.run(main())

这里的 ok 只代表调用得到非空结果,内容是否正确仍需单独评估。超时从任务获得并发槽位后开始计算,未覆盖排队时间;真实系统通常还需要总任务预算。

并发运行多个函数,也不等于获得工具或文件系统隔离。接入真实 Agent 后,应进一步定义:

  • 每个任务允许使用的工具、读写范围与凭证。
  • 结果中的证据路径、版本信息、未覆盖事项与成本。
  • 可重试错误、重试上限,以及已经发生写入时的幂等机制。
  • 中断后从哪里恢复,哪些检查点已经通过。
  • 谁有权判定整体完成,如何进行集成与最终验证。

实践中可以从单 Agent 的“任务—差异—验证—集成”闭环开始,再尝试两三个明确分工的工作单元。自动调度的引入时机,是手动流程已经足够清楚、重复协调工作值得被程序化的时候。

从一个完整的小项目开始 #

上述七个问题最终会落在同一份交付上:需求说得清楚,模块能够连接,结果可以验证,错误能够定位,环境可以重建,修改可以审查与恢复。

可以用下面这份清单检查自己的下一个项目:

  • 新成员能够根据说明运行最小示例。
  • 输入、输出、失败行为与修改范围有明确约定。
  • 数据、配置、代码与模型结果能够相互追溯。
  • 关键边界有测试,模型与系统指标绑定实际条件。
  • 出现异常时,能够找到日志、版本和复现路径。
  • Agent 的修改有可核对的差异与验证证据。
  • 多人或多 Agent 的工作具有明确的集成负责人。

项目越小,越适合把这些习惯练完整。当开发者能够理解系统、组织任务并验证交付,AI 才更容易成为稳定的协作力量。

参考资料与配图来源 #

以下资料优先选用官方文档、原始研究和团队工程复盘。文中数值按对应资料的发表时间引用,历史案例不代表当前所有工具的表现。

  1. Python Packaging:src layout vs flat layout——项目布局与可安装性。
  2. Anthropic:Effective context engineering for AI agents——上下文组织、按需检索与长期任务管理。
  3. METR:Measuring the Impact of Early-2025 AI on Experienced Open-Source Developer Productivity——2025 年随机对照研究与图 2。
  4. METR:We are Changing our Developer Productivity Experiment Design——2026 年的后续说明与测量局限。
  5. OpenTelemetry:Signals——可观测性信号。
  6. Git:git-bisect——版本历史中的回归定位。
  7. Cloudflare:Details of the Cloudflare outage on July 2, 2019——故障复盘与图 3。
  8. OpenAPI Specification 3.1.1——HTTP API 契约;本文引用此版本解释结构。
  9. Anthropic:Building effective agents——2024 年架构模式文章与图 5。
  10. Anthropic:How we built our multi-agent research system——2025 年 Research 系统工程报告。
  11. Git:git-worktree——独立检出目录的使用与限制。
  12. Python:dataclasses、asyncio tasks、unittest——代码示例所用标准库。

评论

0

评论功能待配置数据库后启用。