Manus 这类 AI 智能体产品成为热点后，最常见的讨论是它的执行能力和商业前景。当“独立运营”这类消息出现时，技术团队的第一反应往往是另一个问题：一个能在演示视频里跑通的 Agent，距离一个能独立接受真实用户流量、持续迭代、出了问题能定位的服务，中间到底还差多少改造？从技术视角看，“一夜回到创业状态”不是重新讲故事，而是回到最基础的任务调度、状态持久化、可观测性和容错设计。这里不讨论商业动向，只以 Manus 引发的关注为切入点，梳理 AI Agent 从原型走向独立服务时最值得优先做的工程化改造。读者可以把自己正在做的智能体项目套进来，对比哪些能力已经具备，哪些还在 Demo 阶段。
1. 从 Manus 独立运营说起：AI Agent 为何需要“工程化重做”
1.1 演示 Demo 和独立服务之间差了什么
演示场景里，人工介入成本很低。一个 Agent 执行到一半卡住，演示者可以重新运行；上下文丢了，可以重新输入；LLM 返回格式不对，可以直接调整提示词再来一次。独立运营之后，这些“人工兜底”全部消失，系统必须自己面对异常。
至少在四个维度上，Demo 和独立服务有明确差异：
- 执行时间：一次 Agent 任务可能包含多轮 LLM 调用和多次工具调用，耗时可能从几秒到几分钟。如果所有请求都通过 HTTP 同步返回，非常容易触发网关超时。
- 状态持久化：Demo 可以把会话放在内存里，进程一重启就丢失。独立服务需要把会话记录、任务状态、工具调用结果落到数据库。
- 容错能力：单次任务中某一步失败，是重试当前步骤、重新生成回复，还是直接标记任务失败，都需要有规则。Demo 阶段通常根本没有这个分支。
- 可观测性：用户反馈“回答不对”时，工程师需要能查到该次任务用了什么模型、调用了哪些工具、每步耗时多少、最终是基于哪些上下文生成的回答。没有日志和追踪，问题只能靠猜。
所以，独立运营的本质不是把 Demo 换个域名部署，而是把“人肉驱动的一次性脚本”改造成“机器可观测、可恢复、可治理的异步任务流”。
1.2 AI Agent 的核心执行循环决定了架构边界
AI Agent 的基本执行循环可以概括为：接收用户输入，构造消息上下文，调用 LLM，判断模型是否要求调用工具，执行工具，把工具结果追加到上下文，继续调用 LLM，直到模型给出最终回答。
这个循环是理解整个架构的关键。它意味着：
- 一次用户请求可能产生多次外部调用（LLM 调用、工具调用）。
- 每次工具调用结果都要回到消息列表，消息列表会持续变长。
- 循环次数不可预知，必须有上限保护。
- 循环中间任何一步失败，都不能简单丢给用户重试。
如果只是写一个 Python 函数，在主进程里循环调用，Demo 没有问题。但独立服务需要处理并发请求、任务优先级、失败重试、进程重启后的恢复。因此，必须把“执行循环”放到一个可以离线运行的任务执行器里，API 层只负责接收请求并返回任务标识。
这也是本文选择 FastAPI + Celery + Redis + PostgreSQL 作为技术骨架的原因。FastAPI 负责轻量 API 层，Celery 负责异步任务，Redis 做消息队列，PostgreSQL 保存最终状态和审计记录。这套组合没有引入过于复杂的系统，但足够支撑一个 AI Agent 从本地脚本升级成可独立运营的服务。
2. 先把架构拆分清楚，再写代码
2.1 单体 Demo 的问题
很多 AI Agent 最开始是单个 Python 文件，伪代码如下：
while True: user_input = input("你：") response = agent.run(user_input) print("Agent：", response)代码看起来简单，但进入 Web 服务后会快速暴露出问题：
- 同一个进程里同时跑多个用户请求时，消息上下文会互相污染。
- 任务执行期间进程重启，未完成的任务直接丢失。
- 没有任务 ID，用户无法查询执行状态，日志也无法关联到具体会话。
- 同步执行长任务时，HTTP 连接被长时间占用，体验和稳定性都很差。
因此，第一个工程决策是：把“执行 Agent”和“接收请求”拆开。接收请求的进程只做两件事：写入任务记录、把任务 ID 返回给用户。真正跑 Agent 循环的后台进程从队列里取任务执行。
2.2 目标架构：API入口、任务队列、执行器和持久化
一个最小但完整的 AI Agent 独立服务，至少包含四层：
| 层级 | 职责 | 技术选型示例 |
|---|---|---|
| API 层 | 接收用户输入，创建任务，返回任务状态 | FastAPI |
| 任务队列 | 承接异步任务，支持失败重试，解耦 API 和 Worker | Celery + Redis |
| 执行器 | 执行 Agent 循环，调用 LLM 和工具，更新任务状态 | Python Worker |
| 持久化层 | 保存会话、任务、工具调用日志，支撑排障和审计 | PostgreSQL |
流程如下：用户 POST 一条消息，API 层在数据库创建任务记录，把任务 ID 返回；同时把任务投递到 Celery 队列。Celery Worker 取出任务后，加载会话历史，运行 Agent 循环，期间每执行一个工具都记录到tool_calls表；执行完成后更新任务状态和输出。用户通过 GET 接口轮询任务状态。
这个架构带来的直接好处是：
- 长任务不再占用 Web 进程。
- 执行进度可以被记录和查询。
- Worker 崩溃后，Celery 可以根据配置重试，任务状态可以从数据库中恢复。
- 后续如果要加多个 Worker，可以直接水平扩展。
2.3 数据模型：会话、任务、工具调用、日志
独立运营阶段，数据库表不是可有可无的装饰，而是排障的基础。下面是最小化的四张表，足够支撑一个单 Agent 服务。
CREATE TABLE conversations ( id UUID PRIMARY KEY, user_id VARCHAR(64), title VARCHAR(255), created_at TIMESTAMP NOT NULL DEFAULT NOW() ); CREATE TABLE tasks ( id UUID PRIMARY KEY, conversation_id UUID REFERENCES conversations(id), status VARCHAR(20) NOT NULL DEFAULT 'pending', input TEXT NOT NULL, output TEXT, error TEXT, celery_task_id VARCHAR(64), created_at TIMESTAMP NOT NULL DEFAULT NOW(), updated_at TIMESTAMP NOT NULL DEFAULT NOW() ); CREATE TABLE tool_calls ( id BIGSERIAL PRIMARY KEY, task_id UUID REFERENCES tasks(id), tool_name VARCHAR(128) NOT NULL, arguments JSONB, result JSONB, duration_ms INTEGER, created_at TIMESTAMP NOT NULL DEFAULT NOW() ); CREATE TABLE agent_logs ( id BIGSERIAL PRIMARY KEY, task_id UUID REFERENCES tasks(id), step INTEGER, event_type VARCHAR(32), content JSONB, created_at TIMESTAMP NOT NULL DEFAULT NOW() );会话表记录用户维度。任务表记录每一次用户输入对应的执行结果。工具调用表保留每次工具的参数和返回结果，这是判断 Agent 行为是否正确的关键证据。日志表按步骤记录 Agent 循环中的事件，例如llm_call、tool_call、final_answer。
注意：不要为了省事只存一个 JSON 字段记录所有内容。独立运营之后，查询“某个任务是否调用过某个工具”会变得非常频繁，单表 JSON 查询性能和维护性都很难满足。
3. 用最小工程骨架跑通一个可独立运行的 Agent
3.1 环境准备和依赖说明
建议使用 Python 3.11 及以上版本。下面的版本组合仅用于说明，实际落地前要先确认与项目其他依赖的兼容性。
| 依赖 | 作用 | 典型版本 |
|---|---|---|
| fastapi | API 层框架 | 0.115.x |
| uvicorn | ASGI 服务器 | 0.30.x |
| celery | 异步任务队列 | 5.4.x |
| redis | 消息队列和缓存 | 5.0.x |
| sqlalchemy | ORM 和数据库访问 | 2.0.x |
| psycopg2-binary | PostgreSQL 驱动 | 2.9.x |
| openai | LLM 客户端 SDK | 1.x |
安装命令：
pip install fastapi uvicorn celery redis sqlalchemy psycopg2-binary openai如果是学习环境，可以直接用 SQLite 替换 PostgreSQL，减少第一步的复杂度。但进入独立运营前，建议尽早切换到 PostgreSQL，因为并发写入和备份恢复能力差别很大。
3.2 项目目录结构
一个可扩展的目录结构如下：
agent_service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置读取 │ ├── database.py # 数据库连接 │ ├── models.py # ORM 模型 │ ├── schemas.py # Pydantic 模型 │ ├── api/ │ │ ├── __init__.py │ │ └── tasks.py # 任务相关 API │ ├── agent/ │ │ ├── __init__.py │ │ ├── loop.py # Agent 执行循环 │ │ └── tools.py # 工具注册和调用 │ └── worker/ │ ├── __init__.py │ └── tasks.py # Celery 任务定义 ├── docker-compose.yml ├── .env.example └── requirements.txt这个结构把 API、Agent 逻辑、Celery 任务分开，是为了避免后续在单个文件里不断加代码导致无法维护。实际项目可以根据团队习惯调整，但职责边界建议保留。
3.3 关键代码：Agent 执行循环
Agent 执行循环是整篇文章的核心。下面代码使用 OpenAI SDK 作为 LLM 客户端，但重点是循环逻辑，不局限于具体模型厂商。
# app/agent/loop.py from typing import Any from openai import OpenAI from app.agent.tools import TOOL_SCHEMAS, execute_tool client = OpenAI() # 实际从配置读取 api_key / base_url def run_agent_loop( messages: list[dict[str, Any]], max_steps: int = 8, ) -> tuple[list[dict[str, Any]], str]: """ 执行 Agent 循环。 返回最终消息列表和最终回答文本。 max_steps 用于防止模型无限循环或连续调用工具过多次。 """ for step in range(max_steps): # 1. 请求 LLM response = client.chat.completions.create( model="gpt-4o-mini", messages=messages, tools=TOOL_SCHEMAS, tool_choice="auto", ) message = response.choices[0].message # 2. 把模型回复追加到消息列表 messages.append(message.model_dump()) # 3. 如果没有工具调用，就认为已经生成最终回答 if not message.tool_calls: return messages, message.content or "" # 4. 逐个执行工具，把工具结果追加到消息列表 for tool_call in message.tool_calls: result = execute_tool( tool_call.function.name, tool_call.function.arguments, ) messages.append( { "role": "tool", "tool_call_id": tool_call.id, "content": result, } ) # 5. 超过最大步数，强制结束并抛出业务异常 raise RuntimeError("agent exceeded max_steps")关键点有三个：
- 每轮循环都将模型回复追加到
messages，保证多轮工具调用之间的上下文连续。 - 工具结果以
role=tool追加，并且必须带上tool_call_id，否则模型无法理解结果对应哪个工具。 max_steps必须设置为一个合理值，例如 6 到 10。没有这个限制，遇到模型无限循环时任务会一直消耗 token。
3.4 关键代码：工具注册和调用
工具注册表的思路是：用一个装饰器把 Python 函数注册成 Agent 可调用的工具，同时自动生成模型需要的 JSON Schema。这样新增一个工具，只需要新增一个函数，不需要改 Agent 循环。
# app/agent/tools.py import json from typing import Any, Callable TOOL_REGISTRY: dict[str, Callable] = {} TOOL_SCHEMAS: list[dict] = [] def tool( name: str, description: str, parameters: dict[str, Any], ): def decorator(func: Callable): TOOL_REGISTRY[name] = func TOOL_SCHEMAS.append( { "type": "function", "function": { "name": name, "description": description, "parameters": parameters, }, } ) return func return decorator @tool( name="get_current_time", description="获取服务器当前时间，返回 ISO 格式字符串。", parameters={ "type": "object", "properties": {}, }, ) def get_current_time() -> str: from datetime import datetime, timezone return datetime.now(timezone.utc).isoformat() @tool( name="calculate", description="执行一个简单的四则运算表达式。", parameters={ "type": "object", "properties": { "expression": { "type": "string", "description": "要计算的数学表达式，例如 (1 + 2) * 3", } }, "required": ["expression"], }, ) def calculate(expression: str) -> str: # 仅用于示例，生产环境要使用限制更严格的安全求值方式 return str(eval(expression)) # noqa: S307 def execute_tool(name: str, arguments_json: str) -> str: if name not in TOOL_REGISTRY: return json.dumps({"error": f"unknown tool: {name}"}) try: arguments = json.loads(arguments_json) except json.JSONDecodeError: return json.dumps({"error": "invalid arguments json"}) try: result = TOOL_REGISTRY[name](**arguments) return json.dumps(result, ensure_ascii=False, default=str) except Exception as exc: # noqa: BLE001 return json.dumps({"error": str(exc)})注意execute_tool返回的是字符串。LLM 的工具结果字段通常只接受字符串，不能直接传 Python 对象。另外，工具内部不能崩溃，要把异常转成 JSON 错误信息返回给模型，让模型有机会调整参数重试。
上面示例中的
eval只用于演示思路，生产环境绝不能直接执行用户提供的表达式。实际项目建议使用ast.literal_eval或专门的表达式解析库，并严格控制输入来源。
3.5 任务队列和异步执行
Celery 的任务定义放在app/worker/tasks.py。任务函数负责从数据库读取会话历史，调用 Agent 循环，并更新任务表状态。
# app/worker/tasks.py import uuid from celery import Celery from app.agent.loop import run_agent_loop from app.database import get_db_session from app.models import Conversation, Task, ToolCall, AgentLog celery_app = Celery("agent_service", broker="redis://redis:6379/0", backend="redis://redis:6379/0") @celery_app.task(bind=True, max_retries=3, default_retry_delay=5) def process_task(self, task_id: str): db = get_db_session() task = db.get(Task, uuid.UUID(task_id)) if task is None: return task.status = "running" db.add(task) db.commit() conversation = db.get(Conversation, task.conversation_id) messages = load_conversation_messages(db, conversation.id) try: final_messages, answer = run_agent_loop(messages) task.output = answer task.status = "succeeded" except Exception as exc: task.error = str(exc) task.status = "failed" raise self.retry(exc=exc) db.add(task) db.commit() db.close()load_conversation_messages需要从数据库恢复历史消息。一种简单做法是：把每一轮用户输入和 Agent 输出都保存为消息记录，执行任务时按时间顺序组装成messages。另一种做法是直接在tasks表里保存消息快照，适合短会话场景。更完整的实现还要在 Agent 循环内部记录agent_logs，这样可以在任务失败时回放每一步。
FastAPI 接口只负责两件事：创建任务，返回任务 ID；查询任务状态。
# app/api/tasks.py import uuid from fastapi import APIRouter, BackgroundTasks from app.database import get_db_session from app.models import Task, Conversation from app.schemas import TaskCreate, TaskRead from app.worker.tasks import process_task router = APIRouter(prefix="/api/tasks") @router.post("", response_model=TaskRead) def create_task(body: TaskCreate): db = get_db_session() conversation = Conversation(id=uuid.uuid4(), user_id=body.user_id or "anonymous") db.add(conversation) db.commit() task = Task( id=uuid.uuid4(), conversation_id=conversation.id, status="pending", input=body.input, ) db.add(task) db.commit() process_task.apply_async(args=[str(task.id)], task_id=str(task.id)) return TaskRead( task_id=task.id, status=task.status, conversation_id=conversation.id, ) @router.get("/{task_id}", response_model=TaskRead) def get_task(task_id: str): db = get_db_session() task = db.get(Task, uuid.UUID(task_id)) if task is None: return {"error": "task not found"} return TaskRead(task_id=task.id, status=task.status, output=task.output, error=task.error)这里有一个容易忽略的点：process_task.apply_async传入的task_id会让 Celery 任务 ID 和业务任务 ID 保持一致，后续在 Celery 日志里看到task_id，可以直接关联到数据库记录。
4. 从本地跑通到部署：配置、容器和启动顺序
4.1 环境变量统一管理
配置项包括数据库连接、Redis 地址、LLM API Key、模型名称、任务频控等。不要把配置硬编码在代码里。
# .env.example DATABASE_URL=postgresql://agent:agent@db:5432/agent REDIS_URL=redis://redis:6379/0 OPENAI_API_KEY=sk-xxxx OPENAI_MODEL=gpt-4o-mini MAX_AGENT_STEPS=8 TASK_TIMEOUT_SECONDS=120在 Python 配置文件中读取：
# app/config.py import os DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./agent.db") REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0") OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "") OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini") MAX_AGENT_STEPS = int(os.getenv("MAX_AGENT_STEPS", "8")) TASK_TIMEOUT_SECONDS = int(os.getenv("TASK_TIMEOUT_SECONDS", "120"))生产环境建议由配置中心或容器编排平台注入环境变量，而不是在服务器上手工修改.env文件。
4.2 Docker Compose 编排
一个最小但完整的编排文件包含四个服务：API、Worker、Redis、PostgreSQL。
# docker-compose.yml version: "3.9" services: db: image: postgres:16 environment: POSTGRES_USER: agent POSTGRES_PASSWORD: agent POSTGRES_DB: agent volumes: - pgdata:/var/lib/postgresql/data healthcheck: test: ["CMD-SHELL", "pg_isready -U agent"] interval: 5s timeout: 5s retries: 10 redis: image: redis:7 healthcheck: test: ["CMD", "redis-cli", "ping"] interval: 5s timeout: 5s retries: 10 api: build: . command: uvicorn app.main:app --host 0.0.0.0 --port 8000 env_file: - .env ports: - "8000:8000" depends_on: db: condition: service_healthy redis: condition: service_healthy worker: build: . command: celery -A app.worker.tasks.celery_app worker --loglevel=info env_file: - .env depends_on: db: condition: service_healthy redis: condition: service_healthy volumes: pgdata:启动前要确认.env里DATABASE_URL和REDIS_URL使用的是容器内服务名db和redis，而不是localhost。本地直连的时候才用 localhost。
4.3 启动与验证
docker compose up --build等待三个服务都变成 healthy 后，可以用 curl 验证：
curl -X POST http://localhost:8000/api/tasks \ -H "Content-Type: application/json" \ -d '{"user_id":"1001","input":"现在几点了？"}'预期返回：
{ "task_id": "9a86b098-...", "status": "pending", "conversation_id": "7e1c6f9a-..." }然后轮询任务状态：
curl http://localhost:8000/api/tasks/9a86b098-...当状态变为succeeded时，output字段就是 Agent 的最终回答。如果状态为failed，error字段会给出错误摘要。完整链路需要查看 Worker 的日志，确认 Celery 确实接收并执行了任务。
5. 独立运营阶段必须处理的工程问题
5.1 可观测性：日志、追踪、指标
Demo 可以没有日志，独立运营不能没有。至少要做到三件事：
- 结构化日志：每个关键节点输出一行 JSON 日志，包含
task_id、session_id、event_type、model、step、duration_ms、token_usage等字段。 - 链路追踪：一次 Agent 任务会触发多次 LLM 调用和工具调用，需要能按
task_id关联所有步骤。 - 指标监控：任务成功率、平均执行时长、单次任务最大步数、LLM 调用失败率、工具调用失败率，至少要用 Prometheus 或云厂商监控系统采集。
推荐在 Agent 循环内部统一埋点。不要在业务代码里手动到处打日志，而要在run_agent_loop中通过回调或装饰器统一记录。例如每次 LLM 调用前后记录时间差，每次工具调用后记录tool_calls表。
5.2 限流、超时和成本控制
独立运营后，成本控制会成为第一优先级。一个没有max_steps限制的 Agent，在极端情况下可能一次性消耗大量 token。即使是单用户场景，也需要限制：
| 控制项 | 推荐设置 | 说明 |
|---|---|---|
| Agent 最大循环步数 | 6-10 | 防止模型无限调用工具 |
| 单次 LLM 请求超时 | 30-60 秒 | 避免个别请求拖垮 Worker |
| 整任务超时 | 120 秒 | 超过后强制失败，由调度器处理 |
| 单用户并发任务数 | 1-3 | 防止单用户刷爆资源 |
| 每日预算 | 按业务评估 | 达到阈值后触发告警或降级 |
限流可以在 FastAPI 层用 Redis 实现，也可以使用网关层限流。成本控制不能只靠事后看账单，要在任务执行时记录 token 消耗，并定期聚合分析。
5.3 数据备份和权限安全
数据库备份是长期运营的基本要求。PostgreSQL 容器至少配置每日备份，并将备份文件存储到独立存储桶。除此之外，还要注意：
- 用户隔离：Agent 服务的会话和任务必须绑定
user_id，查询时强制加过滤条件，避免越权访问。 - API Key 管理：LLM API Key 不要出现在前端代码或客户端请求中，只保存在服务端环境变量里。
- 工具权限：Agent 能调用的工具越少越好。每个工具只开放必要权限，不要给 Agent 一个能执行任意命令的 Shell 工具。
- 敏感信息脱敏：日志、数据库、工具参数中可能包含用户隐私，需要按业务规范脱敏后再留存。
6. 常见问题排查链路
6.1 任务一直 Pending
现象：接口返回pending，几十秒后仍是pending，Worker 日志没有任何输出。
排查顺序：
- 检查 Redis 是否连通：
docker compose exec redis redis-cli ping，期望返回PONG。 - 检查 Worker 是否启动：
docker compose logs worker，看 Celery 是否注册了任务。 - 检查 API 是否成功投递任务：查看 API 日志中是否有
process_task相关输出。 - 检查任务是否被其它 Worker 消费：如果存在多个 Worker，可能消息被其他实例消费，需要确认注册的任务名一致。
常见原因和处理方式：
| 问题现象 | 常见原因 | 检查方式 | 处理建议 |
|---|---|---|---|
| 一直 pending | Redis broker 配置错误 | 检查REDIS_URL是否为容器内地址 | 修正环境变量后重启 |
| 一直 pending | Worker 没有启动 | 查看 worker 容器状态和日志 | 启动 worker |
| 一直 pending | Celery 任务名不一致 | 对比 API 和 Worker 的 import 路径 | 统一任务 import 路径 |
| 一直 pending | 数据库未健康 | 检查 db healthcheck | 等数据库 ready 后再启动 Worker |
6.2 Agent 循环卡住或死循环
现象：任务长时间处于running，日志显示多次 LLM 调用，但始终没有最终回答。
根因一般是模型一直要求调用工具，或者工具返回结果让模型无法收敛。处理方式：
- 确认
max_steps是否生效。如果调用了run_agent_loop但没传max_steps，默认值可能被覆盖。 - 查看
agent_logs中每一步的tool_calls，分析模型是否在重复调用同一个工具。 - 如果工具结果包含过长内容，模型可能反复处理同一段文本。此时需要截断工具结果或使用摘要。
- 对高频工具增加缓存，避免相同参数重复执行。
6.3 LLM 调用报错或超时
现象：任务直接失败，error字段出现openai.AuthenticationError、RateLimitError或TimeoutError。
排查链路：
- 检查
OPENAI_API_KEY是否正确，是否被 EnvFile 注入。 - 检查模型名是否支持工具调用参数，不是所有模型都支持
tools字段。 - 检查请求是否触发了限流，日志中会包含
429状态码和retry-after头。 - 检查网络超时配置，SDK 默认超时可能不适合 Agent 长任务。
生产环境建议在 LLM 调用处统一封装重试和熔断。重试要带上指数退避，避免高峰时加重 LLM 服务压力。
6.4 会话状态丢失
现象：用户连续发送两条消息，第二条消息的 Agent 输出完全不记得第一条内容。
根因通常是messages没有正确还原。检查下面几点：
load_conversation_messages是否按时间顺序加载消息。- 保存历史时，是否把工具调用消息也一起保存。仅保存用户和助手消息，会导致工具调用记录缺失。
- 如果使用 Redis 作为消息缓存，要确认 Redis 是否开启了持久化。Redis 重启后缓存丢失，会话会断。
建议把会话历史持久化到 PostgreSQL，Redis 只作为短期缓存。任务执行前从数据库读取完整消息列表，任务结束后写回。
7. 实践建议和扩展方向
7.1 独立运营前的检查清单
下面是一份可以复制到团队文档里的检查清单，覆盖从功能到运维的常见盲区：
- [ ] 任务是否异步化，API 是否有超时保护。
- [ ] 会话历史是否持久化，重启后能否恢复。
- [ ] Agent 循环是否设置最大步数和单任务超时。
- [ ] 每个 LLM 调用和工具调用是否有日志记录，能否按
task_id串起来。 - [ ] 是否有限流、熔断、预算告警。
- [ ] 数据库是否有备份策略和恢复演练记录。
- [ ] 工具权限是否最小化，敏感输入是否脱敏。
- [ ] API 是否按用户隔离，是否存在越权风险。
- [ ] 是否定义任务失败的重试规则，避免无限重试。
- [ ] 是否记录 token 消耗，能否按用户和任务维度统计成本。
清单里的每一项都可以单独写成一篇排障文章。独立运营阶段，真正让人头疼的往往不是模型能力不足，而是这些工程细节在流量冲击下逐一出问题。
7.2 后续扩展方向
这套骨架跑通后，扩展方向可以按优先级推进：
- 多 Agent 协作：把单一执行循环扩展为多个 Agent 角色，需要引入更复杂的消息路由和任务编排，建议先沉淀出稳定的任务接口。
- RAG 支持：在 Agent 循环中增加检索工具，核心是管理文档切分、向量索引和检索结果召回，不会改变外层架构。
- 流式输出：把 Celery 任务执行过程中的部分进度推送到 WebSocket 或 SSE，给用户更快的反馈。需要把任务中间结果写入 Redis，再由 API 层推送。
- 评估体系：建立测试集，每次改提示词或工具定义后自动跑回归，避免模型输出在“感觉上变好，实际却出现回归”的情况。
Manus 引发的讨论让更多人意识到，AI Agent 不只是提示词和模型能力的组合，更是一个需要认真对待的软件系统。希望这篇文章能给准备把 Agent 从 Demo 推向独立服务的团队一个起点，真正的考验在流量进来之后。
