一、开场：老板说”给我做一个AutoGPT”

大家好，我是老金。

2023年，AutoGPT刚火的时候，老板找到我：

“老金，最近有个AutoGPT很火，你看看能不能给我们做一个？”

我当时心里想：

“AutoGPT？那不就是让AI自己给自己下任务，然后循环执行吗？原理不复杂…”

结果，真正动手做的时候才发现：

• 循环终止条件怎么定？
• 任务失败了怎么办？
• 怎么避免死循环？
• 上下文超限了怎么处理？
• 成本怎么控制？

踩了3个星期的坑，终于做出了一个能用的版本。

后来，这个项目被客户验收了，还拿到了尾款。

今天这篇文章，我就把AutoGPT类应用的核心实现思路，从原理到代码，全部讲透。

内容包括：

1. AutoGPT的原理是什么？
2. 核心架构设计
3. 循环终止条件设计
4. 任务执行引擎实现
5. 记忆系统实现
6. 工具调用机制
7. 成本控制策略
8. 完整代码示例
9. 避坑指南

看完这篇，你将能独立实现一个AutoGPT类应用。

二、AutoGPT的原理：从”AI助手”到”AI员工”

2.1 传统AI助手的模式

用户 → AI → 回答

例子： 用户：”帮我写一封道歉信” AI： “好的，这是道歉信的内容…”

传统AI助手是单向的：用户输入，AI输出，结束。

2.2 AutoGPT的模式

用户 → AI思考 → AI规划 → AI执行 → AI反思 → AI决策 → 继续或结束

例子： 用户： “帮我分析这家公司，给出投资建议” AI思考：”需要收集公司财报、行业数据、竞争对手信息…” AI规划：”步骤1：抓取官网；步骤2：查找财报；步骤3：搜索新闻…” AI执行：”执行步骤1…” AI反思：”步骤1完成，但信息不够，需要补充…” AI决策：”继续执行步骤2″ …（循环） 最终： “根据分析，给出投资建议…”

AutoGPT是自主的：AI不只是回答问题，而是规划、执行、反思的循环。

2.3 核心公式

AutoGPT = LLM + 规划能力 + 执行能力 + 记忆能力 + 终止条件

拆解：

1. 规划能力：让AI把大任务拆成小步骤
2. 执行能力：让AI能真正”做事”（调用工具、访问网络等）
3. 记忆能力：让AI记住之前的思考和执行结果
4. 终止条件：让AI知道什么时候该停下来

三、核心架构设计

3.1 架构图

┌─────────────────────────────────────────────────────────┐
│                        用户                              │
│                         ↓                                │
│                   ┌──────────┐                          │
│                   │   输入    │                          │
│                   └────┬─────┘                          │
│                        ↓                                 │
│    ┌──────────────────────────────────────────┐        │
│    │            AutoGPT Agent                  │        │
│    │  ┌─────────┐  ┌─────────┐  ┌─────────┐  │        │
│    │  │ 规划器  │→ │ 执行器  │→ │ 反思器  │  │        │
│    │  │(Planner)│  │(Executor)│ │(Reflector)│  │        │
│    │  └────┬────┘  └────┬────┘  └────┬────┘  │        │
│    │       └────────────┼────────────┘        │        │
│    │                    ↓                      │        │
│    │            ┌────────────┐               │        │
│    │            │   记忆系统  │               │        │
│    │            │ (Memory)   │               │        │
│    │            └────────────┘               │        │
│    └──────────────────────────────────────────┘        │
│                         ↓                               │
│                   ┌──────────┐                          │
│                   │   输出    │                          │
│                   └──────────┘                          │
└─────────────────────────────────────────────────────────┘

3.2 核心组件

组件	职责	输入	输出
规划器	分解任务，制定执行计划	用户目标、当前状态	待执行任务列表
执行器	调用工具，执行具体任务	任务描述	执行结果
反思器	评估结果，决定下一步	执行结果、历史	决策（继续/终止）
记忆系统	存储历史、上下文	新的事件	检索结果

四、循环终止条件设计

这是AutoGPT的核心难点。

4.1 终止条件类型

