在AI开发领域,构建能够无人值守处理长任务的多代理系统是一项具有挑战性但回报丰厚的努力。本文深入探讨一个实战案例——一个多代理系统无人值守运行了4天,生成了14万行代码。我们将探索其架构、关键技术和复现该系统的实践步骤。
AI长任务执行的挑战
处理复杂的长运行任务时,AI模型经常遭受上下文遗忘的困扰。随着任务推进,早期的细节被推出上下文窗口,导致任务偏离。即使拥有大上下文窗口的模型也只是推迟了遗忘曲线,而非消除它。
解决方法是:将任务分解为小块,为每块保持干净的上下文,并让代理在完成后进行任务交接。
架构:基于图的多代理编排
该系统使用基于图的方法来编排多个代理。图中的每个节点是一个代理,边代表代理之间结构化数据的传输。
- 节点职责:每个节点(代理)只负责一个特定任务,如分析、编码、测试等。
- 通信方式:代理之间没有自由对话。它们传输预定义的结构化数据。例如,当编码代理完成时,它向测试代理交接一个包含代码详情、修改日志和下一步的结构化对象。
以下是如何在YAML文件(系统的核心配置文件)中定义编排的简化示例:
pipeline:
- name: analyzer
type: agent
responsibility: task analysis
next: [coder]
- name: coder
type: agent
responsibility: code generation
next: [tester, quality_gate]
- name: tester
type: agent
responsibility: code testing
next: [reviewer]
- name: quality_gate
type: agent
responsibility: code quality check
after: tester
next: [reviewer]
- name: reviewer
type: agent
responsibility: code review
next: [committer]
- name: committer
type: agent
responsibility: code commit
关键组件及其角色
1. Codex:编排器
Codex 充当系统的大脑。其职责包括:
- 任务分配:接收高层任务(如issue或语音命令)并生成结构化的任务草稿。
- 图初始化:根据任务类型选择YAML模板并启动管道。
- 监控与回退:持续轮询每个节点的状态,监控失败情况,并在必要时进行干预(如停止或回滚)。
2. YAML:核心接口
YAML不仅仅是配置文件,它是定义整个编排逻辑的核心接口。包括:
- 拓扑结构:定义代理图的结构(串行、并行、扇出、扇入等)。
- 代理定义:指定每个代理的角色、职责和系统提示。
- 模型配置:定义每个代理使用哪个AI模型(用于成本优化)。
- 环境:指定每个代理的Docker镜像(如编码代理可能使用基础工作镜像,而前端测试代理使用包含Chrome和Puppeteer的镜像)。
3. Scorecard:审计层
为确保系统可靠性,引入了Scorecard层。不再仅仅信任代理声称的"任务完成",Scorecard根据真实记录进行验证:
- 节点交接检查:确保节点之间的交接格式正确。
- 命令执行检查:验证测试命令有实际输出。
- PR关联检查:确保Pull Request关联到正确的Issue。
实践实现步骤
1. 设置编排图
如上所示在YAML文件中定义你的管道。该文件应该进行版本控制。
2. 实现代理
每个代理应该是一个独立模块,具有清晰的系统提示。例如,编码代理的系统提示可能如下:
你是一位专注于TypeScript的高级软件开发者。
你的任务是将Python模块转换为TypeScript,确保
类型安全并遵循定义的API契约。编码完成后,
你必须生成包含代码、修改详情和下一步的
结构化交接信息。
3. 部署环境
使用 Docker 容器化每个代理的环境。对于前端测试代理,Dockerfile可能是:
FROM node:18
RUN npm install -g puppeteer
COPY test-scripts /app/test-scripts
CMD ["node", "/app/test-scripts/run-tests.js"]
4. 使用Codex设置监控
编写Python脚本让Codex监控管道:
import requests
import time
def monitor_pipeline(pipeline_id):
while True:
response = requests.get(f"http://api.example.com/pipeline/{pipeline_id}/status")
status = response.json()["status"]
if status in ["completed", "failed"]:
break
print(f"Pipeline status: {status}")
time.sleep(30) # 每30秒轮询一次
print(f"Pipeline finished with status: {status}")
# 启动管道并获取ID
pipeline_response = requests.post("http://api.example.com/pipeline", json={"task": "TS refactoring"})
pipeline_id = pipeline_response.json()["pipeline_id"]
monitor_pipeline(pipeline_id)
性能与优化
在4天的无人值守运行中:
- 任务量:执行了416个任务,374个成功完成。
- 代码产出:生成了14万行代码,提交了516个文件。
- Token消耗:消耗了12.9亿tokens,主要由工作代理使用。
优化技巧
- Codex监控频率:最初Codex的监控过于频繁,导致成本高昂。根据任务紧急程度调整轮询间隔。
- 并行管道:初始设置是串行的。为提高效率,设计多个并行运行的管道,每个处理不同的模块。
- 环境隔离:使用Docker Compose管理复杂任务的多容器环境。
常见问题
如何在长时间无人值守运行中防止代理产生幻觉?
三种技术协同工作:(1)分块——将工作分解为小的、可验证的单元,使每个代理拥有干净、专注的上下文。(2)结构化交接——代理传递预定义的数据结构,而非自由文本,减少歧义。(3)Scorecard验证——每个代理的输出在管道继续前都根据真实记录(测试输出、PR链接、提交哈希)进行检查。这三层意味着幻觉在交接边界被捕获,而不是级联到整个管道。
建立这样的系统需要多大的团队?
一位有经验的开发者可以在一周内搭建基本版本。你需要熟悉YAML用于管道定义,Docker用于代理环境,以及基本的代理系统提示编写能力。复杂度随代理类型数量和你想处理的任务种类而增长。从简单的3代理管道(分析器→编码器→审查器)开始处理单一任务类型,然后逐步扩展。
如何控制在消耗12.9亿tokens的情况下的成本?
Token成本主要由工作代理而非编排器产生。关键策略:(1)对常规任务使用更便宜的模型(如用Haiku做代码格式化,只在复杂架构决策时用Opus)。(2)调整Codex的轮询间隔——每5秒轮询vs每30秒轮询在4天内有巨大差异。(3)实施提前终止——如果编码代理的输出两次未通过质量门,停止并升级,而不是无限重试。(4)缓存和复用结果——如果两个任务需要相同的分析步骤,运行一次并共享输出。