AutoGPT Platform 后端开发指南：Poetry 工作流、测试约定、Block 开发与 LLM 模型目录实战

本文基于 AutoGPT 仓库中后端模块自带的开发指南（autogpt_platform/backend/CLAUDE.md 指向 autogpt_platform/backend/AGENTS.md）整理而成，完整覆盖 AutoGPT Platform 后端的环境搭建、运行命令、架构分层、代码风格、测试与快照更新流程、LLM 模型目录管理和新增 Block 的实操步骤。读完本文后，你可以独立在本地把 Platform 后端跑起来，按照项目既定的工程规范提交代码，并理解其中缓存安全中间件、媒体文件管线（store_media_file）等关键机制的源码级实现。

一、核心命令速查：一切围绕 Poetry

该指南的第一条硬规则是：只要涉及 Python 包依赖的操作，必须通过 poetry run 执行。仓库在 pyproject.toml 的 [tool.poetry.scripts] 中定义了一整套可执行入口，指南中出现的每条命令都有对应的脚本别名：

# 安装依赖
poetry install

# 运行数据库迁移
poetry run prisma migrate dev

# 启动所有服务（数据库、redis、rabbitmq、clamav）
docker compose up -d

# 整体运行后端
poetry run app

# 运行测试
poetry run test

# 运行单个测试
poetry run pytest path/to/test_file.py::test_function_name

# 运行 Block 测试（验证所有 Block 均能正常工作）
poetry run pytest backend/blocks/test/test_block.py -xvs

# 运行某个具体 Block 的测试（如 GetCurrentTimeBlock）
poetry run pytest 'backend/blocks/test/test_block.py::test_available_blocks[GetCurrentTimeBlock]' -xvs

# 格式化与静态检查（想“一键修好”时优先用 format，它只报无法自动修复的错误）
poetry run format  # Black + isort
poetry run lint    # ruff

这些别名的真实落点可以从 pyproject.toml 直接印证：

• app = "backend.app:main" —— 整体后端的入口；
• executor = "backend.exec:main"、batch-executor、copilot-executor —— 独立的执行器服务进程；
• format = "scripts.linter:format"、lint = "scripts.linter:lint"、test = "scripts.run_tests:test" —— 质量工具封装。

docker compose up -d 拉起的基础设施在 docker-compose.yml 中声明，其服务通过 extends 继承 docker-compose.platform.yml 的定义（如 migrate、redis-0 等）。

快照测试的更新流程

API 层大量使用快照（snapshot）测试。首次编写测试或期望输出变化时，用 --snapshot-update 重新生成快照：

poetry run pytest path/to/test.py --snapshot-update

指南特别强调：提交前必须审查快照变更，用 git diff 确认变化符合预期。仓库中 autogpt_platform/backend/snapshots/ 目录存放了大量真实快照文件（如 admin_add_credits_success、log_metric_ok 等），对应 pyproject.toml 中引入的 pytest-snapshot 依赖。更完整的测试细节可参考 TESTING.md。

二、架构分层：FastAPI + Prisma + RabbitMQ + 独立执行器

指南给出的架构要点如下，每一条都能在仓库中找到对应证据：

• API 层：FastAPI，提供 REST 与 WebSocket 端点。依赖清单中 fastapi = "^0.128.6"、uvicorn、websockets 与之对应；
• 数据库：PostgreSQL + Prisma ORM，并包含 pgvector 用于向量嵌入。pyproject.toml 中同时声明了 prisma = "^0.15.0"、psycopg2-binary，而 schema.prisma 就是 Prisma 的模型定义文件；
• 队列系统：RabbitMQ 处理异步任务，对应依赖 aio-pika = "^9.5.5"；
• 执行引擎：独立的 executor 服务进程负责执行 agent 工作流。从源码结构看，这由 backend/exec.py、backend/batch_executor.py 以及 backend/copilot/executor/ 等多个独立进程入口组成，各自在 poetry.scripts 中有专属别名；
• 认证：基于 JWT，并与 Supabase 集成。JWT/认证相关的可复用实现位于 autogpt_libs 包（jwt_utils.py、service.py、dependencies.py 等）；
• 安全：缓存保护中间件，防止敏感数据被浏览器/代理缓存（详见第六节）。

三、代码风格约定：从导入规则到日志插值

