BabyAGI：基于优先级列表的最小自治代理

FreeGuideOnline 最新 2026-06-14

认识 BabyAGI：最小自治任务代理

BabyAGI 是一个极简、可运行的任务驱动自治代理（Autonomous Agent）。它的核心思路来自前 OpenAI 工程师 Yohei Nakajima，用一个不到 200 行的 Python 脚本演示了“人工智能为自己不断创建、排序、执行新任务”的闭环。它被设计成教学工具，让你理解创建自治代理所需的最小构件：任务列表、执行引擎、优先级排序。

此教程面向完全没有自治代理经验的开发者，你将亲手搭建一个能不断自己出题、自己回答、自己排定下一道题的 AI 助手。

BabyAGI 的核心架构

三个组件

BabyAGI 的身体里只有三个工作组件：

  1. 任务列表（Task List）——存储所有待处理任务的地方，具有优先级标记。
  2. 执行器（Execution Agent）——从列表头取任务，调用大语言模型完成任务。
  3. 任务创建器（Task Creator）——根据刚才的执行结果，产生一批新任务，再通过排序逻辑插入任务列表。

整个系统形成一个永久循环：取任务 → 执行 → 生成新任务 → 重排序优先级 → 重复。

为什么叫“Baby”

因为它没有记忆、没有反思、没有树状规划，只有一个最简单的线性任务队列。但正是这种“零假设”让它成为理解复杂自主代理（AutoGPT、LangChain Agents）的最佳起点。

环境准备与依赖安装

系统要求

  • Python 3.9+
  • OpenAI API 密钥（需要绑定结算账户）
  • 一个向量数据库（本教程使用本地 JSON 存储简化，原版用 Pinecone 或 Chroma）

创建虚拟环境并安装依赖

# 创建项目目录
mkdir babyagi-tutorial && cd babyagi-tutorial

# 创建虚拟环境
python -m venv venv
source venv/bin/activate   # Linux/Mac
# 或 venv\Scripts\activate  (Windows)

# 安装所需库
pip install openai python-dotenv

设置 OpenAI API 密钥

在项目根目录创建 .env 文件：

OPENAI_API_KEY=你的密钥

并在代码开头加载环境变量：

import os
from dotenv import load_dotenv

load_dotenv()
openai_api_key = os.getenv("OPENAI_API_KEY")

任务列表与优先级数据结构

我们将使用 Python 列表模拟任务队列。每一项任务是一个字典：

task = {
    "task_id": 1,
    "task_name": "研究量子计算的最新进展",
    "priority": 10   # 数字越高优先级越高
}

任务列表会始终按 priority 降序排列，最紧急的任务在列表最左侧。我们使用简单的排序函数维持优先级。

构建核心代理循环

执行器：完成一项任务

执行器调用 GPT-3.5/4 解决问题。它把任务名称作为提示，要求模型给出一个简洁答案。

import openai

def execute_task(task_name: str) -> str:
    prompt = f"请用简洁的中文完成以下任务：{task_name}"
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
        max_tokens=500,
        api_key=openai_api_key
    )
    return response.choices[0].message.content.strip()

任务创建器：从结果中推导新任务

任务创建器接收 已完成的任务名称 和 执行结果，要求模型产生 3 个新的相关任务，并返回任务名称列表。

def create_new_tasks(completed_task: str, result: str) -> list:
    prompt = f"""你是一个帮助拓宽思路的任务策划器。
原始任务：{completed_task}
执行结果如下：
{result}

基于该结果，请提出 3 个更深入或相关联的新任务。仅返回任务名称，每行一个，不要带编号或额外解释。"""
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.8,
        max_tokens=300,
        api_key=openai_api_key
    )
    raw = response.choices[0].message.content.strip()
    tasks = [line.strip("- ").strip() for line in raw.split("\n") if line.strip()]
    return tasks[:3]   # 最多保留前3个

优先级排序逻辑

简单优先级：所有新任务默认给予与初始任务相同的优先级。你也可以加入简单的基于关键词的启发式规则，这里我们保持纯教学版：

