首页
/ AutoGPT Forge:构建自定义 AI Agent 的组件化框架实战指南

AutoGPT Forge:构建自定义 AI Agent 的组件化框架实战指南

2026-09-06 17:20:17作者:翟萌耘Ralph

本篇基于 AutoGPT 仓库中 docs/content/forge/get-started.md 快速上手文档及其指向的真实源码编写。Forge 是 AutoGPT 提供的"开箱即用"Agent 应用模板:它把数据库、文件存储、命令执行、权限校验等样板代码全部封装好,让你把精力集中在"Agent 的大脑"——即推理逻辑与组件扩展上。读完本文,你将能够在本机安装并运行 Forge 服务、理解示例 Agent 的完整生命周期(创建任务 → 执行步骤 → 提出动作 → 执行命令),并知道如何通过组件(Component)机制扩展出属于自己的 Agent。

一、Forge 的定位:先分清它"不是什么"

docs/content/forge/get-started.md 中,官方首先给出了一条醒目的警告:如果你只是想"使用 AutoGPT"(运行经典的自主任务执行器),本文档并不适用,应该去 经典版设置文档。Forge 面向的是另一类开发者——希望以 AutoGPT 的代码为底料,构建属于自己的 Agent 应用的人。

Forge 的三大卖点(原文档概括):

  • 免除样板代码:Fork 仓库即可开始构建,无需从零搭建 Agent 基础设施;
  • 以"大脑"为中心的开发:框架提供 LLM 调用、提示词构造、命令体系等全部工具,开发者 100% 的时间可以花在设计 Agent 的推理逻辑上;
  • 一流的配套工具生态:框架本身采用成熟工具链构建。

从仓库结构看,Forge 位于 classic/forge 目录,其 README 将其定义为"Core autonomous agent framework for building AI agents"(用于构建 AI Agent 的核心自主智能体框架)。

二、五分钟跑起来:安装、配置与运行

以下步骤继承自 classic/forge/README.md 的 Quick Start,并结合入口源码补充了运行细节。所有命令都从 classic/ 目录(forge 目录的父目录)执行

# 1. 安装依赖(一次性操作)
cd classic
poetry install

# 2. 配置环境变量
cp .env.example .env
# 编辑 .env,填入你的 OPENAI_API_KEY

# 3. 启动 Agent 服务
poetry run python -m forge

服务默认运行在 http://localhost:8000

入口做了什么?

启动命令 python -m forge 实际执行的是 classic/forge/forge/main.py。从源码看,它做了三件事:

  1. 读取环境变量 PORT(默认 8000)并配置日志;
  2. load_dotenv() 加载 .env 文件——这就是为什么 API Key 只需写在 .env 中;
  3. 通过 uvicorn.run("forge.app:app", ...) 启动 ASGI 应用,并且开启了热重载reload_includes 配置为 forge 包内所有 .py 文件以及 .env,也就是说开发时修改 Agent 代码或环境变量后服务会自动重启。

classic/forge/forge/app.py 本身只有 13 行,完整揭示了 Forge 服务的组装方式:

from forge.agent.forge_agent import ForgeAgent
from forge.agent_protocol.database.db import AgentDB
from forge.file_storage import FileStorageBackendName, get_storage

database_name = os.getenv("DATABASE_STRING")
workspace = get_storage(FileStorageBackendName.LOCAL, root_path=Path("workspace"))
database = AgentDB(database_name, debug_enabled=False)
agent = ForgeAgent(database=database, workspace=workspace)

app = agent.get_agent_app()

三个关键依赖一目了然:数据库AgentDB,存储任务/步骤/产物元数据,连接串来自 DATABASE_STRING)、工作区FileStorage,默认本地 workspace 目录,用于存放 Agent 产出的文件)、以及把两者注入 ForgeAgent 后调用 get_agent_app() 生成 FastAPI 应用。