• 硬限制：步数限制、时间限制、成本限制
• 软限制：目标达成、无进展检测、重复检测
• 主动终止：AI自己判断”任务已完成”

4.2 终止条件实现

class TerminationConditions:
    def __init__(self, config):
        self.max_steps = config.get("max_steps", 50)
        self.max_cost = config.get("max_cost", 10.0)  # 美元
        self.max_time = config.get("max_time", 3600)  # 秒
        self.no_progress_limit = config.get("no_progress_limit", 5)

    # 状态
    self.current_step = 0
    self.total_cost = 0
    self.start_time = None
    self.last_progress_step = 0

def should_terminate(self, context, result):
    """判断是否应该终止"""
    self.current_step += 1

    # 1. 硬限制检查
    if self.current_step > self.max_steps:
        return True, "达到最大步数限制"

    if self.total_cost > self.max_cost:
        return True, "达到成本上限"

    if self.start_time and (time.time() - self.start_time) > self.max_time:
        return True, "达到时间限制"

    # 2. 无进展检测
    if result.get("progress_made", False):
        self.last_progress_step = self.current_step
    elif self.current_step - self.last_progress_step > self.no_progress_limit:
        return True, "连续{}步无进展".format(self.no_progress_limit)

    # 3. 重复检测
    if self._is_repeating(context):
        return True, "检测到重复执行"

    # 4. 目标达成检测
    if result.get("goal_achieved", False):
        return True, "目标已达成"

    return False, "继续执行"

def _is_repeating(self, context, threshold=3):
    """检测是否在重复执行相似的任务"""
    recent_tasks = context.get_recent_tasks(threshold + 1)
    if len(recent_tasks)  0.8:
            return True
    return False

def _task_similarity(self, task1, task2):
    """计算两个任务的相似度"""
    # 简化版本：比较任务描述
    return difflib.SequenceMatcher(
        None, task1["description"], task2["description"]
    ).ratio()

五、任务执行引擎实现

5.1 工具定义

from abc import ABC, abstractmethod
from typing import Any, Dict, List

class Tool(ABC): “””工具基类”””

@property
@abstractmethod
def name(self) -> str:
    """工具名称"""
    pass

@property
@abstractmethod
def description(self) -> str:
    """工具描述（供LLM理解如何使用）"""
    pass

@property
def parameters(self) -> List[Dict]:
    """参数定义（JSON Schema格式）"""
    return []

@abstractmethod
def execute(self, **kwargs) -> Dict[str, Any]:
    """执行工具"""
    pass

预定义工具示例

class WebSearchTool(Tool):
name = “web_search”
description = “搜索网络信息。适用于查找最新新闻、技术文档等。”

@property
def parameters(self):
    return [{
        "name": "query",
        "type": "string",
        "description": "搜索关键词",
        "required": True
    }, {
        "name": "num_results",
        "type": "integer",
        "description": "返回结果数量",
        "default": 5
    }]

def execute(self, query: str, num_results: int = 5) -> Dict:
    # 实际实现调用搜索引擎API
    results = search_engine.search(query, num_results)
    return {
        "success": True,
        "results": results,
        "summary": f"找到{len(results)}条相关结果"
    }

class WebScrapeTool(Tool):
name = “web_scrape”
description = “抓取网页内容。适用于提取特定页面的详细信息。”

@property
def parameters(self):
    return [{
        "name": "url",
        "type": "string",
        "description": "网页URL",
        "required": True
    }]

def execute(self, url: str) -> Dict:
    content = fetch_url(url)
    return {
        "success": True,
        "content": content,
        "url": url
    }

5.2 执行器实现

class Executor:
    """任务执行器"""

def __init__(self, tools: List[Tool], llm_client):
    self.tools = {tool.name: tool for tool in tools}
    self.llm = llm_client

