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 和 WorkerCelery 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. 用最小工程骨架跑通一个可独立运行的 Agent3.1 环境准备和依赖说明建议使用 Python 3.11 及以上版本。下面的版本组合仅用于说明实际落地前要先确认与项目其他依赖的兼容性。依赖作用典型版本fastapiAPI 层框架0.115.xuvicornASGI 服务器0.30.xcelery异步任务队列5.4.xredis消息队列和缓存5.0.xsqlalchemyORM 和数据库访问2.0.xpsycopg2-binaryPostgreSQL 驱动2.9.xopenaiLLM 客户端 SDK1.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( modelgpt-4o-mini, messagesmessages, toolsTOOL_SCHEMAS, tool_choiceauto, ) 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保证多轮工具调用之间的上下文连续。工具结果以roletool追加并且必须带上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( nameget_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( namecalculate, 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: funknown 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_asciiFalse, defaultstr) 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, brokerredis://redis:6379/0, backendredis://redis:6379/0) celery_app.task(bindTrue, max_retries3, default_retry_delay5) 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(excexc) 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_modelTaskRead) def create_task(body: TaskCreate): db get_db_session() conversation Conversation(iduuid.uuid4(), user_idbody.user_id or anonymous) db.add(conversation) db.commit() task Task( iduuid.uuid4(), conversation_idconversation.id, statuspending, inputbody.input, ) db.add(task) db.commit() process_task.apply_async(args[str(task.id)], task_idstr(task.id)) return TaskRead( task_idtask.id, statustask.status, conversation_idconversation.id, ) router.get(/{task_id}, response_modelTaskRead) 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_idtask.id, statustask.status, outputtask.output, errortask.error)这里有一个容易忽略的点process_task.apply_async传入的task_id会让 Celery 任务 ID 和业务任务 ID 保持一致后续在 Celery 日志里看到task_id可以直接关联到数据库记录。4. 从本地跑通到部署配置、容器和启动顺序4.1 环境变量统一管理配置项包括数据库连接、Redis 地址、LLM API Key、模型名称、任务频控等。不要把配置硬编码在代码里。# .env.example DATABASE_URLpostgresql://agent:agentdb:5432/agent REDIS_URLredis://redis:6379/0 OPENAI_API_KEYsk-xxxx OPENAI_MODELgpt-4o-mini MAX_AGENT_STEPS8 TASK_TIMEOUT_SECONDS120在 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 --loglevelinfo 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 的最终回答。如果状态为failederror字段会给出错误摘要。完整链路需要查看 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几十秒后仍是pendingWorker 日志没有任何输出。排查顺序检查 Redis 是否连通docker compose exec redis redis-cli ping期望返回PONG。检查 Worker 是否启动docker compose logs worker看 Celery 是否注册了任务。检查 API 是否成功投递任务查看 API 日志中是否有process_task相关输出。检查任务是否被其它 Worker 消费如果存在多个 Worker可能消息被其他实例消费需要确认注册的任务名一致。常见原因和处理方式问题现象常见原因检查方式处理建议一直 pendingRedis broker 配置错误检查REDIS_URL是否为容器内地址修正环境变量后重启一直 pendingWorker 没有启动查看 worker 容器状态和日志启动 worker一直 pendingCelery 任务名不一致对比 API 和 Worker 的 import 路径统一任务 import 路径一直 pending数据库未健康检查 db healthcheck等数据库 ready 后再启动 Worker6.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 重启后缓存丢失会话会断。建议把会话历史持久化到 PostgreSQLRedis 只作为短期缓存。任务执行前从数据库读取完整消息列表任务结束后写回。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 推向独立服务的团队一个起点真正的考验在流量进来之后。