指南列出了一组强制性编码规范，这里完整继承并结合仓库说明其影响：

• 只允许顶层导入 —— 禁止局部/内部导入；仅当加载重型可选依赖（如 openpyxl）时允许懒加载导入。一个真实例子：file.py 中 WorkspaceManager 就在函数内导入，注释明确写着 “Import here to avoid circular import (file.py → workspace.py → data → blocks → file.py)”，属于为避免循环引用而允许的例外；

• 绝对导入优先 —— 跨包导入用 from backend.module import ...；同包内兄弟模块允许单点相对导入（from .sibling import ...，如 blocks 包内部）；避免双点相对导入（from ..parent import ...），应改用绝对路径；

• 禁止鸭子类型 —— 不用 hasattr/getattr/isinstance 做类型分派，改用类型化接口/联合类型/Protocol；

• 结构化数据一律用 Pydantic 模型，而非 dataclass/namedtuple/dict。依赖中为 pydantic = "^2.12.5"（含 email 校验）；

• 禁止 linter 抑制注释 —— 不写 # type: ignore、# noqa、# pyright: ignore，遇到就修类型/代码本身；

• 用列表推导式替代手动循环 append；

• 卫语句优先（early return），避免深层嵌套；

• 日志插值的区分 —— debug 语句用 %s 延迟插值（避免不必要的字符串构造开销），其他级别用 f-string 提升可读性：

  logger.debug("Processing %s items", count)
  logger.info(f"Processing {count} items")
  

• 错误路径脱敏 —— 错误信息中对路径使用 os.path.basename()，避免泄露目录结构；

• TOCTOU 意识 —— 文件访问与积分扣费等场景避免“先检查再操作”（check-then-act）模式，否则存在竞态风险；

• Security() vs Depends() —— 认证依赖应使用 Security()，这样 OpenAPI 文档才能生成正确的 security spec；

• Redis 管线 —— 多步操作使用 transaction=True 保证原子性；

• max(0, value) 保护 —— 对计算结果中不应为负的值兜底；

• SSE 协议约定 —— data: 行承载前端解析的事件（必须与前端 Zod schema 匹配），: comment 行用于心跳/状态；

• 文件长度控制在约 300 行以内 —— 超出就按职责拆分（抽 helper、models 或子模块到新文件），禁止往长文件里持续追加；

• 函数长度控制在约 40 行以内 —— 变长就抽命名 helper；长函数是“关注点混杂”的信号而非复杂度的标志；

• 自上而下排列 —— 先定义主函数/公开类，其使用的 helper 放在下方，让读者先看到高层逻辑再看实现细节。

四、测试方法论：pytest + 快照 + TDD

指南对测试的约定：

• 使用 pytest，并对 API 响应做快照测试；
• 测试文件与源码同目录共置，命名为 *_test.py（可在 backend/ 目录中大量验证，如 catalog_test.py、retire_test.py）；
• 在“使用处”打桩，而非“定义处”（mock at the use site）；重构后需同步更新 mock 目标到新模块路径；
• 异步函数使用 AsyncMock（from unittest.mock import AsyncMock）。

pyproject.toml 中的 pytest 配置也印证了测试基建的细节（pyproject.toml）：