def execute_task(self, task: Dict, context: "Context") -> Dict:
    """执行单个任务"""
    tool_calls = self._determine_tool_calls(task, context)

    results = []
    for call in tool_calls:
        tool_name = call["name"]
        params = call["parameters"]

        if tool_name not in self.tools:
            results.append({
                "tool": tool_name,
                "success": False,
                "error": f"Unknown tool: {tool_name}"
            })
            continue

        try:
            tool = self.tools[tool_name]
            result = tool.execute(**params)
            results.append({
                "tool": tool_name,
                "success": True,
                "result": result
            })
        except Exception as e:
            results.append({
                "tool": tool_name,
                "success": False,
                "error": str(e)
            })

    return {
        "task": task,
        "results": results,
        "progress_made": any(r["success"] for r in results)
    }

def _determine_tool_calls(self, task: Dict, context: "Context") -> List[Dict]:
    """让LLM决定调用哪些工具"""

    prompt = f"""任务：{task['description']}

可用工具：
{self._format_tools_description()}

当前上下文：
{context.get_summary()}

请决定需要调用哪些工具来完成这个任务。
返回JSON格式的工具调用列表。
“””

    response = self.llm.chat(prompt)
    # 解析LLM响应，提取工具调用
    tool_calls = self._parse_tool_calls(response)
    return tool_calls

def _format_tools_description(self) -> str:
    lines = []
    for name, tool in self.tools.items():
        params_str = ", ".join(
            f"{p['name']}: {p['type']}" + 
            (f" (可选)" if not p.get("required") else "") 
            for p in tool.parameters
        )
        lines.append(f"- {name}({params_str}): {tool.description}")
    return "n".join(lines)

def _parse_tool_calls(self, response: str) -> List[Dict]:
    """解析LLM响应中的工具调用"""
    import json
    import re

    # 尝试提取JSON
    match = re.search(r'[.*]', response, re.DOTALL)
    if match:
        try:
            return json.loads(match.group())
        except:
            pass

    # 降级处理：简单解析
    return []

六、记忆系统实现

6.1 记忆类型

class Memory:
    """记忆系统"""

def __init__(self):
    # 感觉记忆：原始输入（快速遗忘）
    self.sensory_memory = []

    # 工作记忆：当前会话的关键信息
    self.working_memory = []

    # 长期记忆：持久化存储
    self.long_term_memory = VectorStore()

def add(self, event: Dict):
    """添加新记忆"""
    # 添加到感觉记忆
    self.sensory_memory.append(event)

    # 定期提取到工作记忆
    if len(self.sensory_memory) >= 10:
        self._consolidate()

def _consolidate(self):
    """记忆整合：将感觉记忆提炼到工作记忆"""
    # 使用LLM从多条记录中提取关键信息
    raw = "n".join(str(e) for e in self.sensory_memory)
    prompt = f"从以下记录中提取关键信息：n{raw}"
    summary = self.llm.chat(prompt)

    self.working_memory.append({
        "type": "summary",
        "content": summary,
        "timestamp": time.time()
    })

    # 清空感觉记忆
    self.sensory_memory = []

    # 重要信息存入长期记忆
    if self._is_important(summary):
        self.long_term_memory.add(summary)

def retrieve(self, query: str, limit: int = 5) -> List[Dict]:
    """检索相关记忆"""
    # 向量检索长期记忆
    relevant_long = self.long_term_memory.search(query, limit // 2)

    # 搜索工作记忆
    relevant_work = [
        m for m in self.working_memory
        if query.lower() in m["content"].lower()
    ][:limit // 2]

    return relevant_long + relevant_work

def get_context(self, query: str) -> str:
    """获取检索到的上下文"""
    memories = self.retrieve(query)
    if not memories:
        return "无相关记忆"

    lines = ["相关记忆："]
    for m in memories:
        lines.append(f"- {m['content']}")
    return "n".join(lines)

6.2 简单内存实现（无需向量数据库）

class SimpleMemory:
    """简化版记忆系统（不需要向量数据库）"""

def __init__(self, max_size: int = 100):
    self.memories = []
    self.max_size = max_size

def add(self, event: Dict):
    """添加记忆"""
    self.memories.append({
        "content": event.get("content", str(event)),
        "timestamp": event.get("timestamp", time.time()),
        "type": event.get("type", "general")
    })

    # 超出容量时删除最旧的
    if len(self.memories) > self.max_size:
        self.memories.pop(0)

def search(self, query: str, limit: int = 5) -> List[Dict]:
    """简单关键词搜索"""
    query_lower = query.lower()
    scored = []

    for m in self.memories:
        content_lower = m["content"].lower()
        # 简单评分：匹配次数
        score = content_lower.count(query_lower)
        # 越新的权重越高
        recency = (time.time() - m["timestamp"]) / 3600  # 小时
        score = score / (1 + recency * 0.1)

        if score > 0:
            scored.append((score, m))

    scored.sort(reverse=True, key=lambda x: x[0])
    return [m for _, m in scored[:limit]]

def get_recent(self, n: int = 10) -> List[Dict]:
    """获取最近的N条记忆"""
    return self.memories[-n:]

七、成本控制策略

7.1 成本监控

class CostController:
    """成本控制器"""

def __init__(self, max_cost: float = 10.0):
    self.max_cost = max_cost
    self.total_cost = 0.0
    self.cost_by_type = defaultdict(float)

def track(self, operation: str, tokens: int, model: str):
    """追踪成本"""
    price_per_1k = {
        "gpt-4": 0.03,      # $0.03/1K tokens
        "gpt-3.5-turbo": 0.002,
        "claude-3": 0.015,
    }

    price = price_per_1k.get(model, 0.01)
    cost = (tokens / 1000) * price

    self.total_cost += cost
    self.cost_by_type[operation] += cost

def check_budget(self) -> bool:
    """检查预算"""
    if self.total_cost > self.max_cost:
        return False
    return True

def get_report(self) -> str:
    """生成成本报告"""
    lines = [
        f"总成本: ${self.total_cost:.4f}",
        f"预算: ${self.max_cost:.4f}",
        f"使用率: {self.total_cost/self.max_cost*100:.1f}%",
        "",
        "按操作分类："
    ]
    for op, cost in sorted(self.cost_by_type.items(), key=lambda x: -x[1]):
        lines.append(f"  - {op}: ${cost:.4f}")

    return "n".join(lines)

7.2 成本优化策略

• 任务分类路由：简单任务用便宜模型，复杂任务用贵模型
• 缓存复用：相同/相似查询直接返回缓存结果
• 早停机制：任务完成就停，不继续”优化”
• 摘要压缩：长上下文用摘要而非完整记录
• 批量处理：多个相似子任务合并一次调用

# 成本优化的任务路由
class SmartRouter:
    def route(self, task: str) -> str:
        simple_keywords = ["查找", "翻译", "格式化", "检查"]
        complex_keywords = ["分析", "设计", "比较", "评估"]

    if any(kw in task for kw in simple_keywords):
        return "gpt-3.5-turbo"  # 便宜模型
    elif any(kw in task for kw in complex_keywords):
        return "gpt-4"  # 高质量模型
    else:
        return "gpt-4-turbo"  # 平衡之选

八、完整代码示例

import time
from typing import List, Dict, Optional

class AutoGPT: “””AutoGPT核心实现”””

def __init__(
    self,
    llm_client,
    tools: List[Tool],
    max_steps: int = 50,
    max_cost: float = 10.0
):
    self.llm = llm_client
    self.executor = Executor(tools, llm_client)
    self.memory = SimpleMemory()
    self.termination = TerminationConditions({
        "max_steps": max_steps,
        "max_cost": max_cost
    })
    self.termination.start_time = time.time()
    self.cost_controller = CostController(max_cost)

def run(self, goal: str) -> Dict:
    """运行AutoGPT"""
    print(f"🎯 目标: {goal}")

    # 初始化上下文
    context = {
        "goal": goal,
        "history": [],
        "current_task": None
    }

    # 添加目标到记忆
    self.memory.add({
        "type": "goal",
        "content": goal,
        "timestamp": time.time()
    })

    # 主循环
    while True:
        step_num = self.termination.current_step + 1
        print(f"n📍 步骤 {step_num}")

        # 1. 规划下一步
        task = self._plan_next_task(context)
        if not task:
            print("🤔 无法确定下一步任务，结束执行")
            break

        print(f"📋 任务: {task['description']}")
        context["current_task"] = task

        # 2. 执行任务
        result = self.executor.execute_task(task, context)
        print(f"✅ 执行结果: {result.get('summary', '完成')}")

        # 3. 记录到记忆
        self.memory.add({
            "type": "task_result",
            "task": task["description"],
            "result": result,
            "timestamp": time.time()
        })

        # 4. 反思与决策
        should_continue, reason = self.termination.should_terminate(
            context, result
        )

        context["history"].append({
            "step": step_num,
            "task": task,
            "result": result
        })

        if should_continue:
            print(f"n🏁 终止原因: {reason}")
            break

        # 更新上下文
        context["task_result"] = result

    # 生成最终报告
    return self._generate_report(context)

def _plan_next_task(self, context: Dict) -> Optional[Dict]:
    """规划下一步任务"""

    # 获取相关记忆
    relevant_memories = self.memory.get_recent(10)
    memory_context = "n".join(
        f"- {m['content']}" for m in relevant_memories
    )

    prompt = f"""当前目标: {context['goal']}