环境变量配置(.env

classic/forge/.env.example 与 README 中给出的完整配置说明如下:

# 必需
OPENAI_API_KEY=sk-...

# 可选 LLM 设置
SMART_LLM=gpt-4o                    # 复杂推理使用的模型
FAST_LLM=gpt-4o-mini                # 简单任务使用的模型
EMBEDDING_MODEL=text-embedding-3-small

# 可选搜索服务(未配置时会回退到 DuckDuckGo)
TAVILY_API_KEY=tvly-...
SERPER_API_KEY=...
GOOGLE_API_KEY=...
GOOGLE_CUSTOM_SEARCH_ENGINE_ID=...

# 可选基础设施
LOG_LEVEL=DEBUG                     # DEBUG, INFO, WARNING, ERROR
DATABASE_STRING=sqlite:///agent.db  # Agent Protocol 数据库
PORT=8000                           # 服务端口
FILE_STORAGE_BACKEND=local          # local, s3, or gcs

其中 LOG_LEVELPORT.env.example 中有实际默认值(INFO8000);搜索服务为可选项,.env.example 注释明确说明"未设置时会回退到 DuckDuckGo"。

三、示例 Agent 剖析:ForgeAgent 是唯一的起点

官方文档给出的上手路径非常直接:Fork 或下载 AutoGPT 仓库,查看 classic/forge/agent/forge_agent.py 中的示例 Agent,以此作为你自己的 Agent 的起点。 实际文件路径为 classic/forge/forge/agent/forge_agent.py

3.1 状态初始化:BaseAgentSettings

ForgeAgent 同时继承 ProtocolAgentBaseAgent(见文件第 47 行),二者职责分离:

构造函数中首先声明 BaseAgentSettings,这是每个 Agent 的"身份档案":

state = BaseAgentSettings(
    name="Forge Agent",
    description="The Forge Agent is a generic agent that can solve tasks.",
    agent_id=str(uuid4()),
    ai_profile=AIProfile(
        ai_name="ForgeAgent", ai_role="Generic Agent", ai_goals=["Solve tasks"]
    ),
    task="Solve tasks",
)

BaseAgentSettings 定义在 base.py,除 agent_idai_profile(Agent 的"人格")、directives(指令指引)与 task 外,还内嵌了 BaseAgentConfiguration 运行配置。其中几个值得注意的参数(base.py):

参数 默认值 含义
big_brain True True 时用 smart_llm 思考,为 False 时用 fast_llm(混合模式)
cycle_budget 1 Agent 允许无人监督运行的周期数;None 表示无限,0 表示停止,1 表示每步都需要用户批准
send_token_limit None 提示词构造的 token 上限,默认取 LLM max_tokens 的 75%
allow_fs_access False 是否允许文件系统访问

3.2 组件装配:BaseAgent 默认不带任何组件

ForgeAgent.__init__ 中的注释强调:BaseAgent 默认不添加任何组件,示例 Agent 手动装配了以下组件:

# 系统组件:提供 "finish" 命令并注入部分提示词信息
self.system = SystemComponent()

# Todo 组件:多步工作的任务管理
# 注意:ForgeAgent 没有 LLM provider,todo_decompose 不可用;
# 完整功能请使用 original_autogpt 中带有 LLM 访问权限的 Agent
self.todo = TodoComponent()

# 实用工具组件
self.archive_handler = ArchiveHandlerComponent(workspace)
self.clipboard = ClipboardComponent()
self.data_processor = DataProcessorComponent()
self.http_client = HTTPClientComponent()
self.math_utils = MathUtilsComponent()
self.text_utils = TextUtilsComponent()

内置组件完整列表位于 classic/forge/forge/components/,目录下还包括 file_managercode_executorwebimage_gengit_operationsuser_interactionwatchdogcontextaction_history 等组件包,你可以按需引入。

3.3 核心循环:create_task → execute_step → propose_action → execute

Agent Protocol 是 Forge 的核心:先创建任务(task),再为该任务执行步骤(step)。源码中的关键方法:

  • create_task(task_request)forge_agent.py):被协议调用以创建任务。示例中它仅是对 super().create_task() 的一次"钩子"扩展——添加了一条自定义日志。注释明确说"你可以在这里做任何你想做的事",这是定制入口之一。
  • execute_step(task_id, step_request)forge_agent.py):每个步骤的标准实现为三步——
