AI Agent工程化实践:从自主决策到人工审核的完整闭环 “给 AI 150 英镑和三个月时间让它自己把订阅费赚回来”——这个实验最近在海外开发者圈子里讨论度很高。很多人第一反应是这又是一次“AI 取代人类”的营销炒作。但如果你真正拆开这个实验看会发现它其实是一个很典型的AI Agent 工程化压力测试预算有限、周期明确、交付物需要被真实用户认可并付费。这个设定比我们平时 demo 里跑的“AI 自动写周报”“AI 帮你订机票”要残酷得多因为它直接拿商业闭环来检验 AI 的自主能力。这篇文章不打算讨论“AI 会不会取代人”这种宏大话题而是想从工程视角拆解这个实验背后的技术逻辑。我会围绕几个问题展开AI Agent 要完成“自主赚钱”需要哪些核心能力组件为什么现阶段的 Agent 架构很难真正跑通长期盈利闭环如果你想把 AI 接入真实业务应该怎么设计成本边界、任务分解和人工审核机制最后我会给出一个可落地的 Agent 最小闭环示例并总结哪些环节适合放手交给 AI哪些环节必须留给人。1. 这个实验的本质不是“AI 赚钱”而是“Agent 工程闭环”先把这个实验拆开看。给 AI 150 英镑和三个月时间让它赚回自己的订阅费这个任务看起来很容易被理解成“AI 自己去找活儿干”。但从技术角度它真正考验的是以下五个环节能否全部打通。第一个环节是自主决策。AI 不能只等人类下指令它需要根据环境反馈比如某个平台流量不行、某种内容没人买单自己调整策略。第二个环节是任务拆解与工具调用。AI 需要把一个模糊目标“赚到订阅费”拆成若干具体动作做内容、做 SEO、写代码、接外包然后调用对应工具去执行。第三个环节是质量控制和自我纠错。AI 生成的内容能不能达到“有人愿意付费”的质量标准它能不能自己发现问题并迭代这是整个实验里最难的一环。第四个环节是商业化转化。AI 需要把流量或内容转化为真实收入这意味着它要理解支付、定价、客户沟通这些业务逻辑。第五个环节是成本管理。150 英镑是启动资金三个月是时间预算AI 每一步动作都在消耗 token、API 调用额度和时间它需要在这个约束下做取舍。理解了这五个环节你就会明白这个实验的本质不是“让 AI 赚钱”而是“验证 AI Agent 能否独立完成一个商业闭环”。这也直接对应了当前 AI 工程领域最核心的议题——Agent 到底能在多大程度上脱离人类监督独立完成复杂任务从技术现状来看答案并不乐观。以 2025 年的 Agent 能力水平AI 可以在单个环节表现出色比如生成文章、写代码、做图但要在 90 天里自主完成“计划-执行-纠错-变现”的完整闭环几乎一定会遇到瓶颈。这并非 AI 不行而是 Agent 架构在长期任务一致性、跨域知识整合和业务判断力三个维度上仍有明显短板。不过这个实验依然非常有价值。它的价值不在于结论是“能”还是“不能”而在于它逼着我们去思考如果要把 AI Agent 用在实际业务里应该怎么设计架构、怎么控制风险、怎么定义边界。2. AI Agent 的核心能力组件与架构选型要理解前述的五个环节必须先看 Agent 的技术架构。一个完整的 AI Agent 通常由以下四个核心组件组成。2.1 大脑大模型推理引擎这是 Agent 的决策中心。它负责理解用户目标、拆解任务、生成执行计划。你可以选择通用大模型如 GPT-4o、Claude 系列也可以选择垂直领域模型。在这个实验场景里通用模型是合理的起点因为它需要处理内容创作、代码编写、数据分析等多种任务。关键考虑是上下文窗口和记忆管理。Agent 要跑三个月不可能把全部历史记录塞进上下文。你需要一个记忆系统用向量数据库或外部存储保存关键决策和结果。2.2 手工具调用层Agent 不能只停留在“对话”它需要真正执行动作。这个实验涉及的工具有三类内容创作工具生成文章、图片、视频、开发工具写代码、部署 Web 应用、商业工具邮件发送、支付链接、社交媒体发布。在工程实现上工具调用一般通过Function Calling机制实现。大模型输出一个结构化的调用请求由 Agent 框架转发给真实的外部服务。You can use tools along with other functions to make decisions,但是在做题目规划时需要特别思考并设计工具回复的失败处理路径。2.3 神经记忆与状态管理这是一个最容易被低估的组件。Agent 在执行触发商业闭环的任务时如果没有“记住”上一次任务的效果评估和目标用户的反馈就会一次次重复相同的无效动作。所以在实现上你需要一个结构化的记忆模块至少记录三类信息目标状态当前总体目标、当前阶段目标。执行历史已尝试的动作、结果、误差与修正。用户/市场反馈用户偏好、付费意愿、内容表现数据。可以使用 Redis 保存实时状态用 SQLite 定期持久化快照或者直接使用向量数据库做语义检索。你可能需要系统性地组织代码结构然后构建出工程目录agent_earn/ ├── brain/ # 大模型调用与推理逻辑 ├── tools/ # 工具调用封装 ├── memory/ # 状态与记忆管理 ├── tasks/ # 任务定义与调度 ├── main.py # 入口文件 ├── config.yaml # 配置 ├── requirements.txt └── logs/ # 运行日志2.4 约束成本控制与权限边界这是一个实际项目中很少被强调但必须单独说明的组件。AI 每调用一次工具都会产生费用一个被允许自动执行“每天生成 10 篇文章并自动发布”的 Agent一个月可能烧掉几百美元 API 费用却不能带来任何收入。所以在做规划时系统必须内置成本开关。比如定义环境变量 MAX_DAILY_COST 和 MAX_TOTAL_COST在 Agent 每次调用工具前检查剩余预算。如果超限必须暂停执行并通知人类管理员。3. 环境准备与前置条件在搭建一个类似“AI 自主赚钱”的 Agent 工程闭环之前需要一些准备条件。这个实验不等于我们完全模拟海外的整个过程但它对我们构建一个最小可运行的 Agent 工程框架非常有帮助。3.1 硬件与基础环境Agent 本身并不需要高性能 GPU。除非你要在本地跑模型推理否则用 API 调用就可以完成。建议准备一台云服务器或本地开发机最少 2C4G 即可满足基础部署。系统环境Ubuntu 22.04 或 Windows WSL2也可以直接在 Windows 上运行。Python 3.10建议用虚拟环境管理依赖。一个大模型的 API Key比如 OpenAI、Anthropic 或国产兼容接口。一个用于存储状态和日志的数据库开发环境用 SQLite有一定规模后可以切换到 PostgreSQL。3.2 用到的核心库这里给出一个最小依赖集合并在后面附上完整的示例。openai1.0.0 # 大模型接口 langchain0.3.0 # Agent 框架可选也可以直接用原生 API flask3.0.0 # 如果要做 Web 服务 requests2.31.0 # 工具调用 schedule1.2.0 # 定时调度 python-dotenv1.0.0 # 环境变量管理 sqlite3 # Python 内置需要注意的是这些库只是通用依赖具体的版本号以你的实际环境为准。关键是要理解整个工程的架构设计。3.3 三个真实场景的服务器账号可选如果让 Agent 真正接入商业闭环可能需要准备博客平台账号、支付账号和社交媒体账号同时强调正式账号必须有合法合规的使用授权。至少准备一个沙箱/测试环境先验证调用流程再切换到真实环境。不要在生产环境里直接放开 Agent 的自动发布权限必须先经过人工审核。4. 核心流程拆解Agent 如何完成“自主盈利”任务现在我们把“赚回订阅费”这个任务拆成工程可执行的流程。这个流程可以直接复用到其他 Agent 项目里。4.1 任务定义与目标拆解不要把“赚到钱”作为一个单一 Prompt 丢给模型。模型需要的是一个结构化的目标和子任务列表。示例目标定义目标三个月内赚回 150 英镑 预算上限50 英镑用于 API 调用与工具成本 核心资金渠道内容付费 开发者工具订阅 小型外包任务 拆解 1. 第 1-2 周搭建内容基础。创建技术博客发布 10 篇高质量文章。 2. 第 3-4 周启动 SEO。优化已有内容提交搜索引擎收录分析流量数据。 3. 第 5-8 周推出 MVP 产品。基于自身能力开发一个小型命令行工具定价 9 英镑。 4. 第 9-12 周多渠道投放。在社交媒体、开发者社区发布推广内容接触潜在客户。拆解原则是每个子任务必须可验证、必须有明确的交付物、必须给后续任务提供反馈数据。4.2 自主执行基于 ReAct 模式的任务循环ReActReasoning Acting是 Agent 最常用的执行模式模型先推理再行动再观察结果循环迭代。在这个框架下Agent 的运行流为读取目标任务。规划执行步骤。调用对应工具。观察工具执行结果。根据结果决定下一步是继续执行、切换策略还是请求人工介入。代码实现的核心部分可以封装为一个函数# 文件路径agent_earn/tasks/loop.py from typing import Dict, Any from tools.content_tool import ContentTool from tools.seo_tool import SEOTool from memory.manager import MemoryManager def agent_loop(base_task: str) - Dict[str, Any]: agent_memory MemoryManager() content_tool ContentTool() seo_tool SEOTool() step_tasks agent_memory.get_current_plan() for task in step_tasks: if task[type] content: result content_tool.run(task[params]) elif task[type] seo: result seo_tool.run(task[params]) else: result {status: skipped, reason: unknown task type} metric evaluate_result(result) agent_memory.record(task, result, metric) if metric[needs_human]: pause_and_notify(task, result)上面代码的逻辑是Agent 从记忆模块读取当前计划逐个执行任务评估结果记录反馈并在需要时请求人工介入。它代表一种“人机协同”的 Agent 运行模型。4.3 反馈与自动纠错“自动纠错”是 Agent 与普通脚本最大区别。普通脚本按固定规则跑而 Agent 在遇到预期之外的结果时应该能调整策略。举个例子假设 Agent 在公众号平台发布了文章结果只有 50 次点击、0 次订阅。它不能继续无脑发布同类型内容而应该去分析原因是标题不够吸引还是发布时间不对还是内容根本不符合目标读者然后基于分析结果修改下一轮内容策略。在工程实现上这一步可以用如下逻辑控制# 文件路径agent_earn/tasks/feedback.py import json def evaluate_result(result: dict) - dict: clicks result.get(clicks, 0) conversion result.get(conversion, 0) # 定义质量阈值低于阈值的任务标记为 attention if clicks 100 or conversion 0.01: return { status: attention, needs_human: True, reason: fclick{clicks}, conversion{conversion} below threshold } return { status: ok, needs_human: False }注意这里标记为什么需要人工介入低到什么程度算“失败”策略转向的标准是什么如果让 AI 自己定义标准很容易陷入盲目试错。4.4 人工审核安全开关在最核心的商业环节发布内容、回复客户、修改产品定价我建议无论如何保留人工审核。这不是不信任 AI而是避免不可逆的错误比如错误地把产品定价改成 0 元、错误地向客户发送包含错误信息的邮件。工程实现上给每个关键动作增加一个审批回调# 文件路径agent_earn/core/human_review.py def require_human_approval(action: str, context: dict): print([需要人工审批], action) print(上下文:, json.dumps(context, ensure_asciiFalse, indent2)) approval input(是否批准执行y/n) return approval.lower() y在实际部署中这个人工审批可以通过企业微信、钉钉等 webhook 推送到负责人手机上而不是用命令行 input。但原理是一样的关键操作必须暂停等待人类确认。5. 完整示例代码实现搭建最小可用 Agent 工程闭环为了让读者真正能落地这里给一个可运行的最小工程示例。代码量不大但它包含了 Agent 工程最核心的四个能力组件。5.1 环境变量配置创建.env文件# 文件路径.env OPENAI_API_KEYsk-xxxxxxxxxxxx MAX_TOTAL_COST50 MAX_DAILY_COST3 TASK_MODEsandbox ALERT_WEBHOOK_URLhttps://your-notification-hook.example.com这里的 TASK_MODEsandbox 表明当前是沙箱测试模式。只有切到 production 才允许 Agent 直接操作真实账号。5.2 大模型调用封装# 文件路径agent_earn/brain/llm_client.py import os import json from openai import OpenAI from dotenv import load_dotenv load_dotenv() class LLMClient: def __init__(self): self.client OpenAI( api_keyos.getenv(OPENAI_API_KEY), ) self.model gpt-4o-mini def plan_task(self, goal: str) - list: 根据目标生成子任务列表 completion self.client.chat.completions.create( modelself.model, messages[ { role: system, content: 你是一位资深项目管理者。请把复杂目标拆解为可执行的子任务。 }, { role: user, content: f目标{goal}。请输出 JSON 格式的子任务列表。 } ], response_format{type: json_object} ) result completion.choices[0].message.content try: data json.loads(result) return data.get(tasks, []) except json.JSONDecodeError: return []这个文件的作用是把 OpenAI 接口封装成 Agent 的“规划器”。注意设置了 response_format 为 json_object方便后续解析任务。5.3 工具调用层# 文件路径agent_earn/tools/content_tool.py import datetime class ContentTool: 模拟内容发布工具 def __init__(self): self.published [] def publish(self, title: str, body: str) - dict: # 在实际项目中这里调用博客平台 API # 这里模拟一个发布结果避免依赖外部系统 article_id len(self.published) 1 result { article_id: article_id, title: title, status: published, publish_time: datetime.datetime.now().isoformat() } self.published.append(result) return result def analyze_traffic(self, article_id: int) - dict: # 模拟流量分析结果 return { article_id: article_id, clicks: 150, unique_visitors: 45, conversion_rate: 0.006 }5.4 记忆管理器# 文件路径agent_earn/memory/manager.py import sqlite3 import json class MemoryManager: def __init__(self, db_pathmemory.db): self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS task_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_type TEXT, task_name TEXT, params TEXT, result TEXT, metric TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) self.conn.commit() def record(self, task_type: str, task_name: str, params: dict, result: dict, metric: dict): self.conn.execute( INSERT INTO task_history (task_type, task_name, params, result, metric) VALUES (?, ?, ?, ?, ?), (task_type, task_name, json.dumps(params, ensure_asciiFalse), json.dumps(result, ensure_asciiFalse), json.dumps(metric, ensure_asciiFalse)) ) self.conn.commit() def get_recent(self, limit: int 10) - list: cursor self.conn.execute( SELECT task_name, metric, created_at FROM task_history ORDER BY id DESC LIMIT ?, (limit,) ) rows cursor.fetchall() return [{task_name: r[0], metric: r[1], created_at: r[2]} for r in rows]5.5 主入口串起整个闭环# 文件路径agent_earn/main.py import time import json import os from dotenv import load_dotenv from brain.llm_client import LLMClient from tools.content_tool import ContentTool from memory.manager import MemoryManager from tasks.feedback import evaluate_result load_dotenv() def run_daily_round(): llm LLMClient() memory MemoryManager() content_tool ContentTool() goal 生成一篇关于AI Agent入门的技术博客文章发布到技术社区目标获得500次阅读和20个收藏 # 1. AI 规划 tasks llm.plan_task(goal) print(AI 规划的子任务) print(json.dumps(tasks, ensure_asciiFalse, indent2)) # 2. 执行子任务 for task in tasks: if task.get(type) content: result content_tool.publish(task.get(title, AI Agent实践), 文章正文内容...) traffic content_tool.analyze_traffic(result[article_id]) metric evaluate_result(traffic) # 3. 记录到记忆库 memory.record( task_typecontent, task_nametask.get(title, ), paramstask, resultresult, metricmetric ) print(发布结果, result) print(流量指标, traffic) # 4. 根据指标判断是否人工介入 if metric[needs_human]: print(⚠️ 需要人工关注, metric[reason]) time.sleep(0.5) if __name__ __main__: run_daily_round()然后在项目根目录下创建 requirements.txtopenai1.0.0 python-dotenv1.0.0运行前先安装依赖pip install -r requirements.txt然后执行python main.py这样就跑通了一个最简 Agent 闭环AI 自主规划任务、调用工具、记录结果、评估指标、决定是否触发人工介入。你后续要扩展真实工具只需要替换 ContentTool 里的模拟方法为真实 API 调用。6. 运行结果与效果验证运行上面的程序后你会看到类似输出AI 规划的子任务 [ { type: content, title: AI Agent入门教程从零到落地 } ] 发布结果{article_id: 1, title: AI Agent入门教程从零到落地, status: published, publish_time: 2025-01-10T10:23:45} 流量指标{article_id: 1, clicks: 150, unique_visitors: 45, conversion_rate: 0.006} ⚠️ 需要人工关注click150, conversion0.006 below threshold重点看最后一行它表明 Agent 在完成一次任务后根据预设阈值识别出“当前内容策略效果不佳”并请求人类介入。这就是整个闭环的验证标准。怎么判断 Agent 是否正常工作你可以检查以下几点子任务是否合理拆解任务列表不应只有“生成内容”还可以有“SEO优化”“多渠道分发”等。记忆库是否累积了有效数据查询 memory.db是否能看到每次任务的结果。指标判断是否触发如果所有结果都是 ok说明你设置的阈值太低Agent 永远不会求助这会埋下隐患。人工介入是否真正暂停了动作这是最重要的安全机制必须验证。7. 常见问题与排查思路7.1 任务规划不合理子任务过于笼统可能原因Prompt 没有提供足够的约束与背景信息。排查方式检查大模型输出的原始子任务列表看看它是否具备可执行性。解决方案在 system prompt 中加入“每个子任务必须包含可交付物、验证指标和截止时间”的约束并给一个大模型可以参考的示例。7.2 Agent 做了一两步就停了不能持续执行可能原因Agent 在某个步骤中到达了上下文长度限制或工具返回值不符合预期导致推理链条中断。排查方式查看运行日志定位中断时的工具返回。解决方案在工具调用外层增加 try/except捕获异常并反馈给模型同时压缩记忆内容只保留必要的状态信息。7.3 模型生成的 JSON 格式无法解析可能原因模型输出偶尔会夹带说明性文字导致 JSON 不合法。排查方式打印原始 completion 内容。解决方案增加 JSON 解析失败重试机制并尝试截取首尾方括号或花括号后再解析。7.4 API 调用成本超预算可能原因Agent 循环次数太多或者每次调用都塞入了大量上下文。排查方式计算每次运行消耗的 token 数统计单日调用量。解决方案限制任务循环最大次数压缩历史记录使用更便宜的小模型处理简单子任务。在代码中加入预算开关if daily_cost MAX_DAILY_COST: send_alert(当日预算已耗尽暂停自动执行) break7.5 Agent 在真实业务中做出了错误决策可能原因缺少人工审批环节或质量阈值设置不当。排查方式回查任务记录分析当前环境触发条件与实际结果的关系。解决方案对不可逆操作强制接入人工审批对高风险操作加“二次确认”机制建立灰度发布机制先在低风险环境运行一段时间。为了方便查看我将这些问题整理成表格问题现象可能原因排查方式解决方案子任务过于笼统Prompt 约束不足检查原始输出列表在 Prompt 中增加可交付物指标约束Agent 运行中断上下文超限或工具返回异常查看运行日志定位断点增加异常捕获压缩历史记忆JSON 解析失败模型输出附带说明文字打印原始返回值增加重试机制与截断解析调用成本超预算循环次数过多或上下文过长统计 token 与调用量限制循环次数使用小模型处理简单任务业务决策错误缺少人工审批或阈值过低回查任务记录不可逆操作强制人工审批增加灰度验证8. 最佳实践与工程建议这个实验真正想验证的事情其实就是 Agent 在真实场景中的边界。根据前面拆解可以总结出几条对实际项目有价值的工程建议。8.1 坚持“人在回路”设计不要把关键操作完全放开不论你的 Agent 有多强至少在以下环节保留人工节点对外发布、定价变更、客户沟通、付款操作。工程实现上可以用审批机制也可以用任务状态机来限制 Agent 对不可逆操作的权限。8.2 用预算和权限构成双保险AI Agent 出问题往往不是“能力不够”而是“没人告诉它哪些事情是底线”。比如自动调用外部 API 前必须检查预算、不允许修改生产环境数据库、不允许向外部发送 email 等。在配置层面就固化好这些边界# config.yaml agent: max_steps_per_task: 20 max_daily_cost: 3 max_monthly_cost: 50 allowed_tools: - content_publish - seo_analysis - data_analysis restricted_tools: - send_email - modify_payment - delete_database8.3 用数据驱动 Agent 的策略更新Agent 的执行策略不能写死在 Prompt 里随着任务进行它要根据历史反馈动态调整。最好的做法是定期比如每周让模型分析记忆库中的历史数据输出一份“策略复盘报告”然后由人来决定是否采纳报告中的策略调整建议。8.4 关注成本但别只关注成本很多人看到“AI 赚钱实验”第一反应是算 API 成本。但在真实的 Agent 工程里成本不只是 token 费用还包括人工审核的时间成本、错误操作带来的修复成本、以及被风险操作触发的合规成本。在做 Agent 工程预算时要用全链路成本TCO的方式来评估而不是只看单次 API 调用价格。8.5 从简单场景起步先让 Agent 赚到“小钱”如果你也想尝试类似的 Agent 实验不要一开始就设定“三个月赚回 150 英镑”这种目标建议先从一个更小的闭环起步比如让 Agent 每天生成一篇技术文章并自动发布到博客然后人工检查它带来的流量变化。跑通之后再逐步添加 SEO 分析、邮件的自动化、产品页面生成等模块。每个新模块都应该是独立的能力和风险都有独立控制。这样即使某个模块失败也不会摧毁整个系统。为什么这么设计因为 Agent 和传统软件的故障模式完全不同。传统软件的错误是可预测的、可测试的。Agent 的错误往往是行为层面的策略不对、目标理解有误、环境变化导致模型推理失败。这类错误很难通过单元测试发现必须在小范围、可回滚的实验中逐步暴露。8.6 记录一切但要做结构化记录Agent 的调试和传统软件调试不一样你无法靠断点定位问题。你只能依赖日志、记忆库和任务记录来还原它的“思考过程”。因此所有关键动作都必须记录以下信息任务来源与目标。模型输入 Prompt 摘要。工具调用参数与返回结果。决策依据模型输出的推理过程。人工介入点与处理结果。这些记录不仅用于故障排查还可以作为后续微调模型或者设计 Prompt 的语料。9. 总结与后续学习方向回到开头那个实验给 AI 150 英镑和三个月时间让它自己赚回订阅费。单从结果看多数此类实验大概率无法在三个月内实现稳定盈利但这个实验对技术社区的价值恰恰不在结果而在于它把一个 AI Agent 放进了真实世界的约束里。通过这次拆解我们应该记住几个关键判断第一AI Agent 的工程难点不是“模型能力”而是“系统设计”。模型负责生成内容系统负责控制成本、保障安全、记录反馈、触发纠错。后者的成熟度决定了一个 Agent 能否从 demo 走向生产环境。第二“自主盈利”这类复杂目标和“自动生成一篇文章”这类简单任务之间隔着十几层工程细节。每一层都可能失败比如任务分解不准确、工具调用不稳定、反馈评估标准不当、成本控制失效。这些细节才是一名 AI 工程师真正要投入精力的地方。第三如果想让 AI 产生真实商业价值更好的路径不是“完全自主”而是“人机协同”AI 承担高杠杆的生成和执行人类承担高价值的判断和决策。这也是当前阶段最务实、最有落地可能性的 Agent 工程范式。如果你想继续深入研究这个方向我建议按以下路径学习掌握 ReAct 等 Agent 推理模式看 OpenAI Function Calling 和 LangChain AgentExecutor 的实现方式。深入了解向量数据库和记忆系统这对 Agent 在长周期任务中的稳定性至关重要。分析成本控制研究 token 压缩、模型分级路由等工程手段。找一个小业务闭环实践比如“AI 自动生成一篇优质博客并追踪数据”把能力边界摸清楚。学习 MLOps 里的模型评估与监控方法它可以直接迁移到 Agent 的状态监控与策略迭代上。最后提醒一点任何涉及生产环境、自动发布或真实账户的操作一定要在测试环境先行验证保留回滚方案遵守最小权限原则。Agent 带来的效率提升是真实的但它同时也是新的风险源需要在工程上认真设计安全带。