BabyAGI 任务 Agent¶

git clone https://github.com/yoheinakajima/babyagi.git
cd babyagi
pip install -r requirements.txt

# 配置环境变量
cp .env.example .env

# .env
OPENAI_API_KEY=sk-your-key
# 或使用本地模型
# OPENAI_API_BASE=http://localhost:11434/v1
# OPENAI_API_MODEL=llama3

# 向量存储
TABLE_NAME=baby-agi-test
# 使用 Chroma（默认）
RESULTS_STORE_NAME=chroma

# 可选：Pinecone
# PINECONE_API_KEY=your-key
# PINECONE_ENVIRONMENT=us-east-1

python babyagi.py

# 或带参数
python babyagi.py \
  --objective "研究并总结 2024 年 AI Agent 领域的主要进展" \
  --first_task "搜索最新的 AI Agent 论文和项目"

# BabyAGI 的核心就是这个循环：
while True:
    # 1. 取出优先级最高的任务
    task = task_list.pop(0)

    # 2. 执行任务
    result = execution_agent(objective, task, context)

    # 3. 存储结果到向量数据库
    store_result(task, result)

    # 4. 根据结果创建新任务
    new_tasks = task_creation_agent(
        objective, result, task, task_list
    )

    # 5. 将新任务加入队列
    task_list.extend(new_tasks)

    # 6. 重新排列优先级
    task_list = prioritization_agent(objective, task_list)

import openai
from collections import deque

# 配置
OBJECTIVE = "创建一个 Python Web 爬虫教程大纲"
FIRST_TASK = "列出学习 Web 爬虫需要掌握的基础知识"

task_list = deque([{"task_id": 1, "task_name": FIRST_TASK}])
completed_tasks = []

def execution_agent(objective: str, task: str, context: str) -> str:
    prompt = f"""你是一个任务执行 AI。
目标: {objective}
当前任务: {task}
之前的上下文: {context}
请完成这个任务并返回结果。"""

    response = openai.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

def task_creation_agent(objective: str, result: str, 
                        current_task: str, task_list: list) -> list:
    prompt = f"""你是一个任务创建 AI。
目标: {objective}
刚完成的任务: {current_task}
执行结果: {result}
当前待执行: {[t['task_name'] for t in task_list]}

基于结果，创建新的待办任务（如果需要）。
每行一个任务，不要重复已有任务。"""

    response = openai.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    new_tasks = response.choices[0].message.content.strip().split("\n")
    return [{"task_name": t.strip()} for t in new_tasks if t.strip()]

def prioritization_agent(objective: str, task_list: list) -> list:
    prompt = f"""你是一个任务优先级 AI。
目标: {objective}
任务列表: {[t['task_name'] for t in task_list]}

请按优先级重新排序这些任务（最重要的在前）。
每行返回一个任务。"""

    response = openai.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    sorted_tasks = response.choices[0].message.content.strip().split("\n")
    return [{"task_id": i+1, "task_name": t.strip()} 
            for i, t in enumerate(sorted_tasks) if t.strip()]

# 主循环
MAX_ITERATIONS = 5
for i in range(MAX_ITERATIONS):
    if not task_list:
        break

    task = task_list.popleft()
    print(f"\n--- 执行任务 {i+1}: {task['task_name']} ---")

    context = "\n".join([f"- {t['task_name']}: {t.get('result', '')}" 
                        for t in completed_tasks[-3:]])

    result = execution_agent(OBJECTIVE, task['task_name'], context)
    print(f"结果: {result[:200]}...")

    task['result'] = result
    completed_tasks.append(task)

    new_tasks = task_creation_agent(OBJECTIVE, result, task['task_name'], list(task_list))
    task_list.extend(new_tasks)

    if task_list:
        task_list = deque(prioritization_agent(OBJECTIVE, list(task_list)))

print("\n=== 完成 ===")

import requests
from bs4 import BeautifulSoup

TOOLS = {
    "web_search": lambda query: search_web(query),
    "read_url": lambda url: read_webpage(url),
    "write_file": lambda name, content: write_to_file(name, content),
}

def enhanced_execution_agent(objective, task, context, tools):
    # 先让 LLM 决定是否需要使用工具
    tool_prompt = f"""任务: {task}
可用工具: {list(tools.keys())}
是否需要使用工具？如果需要，返回格式: TOOL:tool_name:参数
如果不需要，返回 NO_TOOL"""

    decision = call_llm(tool_prompt)

    if decision.startswith("TOOL:"):
        _, tool_name, params = decision.split(":", 2)
        tool_result = tools[tool_name](params)
        # 将工具结果交给 LLM 处理
        return call_llm(f"基于以下信息完成任务 '{task}':\n{tool_result}")
    else:
        return call_llm(f"完成任务: {task}\n上下文: {context}")

import json
from pathlib import Path

STATE_FILE = "babyagi_state.json"

def save_state(task_list, completed, objective):
    state = {
        "objective": objective,
        "task_list": list(task_list),
        "completed": completed,
        "timestamp": datetime.now().isoformat()
    }
    Path(STATE_FILE).write_text(json.dumps(state, ensure_ascii=False, indent=2))

def load_state():
    if Path(STATE_FILE).exists():
        state = json.loads(Path(STATE_FILE).read_text())
        return (
            deque(state["task_list"]),
            state["completed"],
            state["objective"]
        )
    return None

import chromadb

chroma = chromadb.Client()
collection = chroma.create_collection("babyagi-memory")

def store_in_memory(task_id, task_name, result):
    collection.add(
        documents=[result],
        metadatas=[{"task": task_name}],
        ids=[f"task-{task_id}"]
    )

def retrieve_relevant_context(query, n_results=3):
    results = collection.query(
        query_texts=[query],
        n_results=n_results
    )
    return "\n".join(results["documents"][0])

import asyncio

async def parallel_execute(tasks, max_concurrent=3):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def execute_one(task):
        async with semaphore:
            return await async_execution_agent(task)

    # 找出可并行的任务（无依赖关系的）
    independent_tasks = find_independent_tasks(tasks)
    results = await asyncio.gather(
        *[execute_one(t) for t in independent_tasks]
    )
    return results