“我给了 AI 150 英镑和三个月时间让它自己赚回订阅费”——这其实是一个很值得技术人拆解的题目。它不只是一个关于“AI 能不能赚钱”的尝鲜实验更像是一次小规模、低成本的 AI Agent 工程实践提前设定预算上限、规定运行周期、让 AI 自动执行任务并评估产出。换句话说这个项目的重点不是“AI 神奇地赚钱了”而是如何给 AI 一个代运营式的闭环目标明确、预算有限、时刻监控、按产出复盘。本文会把这个实验拆成一个可落地的技术工程来实现。我会带你从 0 到 1 搭建一个“自主代理 预算控制 任务调度 数据回收”的小系统并给出可直接复制的 Python 代码、配置文件和运行日志示例。无论你是刚开始接触 AI Agent还是已经在做企业级 AI 应用落地这套思路都能迁移到你的实际项目里。1. 背景与核心概念1.1 这个实验到底在做什么这个题目可以理解为给一个 AI 系统一笔启动资金150 英镑约等于日常订阅费用的 6 到 10 个月再给它一个明确的时间窗口3 个月让它在无人干预或极少干预的情况下自动完成一系列的“创收任务”并用这些任务的收益覆盖它自身每个月需要支付的 API 或服务订阅费。从工程角度看这个实验要解决的并不是“AI 有没有自动赚钱的能力”而是三个非常现实的工程问题如何让 AI 在一个固定预算下持续运行而不是一次调用就把额度烧光。如何自动拆解任务、执行任务、校验任务结果并记录每一次投入和产出。如何判断 AI 三个月的运行效果而不只是看它“生成了多少文本”。这三个问题也恰好是企业在把 AI 接入生产环境时最关心的成本控制、任务编排和效果评估问题。1.2 核心概念扫盲在正式开始前先把几个高频词说清楚。AI AgentAI 代理AI Agent 指的是一个能感知环境、自主决策并执行动作的系统。它不只是“聊天”而是能把任务拆解成多个步骤调用工具或 API并根据中间结果继续迭代。比如它可以每天自动生成一篇图文内容再自动把内容推送发布再自动统计阅读和转化。任务编排Orchestration任务编排是把一个大目标拆成多个可执行的小任务并按依赖关系串起来。在本文的项目里我们可以把“三个月实现自给自足”的目标拆成“生成内容、发布内容、回收数据、调整策略”四个循环步骤。API 成本控制调用大模型 API 是按 token 计费的没有成本控制的话一个失控的循环可以在几分钟内消耗大量费用。所以预算管理是自主 Agent 项目中最重要的模块之一。ROI投入产出比ROI 衡量的是“花出去的钱”和“赚回来的价值”的比例。如果 AI 在三个月后每月的产出价值大于它每月的订阅和 API 成本那这个实验才算真正跑通。1.3 为什么这对开发者很重要过去大家关注 AI 的方式是“它能做什么”。但现在企业更关心的是“让它持续做需要多少钱能换回多少收益”。这套带预算、带监控、带回流的 Agent 工程能力正好是这两种关心之间的桥梁。掌握它你就能写出“可持续运行”的 AI 应用而不是只能在本地跑一次的 Demo。2. 项目目标与约束设计2.1 实验目标拆解原始标题里的目标是“让 AI 赚回自己的订阅费”。为了让目标可执行我们需要把它拆成技术指标。这里我制定一组最简可验收指标指标数值说明初始预算150 英镑可用于所有 API 调用和其他必要支出运行周期90 天三个月每月订阅成本20 英镑需要 AI 每月产出覆盖的固定成本最低目标每月 20 英镑三个月累计至少 60 英镑风控线预算剩余低于 10 英镑时自动降低任务频率注意货币和金额都可以替换成人民币或你自己的虚拟预算。重要的是预算上限、周期和最低回报这三组参数清晰。2.2 延迟满足还是立即回报在真实业务里AI 给你带来的收益有两种一种是即时收益比如自动接单、生成可售卖的文案另一种是延迟收益比如内容涨粉、品牌曝光、用户留存。这个实验要控制变量所以最好选择“即时收益”型任务比如自动生成指定主题的短文案打包卖给小商家。自动完成一批数据整理任务按条计费。自动生成产品介绍视频脚本按稿计费。选择“即时收益”任务的原因很简单可量化、可回传、适合自动化流程验证。2.3 技术约束清单技术实现上有几个硬约束单次任务必须设定 API 调用成本上限超过则拒绝执行。如果模型返回格式不正确重试次数必须有限避免死循环。每天最多执行 N 个任务避免“勤奋”过度导致预算失控。所有收支记录写入本地文件方便事后复盘。有了这些约束下面的代码才能保证不会让 Agent 疯狂烧钱。3. 环境准备与项目结构3.1 运行环境本文示例使用下面的环境版本可以根据你的实际项目调整操作系统Windows 10 / macOS / Ubuntu 均可Python3.10 及以上OpenAI Python 库1.x 版本密钥管理使用环境变量不硬编码在代码里代码编辑器VS Code 或 PyCharm推荐提前创建一个虚拟环境避免依赖冲突。3.2 安装依赖只需要安装一个核心依赖pip install openai如果你希望把数据写进 SQLite 而不是本地 JSON 文件可以再加上标准库sqlite3它不需要额外安装。3.3 项目结构规划为了让整个系统清晰我们按下面目录结构来组织代码ai_agent_lab/ ├── config.py # 全局配置 ├── budget_tracker.py # 预算与收支控制器 ├── agent_core.py # AI Agent 核心负责调用模型 ├── task_scheduler.py # 任务调度器 ├── task_bank.py # 任务池定义具体任务 ├── main.py # 入口脚本 ├── data/ │ └── ledger.json # 收支账单自动生成 └── logs/ └── agent.log # 运行日志自动生成如果你只是做快速验证也可以把前三个模块合并成一个文件。但为了可维护性和文章讲解方便我按模块拆分。3.4 API 密钥配置这里以 OpenAI 通用接口为例其他厂商的 API 结构字段一致把配置替换掉即可。export OPENAI_API_KEY你的密钥如果你的 API 服务需要设置 base_url也可以在config.py中指定。4. 核心系统设计4.1 整体运行闭环整个自动运行流程可以用下面这条链路表示预算检查 → 任务生成或任务选取 → 模型调用 → 结果校验 → 产出记录 → 收益回写 → 日志输出 → 等待下一次调度每一次循环结束预算控制器都会更新剩余额度。如果额度低于阈值系统会自动降低执行频率如果额度耗尽系统自动休眠。4.2 预算控制模块设计预算模块是整个系统的“刹车”。我选择用一个独立的BudgetTracker类来管理而不是把余额逻辑散落在各处。它需要承担四件事记录初始预算。扣减每一次模型调用费用。收入回写。查询是否还有余额、还能跑多少个任务。这里有一个容易被忽略的点大模型调用的费用是不确定的因为它取决于输入输出 token 数。所以每次调用后必须先用返回的usage字段计算出真实费用再写入账本。4.3 任务执行器设计任务执行器需要具备三个特征每个任务有独立的名字。每个任务有预估成本上限。每个任务执行前后都会更新账本。如果某个任务连续失败应该记录失败次数并在达到阈值后切换到别的任务。4.4 运行日志设计生产环境里的 AI 应用一定要有日志。我们可以用标准库logging输出到控制台和文件记录每次调用的模型、token 数、成本、结果摘要。5. 完整实战案例下面我们开始写完整可运行的项目。为了让代码尽量通用我会把“收益逻辑”做成一个可替换的函数。你可以在实际项目中把它换成你的真实业务通道。5.1 配置文件 config.py# 文件路径ai_agent_lab/config.py import os # 初始实验预算英镑 INITIAL_BUDGET 150.0 # 每月需要覆盖的订阅成本英镑 SUBSCRIPTION_COST 20.0 # 实验周期天 DEADLINE_DAYS 90 # 成本红线低于该金额时系统会自动进入保守模式 LOW_BUDGET_THRESHOLD 10.0 # 单次任务模型调用成本上限英镑 MAX_TASK_COST 0.5 # 最大重试次数 MAX_RETRIES 2 # 每天最大执行任务数 MAX_TASKS_PER_DAY 10 # 模型配置 LLM_MODEL gpt-4o-mini TEMPERATURE 0.7 # API 配置 API_KEY os.environ.get(OPENAI_API_KEY, ) BASE_URL os.environ.get(OPENAI_BASE_URL, ) # 账本和日志路径 LEDGER_PATH data/ledger.json LOG_PATH logs/agent.log5.2 预算追踪器 budget_tracker.py这个模块负责所有收支记账。它会把每笔明细写入ledger.json保证程序重启后数据还在。# 文件路径ai_agent_lab/budget_tracker.py import json import os from datetime import datetime class BudgetTracker: def __init__(self, initial_budget: float, ledger_path: str): self.initial_budget initial_budget self.ledger_path ledger_path self.balance initial_budget self.records [] self._load() def _load(self): if not os.path.exists(self.ledger_path): return with open(self.ledger_path, r, encodingutf-8) as f: data json.load(f) self.balance data.get(balance, self.initial_budget) self.records data.get(records, []) def _save(self): os.makedirs(os.path.dirname(self.ledger_path), exist_okTrue) with open(self.ledger_path, w, encodingutf-8) as f: json.dump( { initial_budget: self.initial_budget, balance: self.balance, records: self.records, }, f, ensure_asciiFalse, indent2, ) def spend(self, amount: float, item: str): if amount 0: raise ValueError(支出金额必须大于0) if amount self.balance: raise ValueError(支出超出剩余预算) self.balance - amount self.records.append( { type: spend, amount: amount, item: item, time: datetime.now().isoformat(), } ) self._save() def earn(self, amount: float, item: str): if amount 0: raise ValueError(收入金额必须大于0) self.balance amount self.records.append( { type: earn, amount: amount, item: item, time: datetime.now().isoformat(), } ) self._save() def can_afford(self, amount: float) - bool: return self.balance amount property def is_low_budget(self) - bool: return self.balance 10.0 def summary(self) - dict: total_spend sum(r[amount] for r in self.records if r[type] spend) total_earn sum(r[amount] for r in self.records if r[type] earn) return { initial_budget: self.initial_budget, balance: self.balance, total_spend: total_spend, total_earn: total_earn, }5.3 Agent 核心 agent_core.pyAgentCore只负责与大模型交互并返回标准化结果。为了让费用计算更准确我直接读取response.usage中的 token 数再乘一个模拟单价。注意不同模型的价格不同代码里我使用一个CALC_PRICE_PER_1K_TOKENS的模拟常量方便演示。你在实际项目中应该把这里的模拟价格替换为你的 API 实际报价。# 文件路径ai_agent_lab/agent_core.py from openai import OpenAI # 模拟价格每 1000 个 token 的费用英镑 # 这里只是演示请替换成你的模型真实价格 CALC_PRICE_PER_1K_TOKENS 0.002 class AgentCore: def __init__(self, api_key: str, base_url: str , model: str gpt-4o-mini): if base_url: self.client OpenAI(api_keyapi_key, base_urlbase_url) else: self.client OpenAI(api_keyapi_key) self.model model def run(self, system_prompt: str, user_message: str) - dict: 执行一次模型调用返回结果文本和费用信息。 response self.client.chat.completions.create( modelself.model, messages[ {role: system, content: system_prompt}, {role: user, content: user_message}, ], temperature0.7, ) output_text response.choices[0].message.content usage response.usage input_tokens usage.prompt_tokens output_tokens usage.completion_tokens total_tokens usage.total_tokens # 模拟费用计算实际请用官方价格表 total_cost (total_tokens / 1000) * CALC_PRICE_PER_1K_TOKENS return { output: output_text, input_tokens: input_tokens, output_tokens: output_tokens, total_tokens: total_tokens, cost: round(total_cost, 6), }5.4 任务池 task_bank.py为了演示我在这里设计两个任务write_slogan为一家小店生成推广文案。summarize_notes把一段杂乱的笔记整理成结构化文档。你可以把这两个任务替换成任何真实业务。# 文件路径ai_agent_lab/task_bank.py TASKS [ { name: write_slogan, description: 为奶茶店撰写一句20字以内的推广标语, system_prompt: 你是一名资深营销文案专家。输出必须简洁、有传播力。, user_message: 这家奶茶店主打鲜奶和手作茶目标人群是年轻上班族请你写一句20字以内的中文推广标语。, expected_output_type: text, max_cost: 0.3, }, { name: summarize_notes, description: 把产品需求笔记整理成清晰文档, system_prompt: 你是一名产品经理擅长整理需求文档。, user_message: 请把下面这段原始笔记整理成三段式文档背景、方案、验收标准。原始笔记用户反馈搜索很慢希望增加关键词联想功能最好能按热度排序也需要考虑搜索历史。, expected_output_type: text, max_cost: 0.4, }, ]5.5 任务调度器 task_scheduler.py调度器负责最外层的循环控制。它会检查预算挑选任务调用模型并根据结果模拟“收入入账”。这里的收入入账是演示逻辑实际项目中你可以把“收入”替换成真实的订单、成交额或打款数据。演示逻辑采用“任务成功执行则获得固定收益”的假设方便验证闭环。# 文件路径ai_agent_lab/task_scheduler.py import time import random import logging from budget_tracker import BudgetTracker from agent_core import AgentCore from task_bank import TASKS logger logging.getLogger(agent_scheduler) class TaskScheduler: def __init__( self, tracker: BudgetTracker, agent: AgentCore, max_tasks_per_day: int 10, low_budget_threshold: float 10.0, ): self.tracker tracker self.agent agent self.max_tasks_per_day max_tasks_per_day self.low_budget_threshold low_budget_threshold def run_one_round(self) - dict: 执行一轮任务选任务→检查预算→调用模型→记账。 if not self.tracker.can_afford(0.1): logger.warning(预算不足停止执行) return {status: no_budget} task random.choice(TASKS) logger.info(选择任务%s, task[name]) if not self.tracker.can_afford(task[max_cost]): logger.warning(任务 %s 的预估成本超出剩余预算跳过, task[name]) return {status: task_too_expensive, task: task[name]} # 先预留成本执行前扣一次模拟费用 self.tracker.spend(task[max_cost], ftask_start:{task[name]}) try: result self.agent.run( system_prompttask[system_prompt], user_messagetask[user_message], ) except Exception as e: logger.error(任务 %s 调用模型失败%s, task[name], str(e)) return {status: failed, task: task[name], error: str(e)} # 把实际费用和预留费用的差额补扣或退回 actual_cost result[cost] if actual_cost task[max_cost]: refund task[max_cost] - actual_cost self.tracker.earn(refund, frefund:{task[name]}) else: extra actual_cost - task[max_cost] self.tracker.spend(extra, fextra_cost:{task[name]}) # 模拟收入任务成功完成后返回一部分收益 # 实际项目中这里应调用你的业务接口 revenue round(random.uniform(0.05, 0.15), 4) self.tracker.earn(revenue, frevenue:{task[name]}) logger.info( 任务 %s 完成token 消耗 %s成本 %s模拟收益 %s, task[name], result[total_tokens], result[cost], revenue, ) return { status: success, task: task[name], cost: actual_cost, revenue: revenue, output: result[output], } def run_many_rounds(self, rounds: int): 连续执行多轮任务供实验验证使用。 results [] for i in range(rounds): r self.run_one_round() results.append(r) if r[status] no_budget: break time.sleep(0.5) return results5.6 入口程序 main.pymain.py负责把预算追踪器、Agent Core、调度器组合起来并输出一份完整实验报告。# 文件路径ai_agent_lab/main.py import json import logging import os from datetime import datetime from budget_tracker import BudgetTracker from agent_core import AgentCore from task_scheduler import TaskScheduler from config import ( API_KEY, BASE_URL, INITIAL_BUDGET, LEDGER_PATH, LOG_PATH, LLM_MODEL, MAX_TASKS_PER_DAY, LOW_BUDGET_THRESHOLD, ) # 配置日志 os.makedirs(logs, exist_okTrue) logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(LOG_PATH, encodingutf-8), logging.StreamHandler(), ], ) logger logging.getLogger(main) def main(): if not API_KEY: logger.error(请先设置 OPENAI_API_KEY 环境变量) return tracker BudgetTracker(initial_budgetINITIAL_BUDGET, ledger_pathLEDGER_PATH) agent AgentCore(api_keyAPI_KEY, base_urlBASE_URL, modelLLM_MODEL) scheduler TaskScheduler( trackertracker, agentagent, max_tasks_per_dayMAX_TASKS_PER_DAY, low_budget_thresholdLOW_BUDGET_THRESHOLD, ) logger.info(开始实验初始预算%s, INITIAL_BUDGET) results scheduler.run_many_rounds(rounds10) summary tracker.summary() logger.info(实验结束账单摘要%s, json.dumps(summary, ensure_asciiFalse)) # 简单统计分析 success_count sum(1 for r in results if r[status] success) failed_count sum(1 for r in results if r[status] failed) total_revenue sum(r.get(revenue, 0) for r in results) print(\n 实验报告 ) print(成功任务数, success_count) print(失败任务数, failed_count) print(当前余额, summary[balance]) print(累计收益, total_revenue) print(累计支出, summary[total_spend]) print(\n) if __name__ __main__: main()6. 运行与验证6.1 初始化数据目录第一次运行前先创建数据目录mkdir -p ai_agent_lab/data cd ai_agent_lab6.2 执行实验入口在终端执行python main.py如果一切正常你会看到类似下面的输出2025-01-06 10:00:01 - main - INFO - 开始实验初始预算150.0 2025-01-06 10:00:02 - agent_scheduler - INFO - 选择任务write_slogan 2025-01-06 10:00:03 - agent_scheduler - INFO - 任务 write_slogan 完成token 消耗 198成本 0.0004模拟收益 0.08 2025-01-06 10:00:04 - agent_scheduler - INFO - 选择任务summarize_notes ... 实验报告 成功任务数10 失败任务数0 当前余额150.76 累计收益0.82 累计支出0.24 6.3 验证账本文件程序结束后打开data/ledger.json你应该能看到每一笔支出和收入的记录。这就是 Agent 实验里的“财务报表”。{ initial_budget: 150.0, balance: 150.76, records: [ { type: spend, amount: 0.3, item: task_start:write_slogan, time: 2025-01-06T10:00:02 } ] }6.4 延长周期模拟真实实验当然不可能只跑 10 轮。你可以在task_scheduler.py中修改run_many_rounds的循环条件让它跑满 90 天。只需要把原来的固定轮数改成按日期判断即可def run_until_deadline(self, deadline_days: int): start time.time() deadline_seconds deadline_days * 24 * 3600 rounds 0 while time.time() - start deadline_seconds: if self.tracker.can_afford(0.1) is False: break r self.run_one_round() rounds 1 time.sleep(3600) # 每小时跑一轮模拟真实调度频率 return rounds注意真实跑 90 天时建议把time.sleep(3600)改成定时任务或 cron 调度而不是让 Python 进程一直挂着。7. 常见问题与排查思路自主 Agent 项目的坑比普通脚本多很多。下面把最容易遇到的几类问题整理成表格。问题现象常见原因解决思路运行后预算迅速耗尽没有设置单次任务成本上限或模型参数过大为每个任务设置max_cost并检查模型调用是否在循环内被重复触发API 返回超时或限流请求频率过高或 API Key 余额不足增加退避重试使用指数退避算法并降低调度频率模型返回 JSON 解析失败输出格式不稳定在 system prompt 中明确要求输出 JSON并使用response_format{type: json_object}收入入账为零业务回传接口没接通先在小范围测试业务接口确认返回后再接入调度器程序重启后余额不对账本文件没有正确持久化检查_save是否在每次spend和earn后都会调用日志文件无限增长没有日志轮转配置RotatingFileHandler设置最大字节数和备份数量跑了一段时间后任务不执行预算达到红线检查can_afford判断逻辑和LOW_BUDGET_THRESHOLD阈值设置7.1 一个典型的死循环案例假设你把任务执行写成了while True但是忘了在循环体里判断余额那么模型调用会一直进行直到 API Key 欠费。这种事在真实项目里很常见。解决办法是在循环入口处强制检查while True: if not tracker.can_afford(single_task_max_cost): break # 执行任务这样即使前面的任务因为其他原因全部失败预算也不会清零。7.2 日志查错顺序如果系统出了问题不要直接看代码细节先按下面顺序排查检查agent.log中最后一次任务日志是成功还是失败。检查ledger.json中最近几笔花费是否合理。如果余额异常检查是否有“重复扣费”逻辑。如果任务调用失败检查 API Key 和模型名称是否正确。8. 最佳实践与工程建议8.1 预算控制永远是第一优先级我见过很多 AI 项目死在“失控调用成本”上。最有效的预防方法不是事后看账单而是在代码里强制设置三重上限全局总预算上限。单次任务成本上限。每日任务数量上限。这三层限制同时在代码里生效才能保证任何单点故障都不会造成严重资金损失。8.2 日志和账本要分离账本文件只记录“钱”的流动日志文件只记录“运行过程”。如果混在一起后面复盘时会很难查。上面的项目里我把它们放到了data/和logs/两个不同目录。8.3 使用模拟收益时要标注清楚在实验环境里你可以使用“模拟任务成功即获得收益”的逻辑。但这只是验证流程不是真实商业结果。当你接入真实业务时一定要把收益计算逻辑替换成平台流水、订单系统、支付回传等真实数据否则整个项目的结论没有参考价值。8.4 生产环境的安全与权限如果把这套系统部署到线上有几个安全项必须注意API Key 使用环境变量或密钥管理服务不要写进代码仓库。Agent 能访问的数据库和服务都使用最小权限账号。涉及删除、修改、发布等敏感操作前必须加人工确认步骤。在生产环境改动前先在测试环境完整跑通并做好数据备份。8.5 任务池的多样性单一任务会让实验结果有偏差。比如如果只做“写标语”这一件事那 AI 的收益能力就非常有限。更现实的做法是让 Agent 拥有多个技能每个技能对应一个任务模块这样才能验证“多个收入来源”是否能让它更快赚回订阅费。8.6 每周做一次输出质量抽查AI Agent 并不是“执行完就代表成功”。你需要定期抽查它的输出内容看是否真的能用于业务。如果输出质量持续下滑可能是提示词需要优化也可能是模型参数需要调整。9. 总结与下一步这个“给 AI 150 英镑和三个月时间”的项目核心并不是让 AI 真的变成赚钱机器而是训练你具备一套工程化思维怎么在有限预算下设计 AI 自动运行闭环怎么做成本控制怎么通过账本评价 AI 的真实价值。掌握这些能力后你就能把任意一个“智能客服”“内容生成器”“自动化运营脚本”改造成可监控、可预算、可回流的正式业务系统。下一步你可以继续尝试的方向有三个。第一把任务池换成你自己的真实业务接入订单系统或支付回调跑一轮真实的投入产出测试。第二给 Agent 增加一个简单的评估模块比如用“输出字数、语言流畅度、有效转化”给每次任务打分让调度器自动优先执行得分高的任务。第三把账本数据接入可视化平台比如 Grafana 或轻量级 Web 页面实时观察预算变化和任务成功率。动手跑一遍上面的代码比只看文章理解会深很多。如果你在运行中遇到报错也可以先把日志和账本文件重新打开看看大部分问题都能从这两份数据中找到线索。