已完成的任务：
{memory_context if memory_context else ‘暂无’}

请根据当前状态，决定下一步要执行的任务。
如果目标已经达成，返回空。
如果需要继续，返回任务描述。
返回格式：任务描述或”完成”
“””

    response = self.llm.chat(prompt)

    if "完成" in response or "达成" in response:
        return None

    return {"description": response, "timestamp": time.time()}

def _generate_report(self, context: Dict) -> Dict:
    """生成最终报告"""
    return {
        "goal": context["goal"],
        "total_steps": len(context["history"]),
        "total_cost": self.cost_controller.total_cost,
        "history": context["history"],
        "summary": self.memory.get_recent(5)
    }

使用示例

if name == “main“:

初始化LLM客户端

llm = OpenAIClient(api_key="sk-xxx")

# 加载工具
tools = [
    WebSearchTool(),
    WebScrapeTool(),
    CalculatorTool(),
    FileReadTool(),
]

# 创建AutoGPT
agent = AutoGPT(
    llm_client=llm,
    tools=tools,
    max_steps=20,
    max_cost=5.0
)

# 运行
result = agent.run("分析OpenClaw项目的技术架构和优劣势")

print("n" + "="*50)
print("📊 执行报告")
print(f"总步数: {result['total_steps']}")
print(f"总成本: ${result['total_cost']:.4f}")

