BabyAGI — Minimal Task-Queue Agent Pattern

Quick Start — The Pattern in 60 Lines

This is a faithful 60-line port of the BabyAGI idea. Three LLM roles — execute, create, prioritize — on a shared task queue. For a real product, migrate each role to a CrewAI agent or a LangGraph node with memory and tools; BabyAGI’s value is making the pattern legible, not running in production.

# A minimal BabyAGI-style loop using the modern OpenAI SDK.
# This is the pattern — use CrewAI / LangGraph for production.

from collections import deque
from openai import OpenAI

client = OpenAI()
OBJECTIVE = "Write a 3-step plan to launch a small SaaS landing page."
task_list = deque([{"id": 1, "name": "Brainstorm the first task list for the objective."}])

def llm(prompt: str) -> str:
    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
    )
    return r.choices[0].message.content.strip()

def execute(task: str, context: str) -> str:
    return llm(f"Objective: {OBJECTIVE}\nPrior context: {context}\nTask: {task}\nExecute it now.")

def creation(task: str, result: str) -> list[str]:
    raw = llm(f"Objective: {OBJECTIVE}\nLast task: {task}\nResult: {result}\n"
              "Generate up to 3 NEW tasks (no duplicates). Return a bullet list, no preamble.")
    return [ln.lstrip("-•* ").strip() for ln in raw.splitlines() if ln.strip()]

def prioritize(tasks: list[dict]) -> list[dict]:
    raw = llm(f"Objective: {OBJECTIVE}\nUnordered tasks:\n"
              + "\n".join(f"- {t['name']}" for t in tasks)
              + "\nRe-order by importance toward the objective. Return a bullet list.")
    names = [ln.lstrip("-•* ").strip() for ln in raw.splitlines() if ln.strip()]
    return [{"id": i + 1, "name": n} for i, n in enumerate(names) if n]

context = ""
for _ in range(5):     # cap iterations
    if not task_list: break
    task = task_list.popleft()
    result = execute(task["name"], context)
    context += f"\n[TASK] {task['name']}\n[RESULT] {result[:300]}"
    print(f"\nDone: {task['name']}")
    new = creation(task["name"], result)
    current = list(task_list) + [{"id": t["id"] + i, "name": n} for i, n in enumerate(new, 1)]
    task_list = deque(prioritize(current))

print("\n=== FINAL CONTEXT ===")
print(context[-1200:])

Key Features

Comparison

Use Cases

Pricing & License

Frequently Asked Questions