step = await self.db.create_step(task_id=task_id, input=step_request, is_last=False)
proposal = await self.propose_action()   # 1. 让 Agent "思考"出下一步动作
output = await self.execute(proposal)    # 2. 执行该动作
if isinstance(output, ActionSuccessResult):
    step.output = str(output.outputs)
elif isinstance(output, ActionErrorResult):
    step.output = output.reason
return step

任务与步骤请求体都包含 input 字符串(基准测试中即"要求 Agent 解决的任务")和一个任意字典 additional_input;需要时可用 task = await self.db.get_task(task_id) 取回完整任务。所有工作可以放在单个步骤中,也可以拆分为多步,并在步骤输出中请求继续,由用户决定是否让 Agent 继续。

  • propose_action()forge_agent.py):这是最需要你替换的方法。它会先执行三条组件流水线收集 directives(resources / constraints / best_practices)、commandsmessages,组装出 ChatPrompt(消息 + 由命令生成的 function 规格),然后调用 LLM 并解析结果。当前示例的实现是一个占位桩,直接返回"finish(reason='Unimplemented logic')",源码注释写着 "THIS NEEDS TO BE REPLACED WITH YOUR LLM CALL/LOGIC",并指向 original_autogpt 中的 complete_and_parse 作为完整示例。
  • execute(proposal)forge_agent.py):执行逻辑从 run_pipeline(CommandProvider.get_commands) 重新拉取全部命令,倒序匹配 tool.name,找到对应 Command 后同步或异步执行;AgentTerminated 视为成功终止,AgentException 转换为 ActionErrorResult,最后统一触发 AfterExecute.after_execute 流水线,供组件在每次执行后做收尾(如记录历史、监控等)。
  • do_not_execute(denied_proposal, user_feedback):动作被用户拒绝时的处理路径,返回 ActionErrorResult(reason="Action denied") 并同样触发 AfterExecute 流水线。

3.4 组件流水线:run_pipeline 的底层机制

上面反复出现的 self.run_pipeline(...) 是 Forge 组件体系的执行引擎,实现在 base.py

  1. 遍历 self.components跳过未实现该协议(protocol)的组件enabled=False 的组件;
  2. 依次调用各组件上同名方法,收集返回值(如 DirectiveProvider.get_constraints 收集所有组件产出的约束字符串);
  3. 具备两级重试ComponentEndpointError 在同一组件上重试;EndpointPipelineError 则回滚到原始参数后整条流水线重来,重试上限均为 retry_limit=3
  4. 全程写入 self.trace(带彩色标记的成功/失败记录),执行结束后可用 logger.debug("\n".join(self.trace)) 输出调试轨迹。

组件顺序方面,AgentMeta 元类在 Agent 实例化后自动调用 _collect_components():它扫描实例上所有 AgentComponent 属性,若组件声明了 _run_after 依赖,则用拓扑排序保证执行顺序;若发现组件挂在实例上但漏加进 components 列表,会发出警告。这就是"给 Agent 挂属性即可生效"的魔法所在。

四、SystemComponent:一个最小组件长什么样

forge/components/system/system.py 是理解组件协议的最佳样本——SystemComponent(DirectiveProvider, MessageProvider, CommandProvider) 同时实现了三个协议:

  • get_constraints():产出行为约束,例如"只能使用下面列出的命令"、"无法主动启动后台任务或 Webhook"、"不能修改测试文件来让测试通过"、"永不泄露/记录/提交密钥"等;
  • get_resources()get_best_practices():分别产出 Agent 可用的"资源"描述与最佳实践(如"修改前先读文件"、"独立操作尽量并行"、"每次命令都有成本,尽量用最少步骤完成任务");
  • get_messages():注入一条包含当前时间日期的用户消息;
  • get_commands():产出唯一的 finish 命令。

命令通过 @command 装饰器定义,参数由 JSONSchema 描述(finishreason 必填、suggested_next_task 选填),执行时直接抛出 AgentFinished 异常来终止 Agent 循环。组件概念(Component / Protocol / Command / Pipeline)的完整文档见 docs/content/forge/components/introduction.md,其中还特别说明:旧版 plugins 已不再支持,components 是取代它们的新体系propose_actionexecute 就是默认 Agent 中两条核心流水线。