九、避坑指南

坑1：死循环

现象：AI不断执行相同任务，永远不停。

解决：

• 设置max_steps硬限制
• 实现无进展检测
• 实现重复检测

坑2：上下文爆炸

现象：随着执行，历史记录越来越长，LLM响应越来越慢。

解决：

• 定期压缩/摘要历史
• 使用向量数据库检索关键信息
• 限制上下文长度

坑3：成本失控

现象：跑了1小时，账单$100+。

解决：

• 设置成本上限
• 使用便宜模型处理简单任务
• 实现缓存机制

坑4：工具调用失败导致崩溃

现象：某个工具调用失败，整个流程崩溃。

解决：

• 每个工具调用都要try-catch
• 工具失败时返回错误信息，让AI决定重试还是跳过
• 实现降级策略

坑5：任务分解不合理

现象：规划出来的任务要么太大，要么太细碎。

解决：

• 在Prompt中明确要求分解粒度
• 给LLM提供分解示例
• 根据执行情况动态调整

十、写在最后：从AutoGPT到真正的AI Agent

这篇文章，我讲了AutoGPT的核心实现。

但我想说：

AutoGPT只是起点，不是终点。

真正的AI Agent，还需要：

• 更强的规划能力（CoT、ToT、GoT）
• 多Agent协作
• 持续学习与适应
• 安全与可控性
• 可解释性

OpenClaw就是在AutoGPT的基础上，加了这些能力。

如果你想深入学习，建议：

1. 先把本文的代码跑通
2. 再去看OpenClaw的源码
3. 最后尝试自己扩展功能

纸上得来终觉浅，绝知此事要躬行。

动手实践，才是最好的学习方式。

完。

---