从零构建单循环智能体：基于AutoGPT/BabyAGI的提示工程实战指南

1次阅读

共计 3879 个字符，预计需要花费 10 分钟才能阅读完成。

单循环智能体是 2023 年兴起的新型 AI 架构，其核心特点是 通过单一决策循环 实现自主任务处理。与传统 AI 系统相比：

AutoGPT/BabyAGI 通过持续自我提示（self-prompting）驱动任务执行
每次循环包含完整的 ” 思考 - 行动 - 观察 ” 流程
依赖 LLM 而非预编程规则进行动态决策

典型工作流程如下：

接收初始目标
分解生成子任务
执行并收集结果
根据结果调整策略
循环直到任务完成

graph TD
    A[目标输入] --> B(任务分解器)
    B --> C[任务队列]
    C --> D(执行器)
    D --> E{结果评估}
    E -->| 继续 |B
    E -->| 完成 |F[输出结果]
    D --> G[(记忆模块)]
    G --> B

实现方式推荐：

短期记忆：Python 双端队列（collections.deque）
长期记忆：JSON 文件或轻量级 SQLite

from collections import deque

class Memory:
    def __init__(self, max_short_term=10):
        self.short_term = deque(maxlen=max_short_term)
        self.long_term = []

    def add(self, item: str, is_important=False):
        self.short_term.append(item)
        if is_important:
            self.long_term.append(item)

关键提示词设计示例：

多步思考模板

请将以下目标分解为可执行步骤：目标：{user_input}
要求：- 每个步骤应包含明确动作
- 最多分解为 5 个步骤
- 使用数字编号

优先级排序模板

请评估以下任务的紧急程度（1-5）：{task_list}
输出格式：1. 任务内容 - 优先级
...

结果验证模板

根据当前结果判断是否继续：已完成：{completed_tasks}
当前结果：{latest_output}
剩余目标：{original_goal}
请回答：继续 / 终止

使用 LangChain 的基础框架：

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

class Executor:
    def __init__(self, api_key):
        self.llm = OpenAI(temperature=0.7, api_key=api_key)

    def run_task(self, prompt_template, **kwargs):
        try:
            prompt = PromptTemplate(
                template=prompt_template,
                input_variables=list(kwargs.keys())
            )
            return self.llm(prompt.format(**kwargs))
        except Exception as e:
            print(f"执行失败: {str(e)}")
            return None

import logging
from typing import List, Dict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SingleLoopAgent:
    def __init__(self, openai_key):
        self.memory = Memory()
        self.executor = Executor(openai_key)

    def process_goal(self, goal: str) -> List[Dict]:
        """主处理循环"""
        tasks = self._breakdown_task(goal)
        results = []

        while tasks:
            current = tasks.pop(0)
            logger.info(f"正在处理: {current}")

            output = self.executor.run_task("执行任务: {task}\n 可用上下文: {context}",
                task=current,
                context=list(self.memory.short_term)[-3:]
            )

            if not output:
                logger.warning("任务执行失败")
                continue

            results.append({"task": current, "result": output})
            self.memory.add(output)

            # 动态调整任务队列
            if "下一步" in output:
                new_tasks = self._breakdown_task(output.split("下一步:")[1])
                tasks.extend(new_tasks)

        return results

    def _breakdown_task(self, goal: str) -> List[str]:
        """任务分解实现"""
        prompt = """请将目标分解为最多 5 个具体步骤:\n 目标:{goal}\n 步骤:\n1."""
        response = self.executor.run_task(prompt, goal=goal)
        return [x.strip() for x in response.split("\n") if x.strip()]

滑动窗口法：保持最近 3 - 5 条关键记忆
摘要压缩：对历史记录生成简洁摘要

def summarize_context(messages: List[str]) -> str:
    """生成上下文摘要"""
    if len(messages) <= 3:
        return "\n".join(messages)

    return self.executor.run_task("请用 1 句话总结以下内容:\n{content}",
        content="\n".join(messages)
    )

设置每分钟调用上限
对简单任务使用 gpt-3.5-turbo
实现本地缓存机制

from datetime import datetime, timedelta

class APIBudget:
    def __init__(self, max_calls=10, period_minutes=1):
        self.calls = []
        self.max_calls = max_calls
        self.period = timedelta(minutes=period_minutes)

    def check_quota(self) -> bool:
        now = datetime.now()
        self.calls = [t for t in self.calls if now - t < self.period]
        return len(self.calls) < self.max_calls

常见场景及解决方案：

重复任务

解决方案：检查记忆模块中的相似记录

def is_repetitive(task: str, memory: Memory) -> bool:
    return any(task in m for m in memory.short_term)

模糊指令

解决方案：强制要求明确退出条件

你的任务必须包含明确的完成标准:
- 当 X 发生时停止
- 最多执行 N 次

无限分解

解决方案：设置最大递归深度

MAX_DEPTH = 3

def breakdown_task(self, goal, current_depth=0):
    if current_depth >= MAX_DEPTH:
        return ["达到最大分解深度"]
    # ... 原有逻辑

增强工具调用
集成 Python REPL 执行环境
添加网络搜索 API 调用

多智能体协作

class MultiAgentSystem:
    def __init__(self, n_agents=3):
        self.agents = [SingleLoopAgent() for _ in range(n_agents)]

    def assign_task(self, task):
        return min((agent for agent in self.agents),
            key=lambda a: len(a.memory.short_term)
        ).process_goal(task)

向量数据库集成
使用 Pinecone/Weaviate 存储长期记忆
实现语义相似度检索

# 示例：Weaviate 集成
import weaviate

class VectorMemory:
    def __init__(self, url):
        self.client = weaviate.Client(url)

    def search(self, query: str, limit=3) -> List[str]:
        res = self.client.query.get("Memory", ["content"])\
            .with_near_text(query)\
            .with_limit(limit)\
            .do()
        return [r["content"] for r in res["data"]["Get"]["Memory"]]

经过实际项目验证，这种架构在自动化数据处理场景下展现出了惊人效率。曾用其实现了一个自动数据清洗工具，相比传统脚本开发时间缩短 70%。关键收获：