def prioritize_tasks(task_list, new_tasks, base_priority=10):
    next_id = max([t["task_id"] for t in task_list], default=0) + 1
    for task_name in new_tasks:
        # 去重：避免完全相同任务反复添加
        if not any(t["task_name"] == task_name for t in task_list):
            task_list.append({
                "task_id": next_id,
                "task_name": task_name,
                "priority": base_priority
            })
            next_id += 1
    # 按优先级降序排序
    task_list.sort(key=lambda x: x["priority"], reverse=True)
    return task_list

主循环：让 BabyAGI 跑起来

def main(objective: str, max_iterations: int = 5):
    # 初始化任务列表，只含一个最高优先级的目标
    task_list = [{"task_id": 1, "task_name": objective, "priority": 10}]
    
    for iteration in range(max_iterations):
        if not task_list:
            print("✅ 所有任务已完成。")
            break
            
        # 1. 取优先级最高的任务
        current_task = task_list.pop(0)
        print(f"🔧 执行任务 #{current_task['task_id']}: {current_task['task_name']}")
        
        # 2. 执行任务
        result = execute_task(current_task["task_name"])
        print(f"📌 执行结果: {result[:100]}...")
        
        # 3. 根据结果生成新任务
        new_tasks = create_new_tasks(current_task["task_name"], result)
        print(f"➕ 新任务生成: {new_tasks}")
        
        # 4. 重新排序并插入任务列表
        task_list = prioritize_tasks(task_list, new_tasks, base_priority=current_task["priority"])
        
        # 5. 显示当前队列状态
        print("📋 当前任务队列:")
        for t in task_list:
            print(f"   [{t['priority']}] {t['task_name']}")
        print("-" * 40)
        
    print("🛑 达到最大迭代次数，代理停止。")

if __name__ == "__main__":
    main(objective="撰写一篇关于机器学习在医疗领域应用的博客提纲", max_iterations=3)

运行体验

执行脚本，你会看到 BabyAGI 先写出博客提纲，然后立刻根据提纲内容生成“如何收集医疗数据研究”、“伦理问题探讨”、“深度学习案例”等新任务，并将它们排入队列，继续往下执行。这个链式反应展示了 任务驱动自治 的雏形。

解读 BabyAGI 的行为模式

为什么任务会不断变多？

因为每次 create_new_tasks 都会从执行结果中推导出“更进一步”的任务。如果没有限制，任务会指数增长。这就是为什么我们在 max_iterations 加了上限，并做了简单去重。

优先级动态调整的重要性

你看到了，我们让新任务继承被执行任务的优先级。如果你希望某些子主题更紧急，可以在 prioritize_tasks 中加入规则，例如：

  • 如果任务名称包含“安全”或“伦理”，优先级+2
  • 如果任务在结果中反复出现，适当提权

真正的自治代理会利用嵌入向量和向量数据库做语义相关性排序，但 BabyAGI 故意保持简单。

常见问题与扩展思路

如何加入持久化存储？

把 task_list 写入 JSON 文件，每次循环读写，即可中断续跑：

import json
# 加载
with open("tasks.json", "r") as f:
    task_list = json.load(f)
# 保存
with open("tasks.json", "w") as f:
    json.dump(task_list, f, indent=2)

如何添加记忆？

将每次执行的结果存入向量数据库（如 Chroma），在新任务创建时把相关记忆检索出来作为上下文，可以让 BabyAGI 拥有“长期记忆”。

如何防止无限循环？

  • 设置最大迭代次数
  • 用文本去重（完整匹配或相似度阈值）
  • 检测结果质量，当结果重复或无效时停止

总结

BabyAGI 用不到 200 行代码证明了自治代理并不神秘。它的本质就是 “执行-生成-排序”循环。掌握这个最简模型后，再去阅读 LangChain Agents、AutoGPT、CrewAI 等复杂框架，你将能瞬间抓住它们的骨架。

你现在可以：

  • 修改 execute_task 使用更复杂的工具（搜索、计算器）
  • 用真实向量数据库做优先级语义排序
  • 在子任务间注入上下文链，形成“工作流程”

这个小小的 Baby 就是你进入自主 AI 世界的第一块敲门砖。