[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"
addopts = "-p no:syrupy"   # 禁用 syrupy，避免与 pytest-snapshot 的 --snapshot-update 冲突
faulthandler_timeout = 300 # 单个测试超过 5 分钟就 dump 全部线程栈，让挂死“自报家门”
markers = [
    "supplementary: tests kept for coverage but superseded by integration tests",
    "integration: end-to-end tests that require a live database (skipped in CI)",
    "slow: tests that take more than a few seconds",
]

faulthandler_timeout = 300 尤其值得注意：注释说明其动机是曾有单元测试触达真实 DatabaseManager RPC 后 socket 读取永不返回，挂死 300 秒后自动打印线程栈定位问题。

TDD：先写失败测试，再修复

指南要求：修 bug 或加功能时，先写测试再写实现，标准流程是：

# 1. 写一个标记 xfail 的失败测试
@pytest.mark.xfail(reason="Bug #1234: widget crashes on empty input")
def test_widget_handles_empty_input():
    result = widget.process("")
    assert result == Widget.EMPTY_RESULT

# 2. 运行它 —— 确认它失败（XFAIL）
# poetry run pytest path/to/test.py::test_widget_handles_empty_input -xvs

# 3. 实现修复

# 4. 移除 xfail，再跑 —— 确认通过
def test_widget_handles_empty_input():
    result = widget.process("")
    assert result == Widget.EMPTY_RESULT

这一流程能同时捕捉回归并证明修复真实生效。指南的结论是：每个 bug 修复都应包含一个“本可以抓住该 bug”的测试。

五、数据库核心模型与环境配置

关键数据模型

核心模型定义在 schema.prisma 中，指南列出五个关键模型：

模型	职责
User	认证与个人资料数据
AgentGraph	带版本控制的工作流定义
AgentGraphExecution	执行历史与结果
AgentNode	工作流中的单个节点
StoreListing	用于共享 agent 的 Marketplace 上架条目

数据库演进由 migrations/ 目录下的 SQL 迁移文件驱动（如 20241212141024_agent_store_v2、20260610000000_seat_assignment_cascade_on_member_delete），本地开发用 poetry run prisma migrate dev 应用。

环境变量加载顺序

• 后端：先读 .env.default（默认值），再读 .env（用户覆盖）。仓库中确实存在 .env.default 文件，.env 由开发者按本地环境自行创建。

六、安全实现：缓存保护中间件的源码级解析

指南指出安全中间件位于 security.py，其行为要点与源码完全一致：

• 默认禁用所有端点的缓存，响应头为 Cache-Control: no-store, no-cache, must-revalidate, private，并附带 Pragma: no-cache 与 Expires: 0；
• 白名单机制：只有显式列入 CACHEABLE_PATHS 的路径才允许被缓存。从 SecurityHeadersMiddleware 的 CACHEABLE_PATHS 集合可以看到完整白名单，涵盖：
  • 静态资源：/static、/_next/static、/assets、/images、/css、/js、/fonts；
  • 公开 API：/api/health、/api/status、/api/blocks（含 /api/v1/* 变体）；
  • 用户私有但可 private 缓存的 workspace 文件预览：/api/workspace/files/*/preview（支持 * 通配符，* 在构造正则时会被替换为 [^/]+）；
  • 只读的 Store/Marketplace 页面：/api/store/agents、/api/store/categories、/api/store/featured；
  • 只读的图模板：/api/graphs/templates；
  • 文档端点：/api/docs、/docs、/swagger、/openapi.json；
  • Favicon、manifest、robots.txt、sitemap.xml；
• 目的：防止认证 token、API key、用户数据等敏感内容被浏览器/代理缓存；
• 扩展方式：要为新端点放行缓存，只需把路径加入该中间件中的 CACHEABLE_PATHS；
• 该中间件同时应用于主 API 服务与外部 API 应用，并额外注入 X-Content-Type-Options: nosniff、X-Frame-Options: DENY 等安全头；对共享执行页面（路径含 /public/shared）还会加上 X-Robots-Tag: noindex, nofollow，防止共享页被搜索引擎收录。

实现上它选择纯 ASGI 中间件（__call__ + send_wrapper）而非 BaseHTTPMiddleware，注释说明理由是为获得更好的性能。

七、常见开发任务

7.1 新增 / 修改 / 退役 LLM 模型

指南描述的流程是 “catalog-as-code”（目录即代码），核心事实如下：

• 模型定义、成本与 AutoPilot 路由集中在 catalog.py；直接编辑该文件并开 PR（仅目录变更的 diff 可走 hotfix/* → master 通道以加快事故响应）；

• 目录是唯一事实来源：元数据与计费字典在 import 时从它派生；

• 一个可被 Block 选择的模型，还需要在 llm_models.py 中加一行 LLMModel 名称（import 时的检查会强制两者配对）；仅供 copilot 使用的模型则只需目录条目；

• 退役模型：提交一个 is_enabled: False 的目录 PR，并运行

  python -m backend.data.llm_registry.retire <slug> --replacement <slug> --yes
  

  该 retire 命令负责迁移现有图节点中的模型引用，默认 dry-run，且可回滚。完整参考见 Managing LLM Models。

配套的 registry_test.py、retire_test.py 等共置测试保证目录派生逻辑有回归保护。

7.2 新增一个 Block

完整流程在 Block SDK Guide 中，覆盖：

• 使用 ProviderBuilder 的 Provider 配置；
• Block schema 定义；
• 认证方式（API key、OAuth、webhook）；
• 测试与验证；
• 文件组织方式。

快速步骤：

1. 在 backend/blocks/ 下创建新文件；
2. 在 _config.py 中用 ProviderBuilder 配置 provider；
3. 继承 Block 基类（其定义见 blocks/_base.py，是一个带输入/输出 schema 泛型的 ABC）；
4. 用 BlockSchema 定义输入/输出 schema；
5. 实现异步 run 方法；
6. 用 uuid.uuid4() 生成唯一 Block ID；
7. 用 poetry run pytest backend/blocks/test/test_block.py 测试。

指南还附了一条产品化提醒：批量新增 Block 时，先分析各 Block 的接口，想象它们在图编辑器里能否顺畅连接——输入输出是否“接得上”。遇到 pushback 或复杂 Block 条件时，查看 docs 中的 new_blocks 指南。

7.3 用 store_media_file() 处理 Block 中的文件

当 Block 需要处理文件（图片、视频、文档）时，应使用 backend/util/file.py 中的 store_media_file()。return_format 参数决定返回值形态：

Format	适用场景	返回值
"for_local_processing"	用本地工具处理（ffmpeg、MoviePy、PIL）	本地文件路径（如 "image.png"）
"for_external_api"	把内容发给外部 API（Replicate、OpenAI）	Data URI（如 "data:image/png;base64,..."）
"for_block_output"	从 Block 返回输出	智能适配：CoPilot 中为 workspace://，图中为 data URI

# 输入：需要用 ffmpeg 在本地处理文件
local_path = await store_media_file(
    file=input_data.video,
    execution_context=execution_context,
    return_format="for_local_processing",
)
# local_path = "video.mp4" - 用 Path/ffmpeg 等处理

# 输入：需要发给 Replicate 等外部 API
image_b64 = await store_media_file(
    file=input_data.image,
    execution_context=execution_context,
    return_format="for_external_api",
)
# image_b64 = "data:image/png;base64,iVBORw0..." - 发给 API

# 输出：从 Block 返回结果
result_url = await store_media_file(
    file=generated_image_url,
    execution_context=execution_context,
    return_format="for_block_output",
)
yield "image_url", result_url
# 在 CoPilot 中: result_url = "workspace://abc123"
# 在图中:      result_url = "data:image/png;base64,..."

要点：

• for_block_output 是唯一会按执行上下文自动适配的格式；
• 除非有特定理由，Block 输出一律用 for_block_output；
• 永远不要硬编码 workspace 判断——让 for_block_output 自己处理。

源码中还能看到若干值得了解的安全与健壮性设计：函数开头强制校验 graph_exec_id 与 user_id；对执行目录设置了 1GB 磁盘用量上限防止 DoS；用 resolve() 解析符号链接并校验本地路径必须落在 {tempdir}/exec_file/{exec_id}/ 之内（防止路径逃逸）；输入既支持 data URI、URL、workspace:// 引用，也支持执行目录内的相对本地路径（file.py）。

7.4 修改 API 的标准步骤

1. 更新 backend/api/features/ 中的路由；
2. 在同目录添加/更新 Pydantic 模型；
3. 在路由文件旁编写测试（同置 *_test.py 约定）；
4. 运行 poetry run test 验证。

八、Workspace 与媒体文件：何时需要深入阅读

指南明确列出四种需要参考 Workspace & Media Architecture 的场景：

• 开发 CoPilot 文件上传/下载功能；
• 构建处理 MediaFileType 输入/输出的 Block；
• 修改 WorkspaceManager 或 store_media_file()；
• 排查文件持久化或病毒扫描问题。

该文档覆盖：WorkspaceManager（带会话作用域的持久化存储）、store_media_file()（媒体归一化管线），以及病毒扫描与持久化的责任边界划分。

总结

这份后端开发指南的价值在于把 AutoGPT Platform 后端的工程纪律固化成可执行的清单：poetry run 统一环境入口、_test.py 同置 + 快照测试 + xfail 式 TDD 的测试闭环、以 schema.prisma 与 migrations/ 为骨架的数据层演进、以 llm_registry/catalog.py 为单一事实来源的模型治理，以及白名单式的缓存安全中间件。文中引用的关键文件——pyproject.toml、security.py、file.py、schema.prisma——均可在当前仓库中直接查阅，作为本文每个结论的验证入口。