五、工作区与权限:Agent 的"活动范围"

Forge 的 Agent 并不是在任意文件系统上横冲直撞。classic/forge/README.md 定义了工作区结构:

{workspace}/
├── .autogpt/
│   ├── autogpt.yaml              # 工作区级权限
│   ├── ap_server.db              # Agent Protocol 数据库
│   └── agents/
│       └── AutoGPT-{agent_id}/
│           ├── state.json        # Agent 状态
│           ├── permissions.yaml  # Agent 级权限
│           └── workspace/        # Agent 的工作目录

权限采用 allow / deny 两级清单,模式语法为 command_name(glob_pattern)

# .autogpt/autogpt.yaml(工作区默认)
allow:
  - read_file({workspace}/**)
  - write_to_file({workspace}/**)
  - list_folder({workspace}/**)
  - web_search(*)
deny:
  - read_file(**.env)
  - read_file(**.key)
  - execute_shell(rm -rf:*)
  - execute_shell(sudo:*)
# .autogpt/agents/{id}/permissions.yaml(Agent 级覆盖)
allow:
  - execute_python(*)
deny:
  - execute_shell(*)

特殊 token 的语义:{workspace} 会被替换为实际工作区路径;** 匹配任意路径(含 /);* 匹配 / 之外的任意字符。权限判定顺序是首个匹配生效:Agent deny → Workspace deny → Agent allow → Workspace allow → 交互式询问用户批准。

六、扩展你的 Agent:官方推荐的姿势

结合示例 Agent 的注释与 组件创建文档,扩展路径可以归纳为:

  1. 优先加组件,而不是改主循环execute_step 的文档字符串明确写着"添加 Agent 逻辑的推荐方式是添加自定义组件"(参考 docs/content/forge/components/creating-components/)。为组件实现相应协议(如 CommandProvider 提供命令、DirectiveProvider 提供提示词内容、AfterExecute 做执行后处理),然后在 ForgeAgent.__init__ 中挂一个属性即可——元类会自动收集并按拓扑序执行。
  2. 替换 propose_action 中的桩实现:接入你自己的 LLM 调用与结果解析,把命令解析出的 AssistantFunctionCall 转成 ActionProposal
  3. create_task / execute_step 钩子里做定制:这是 Agent Protocol 暴露给你的扩展点,适合打日志、加前置校验或改变步骤拆分策略。
  4. 用权限清单约束行为:通过工作区/Agent 两级 permissions.yaml 控制命令与文件访问。
  5. 完整功能的参考实现:若需要 Todo 分解等依赖 LLM 的能力,forge_agent.py 的注释建议使用 original_autogpt 中带有 LLM 访问权限的 Agent 作为对照。

关于教程:get-started.md 原样保留了一份 Medium 教程系列的指引,但文档自身标注该系列已过期(out of date),且 forge_agent.py 源码中也把同样的教程链接标记为 "Outdated tutorial"。因此以当前仓库源码为准是最可靠的学习路径。

七、小结

  • Forge 面向"构建自己的 Agent 应用"的开发者,与"直接使用 AutoGPT"是两条不同的路线,入手前先看 docs/content/forge/get-started.md 开头的警告;
  • 三步启动:poetry install → 配置 .envOPENAI_API_KEY 为必需)→ poetry run python -m forge,服务位于 http://localhost:8000 并支持热重载;
  • 一切定制围绕 ForgeAgent 展开:BaseAgentSettings 定义身份与运行参数,组件提供命令/指令/消息,create_task/execute_step 是协议钩子,propose_action 是你必须替换的 LLM 大脑;
  • 工作区 + 两级权限清单(deny 优先、首个匹配生效)划定了 Agent 的行动边界。

参考文件:docs/content/forge/get-started.md · classic/forge/README.md · classic/forge/forge/agent/forge_agent.py · classic/forge/forge/agent/base.py · classic/forge/forge/components/system/system.py · docs/content/forge/components/introduction.md

登录后查看全文
热门项目推荐
相关项目推荐