“A daily estimation game where the crowds answer fights an AI”——这个标题看起来像是一个创意游戏但真正动手做的人第一反应往往是这不就是个“猜数字”的套壳吗如果你也这么想那这篇文章值得看完。它真正有意思的地方不在“猜”而在于它把两个成熟但很少被放到同一个系统里的东西强行放到了一起一边是人类的群体智慧crowd wisdom一边是大模型的概率推理。系统每天提出一道估算题人提交自己的答案AI也会给出自己的答案最后通过距离得分决定谁赢。这里有两个工程问题会被立刻暴露出来AI 的输出不稳定以及人类的答案分布极不均匀。如果你能把这两个问题处理干净这个游戏就能跑起来并且有真实的数据价值。本文会从玩法设计、系统架构、后端实现、AI 接入、任务调度、评分逻辑到上线后的常见坑完整走一遍。即使你最后不打算做游戏里面涉及的任务调度、AI 网关封装、评分引擎设计也可以直接移植到其他 AI 产品里。1. 这篇文章真正要解决的问题先说结论这个项目的核心不是“AI 有多聪明”而是“怎么样让一群人的答案持续稳定地和 AI 的答案做一次可验证的对抗”。日常估算题看起来简单比如“一张 A4 纸的厚度是多少毫米”“2024 年全球咖啡豆产量大概是多少万吨”。每个人都可以凭直觉给出一个数AI 也可以根据语料做一个估算。但一旦进入工程实现你会发现三个非常现实的问题第一答案不是一个点而是一个范围。人和 AI 给出的数值天然带有不确定性。游戏需要设计一个评分规则让“接近真实值”成为唯一目标但又不能因为某个人的极端离谱答案破坏整体体验。第二AI 不是免费的也不是想调就调。如果每天只出一道题直接调用大模型 API 成本可控但如果用户量大每个用户提交答案后都要实时向 AI 要一个“参考答案”成本就会失控。更合理的设计是每天在题目发布时就一次性让 AI 生成答案而不是在每个人提交时都调用一次。第三日更任务要可靠。不是只在开发机跑一次就完事而是要自动地在每天固定时刻出题、锁卷、评分、公布结果。这需要任务调度、状态锁定和时区处理。这篇文章就是围绕这三个问题展开。适合的人群包括想把 AI 创意做成真正可运行产品的开发者想学习 FastAPI 调度器 AI 网关怎么配合的读者以及想理解“如何度量人群智慧”的产品经理。2. 游戏机制与核心概念在写代码之前先把游戏规则讲清楚。只有规则清晰数据模型和评分逻辑才能稳定。2.1 每日估算题每天一个题目格式统一Question: 一个标准篮球的周长是多少厘米 Unit: cm Answer: 可在 [10, 100] 之间估算题目要满足几个条件答案是一个连续数值否则不好评分具有客观真实值否则无法判定胜负不能太专业否则群体智慧没有意义。2.2 人类答案与 AI 答案每个玩家可以提交一个估算值。系统在主题库中保存玩家的答案。AI 答案由题目发布时自动生成。系统要求大模型输出一个数值和一段简短推理。因为大模型对“客观数值”的直接预测经常不准所以更稳妥的方式是让模型先思考再输出一个区间再取中点。这个设计后面会讲。2.3 评分规则最简单也最公平的规则是相对误差error |player_answer - true_value| / true_value score round(100 * (1 - min(error, 2)), 1)也就是说答案误差越小得分越高。误差达到真实值的 200% 或以上得分为 0。AI 也按同样的规则评分。每天游戏结束时公布 AI 得分、人群中位数得分、每个玩家的得分并按“谁赢过 AI”标记胜负。2.4 群体智慧指标这里有个容易误解的点。很多人以为“ crowd 获胜”就是所有玩家中分数最高的那个人赢了。但更符合群体智慧理念的比较是把所有人当天答案的中位数当成“群体答案”再拿这个中位数去和 AI 的答案比较。这个设计很有意思单个玩家可能被 AI 碾压但几十个普通人的中位数往往非常接近真实值。这就是 Galton 在 1907 年发现的“群体智慧”现象。作为游戏也可以设置一个每日奖池或者一个“击败 AI 次数”的连续记录。在实际实现中我会同时计算两个指标group_median_score: 用玩家答案中位数计算的得分 ai_score: 用 AI 答案计算的得分group_median_score ai_score时当天判定“人群击败 AI”。2.5 日更状态机一个日更游戏本质上是一个状态机DRAFT - PUBLISHED - LOCKED - SCORED - ARCHIVEDDRAFT每日题目生成中。PUBLISHED玩家可以提交答案。LOCKED截止时间到锁定答案。SCORED开始评分生成分数和结果。ARCHIVED结果公布归档。这样设计的好处是一旦状态机清晰任务调度和前端展示都变得简单。前端只需要根据状态显示“距离截止还有多久”还是“查看今日结果”。3. 系统架构与技术选型按照最小可运行版本我们不需要复杂微服务推荐这样的架构HTML JavaScript (前端页面) | | REST API v FastAPI (后端) |-- SQLAlchemy ORM |-- PostgreSQL / SQLite (数据存储) |-- APScheduler (定时任务) |-- AI Provider (大模型 API 封装)技术选型理由FastAPI类型提示友好自带 OpenAPI 文档非常适合快速构建 API也方便前后端联调。SQLAlchemy PostgreSQL数据模型清晰事务管理方便。小项目也可以用 SQLite 起步但生产环境建议 PostgreSQL。APScheduler轻量级任务调度支持 cron 表达式可以每天固定时间触发。比 Celery 轻很多适合单机部署。大模型 API优先使用 OpenAI 兼容接口方便切换模型。比如可以是 gpt-4o-mini、DeepSeek、Qwen 等只要接口格式兼容。设计上要注意的一点是不要把 AI Provider 直接写死在业务代码里。因为大模型服务商随时可能变更用一个AIProvider类隔离外部依赖后续切换模型只需要改配置文件。4. 环境准备与项目初始化这里以 macOS / Linux 环境为例Windows 也基本适用。先准备 Python 3.10推荐 3.11。然后创建虚拟环境mkdir estimation-game cd estimation-game python3.11 -m venv venv source venv/bin/activate安装基础依赖pip install fastapi uvicorn sqlalchemy apscheduler pydantic pydantic-settings httpx如果使用 PostgreSQL还要装驱动pip install psycopg2-binary项目结构建议按功能拆分而不是把所有代码丢到一个文件estimation-game/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置项 │ ├── models.py # SQLAlchemy 数据模型 │ ├── schemas.py # Pydantic 数据模型 │ ├── ai_provider.py # AI 接口封装 │ ├── scoring.py # 评分逻辑 │ ├── scheduler.py # 定时任务 │ └── routers/ │ ├── __init__.py │ └── game.py # 游戏相关 API ├── templates/ │ └── index.html ├── requirements.txt └── .env这种结构看起来“多了一点”但长期维护会非常舒服。尤其是 AI 项目需求变化快拆得清楚才能改得快。5. 核心后端实现下面从数据模型开始逐步实现一个最小可运行版本。5.1 数据模型文件路径app/models.pyfrom datetime import datetime, date from sqlalchemy import String, Float, Date, DateTime, Integer from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class Question(Base): __tablename__ questions id: Mapped[int] mapped_column(primary_keyTrue) question_date: Mapped[date] mapped_column(Date, uniqueTrue, indexTrue) content: Mapped[str] mapped_column(String(512)) unit: Mapped[str] mapped_column(String(32), default) true_value: Mapped[float | None] mapped_column(Float, nullableTrue) status: Mapped[str] mapped_column(String(16), defaultDRAFT) # PUBLISHED / LOCKED / SCORED / ARCHIVED ai_answer: Mapped[float | None] mapped_column(Float, nullableTrue) ai_reasoning: Mapped[str | None] mapped_column(String(1024), nullableTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) class PlayerAnswer(Base): __tablename__ player_answers id: Mapped[int] mapped_column(primary_keyTrue) question_id: Mapped[int] mapped_column(Integer, indexTrue) player_name: Mapped[str] mapped_column(String(64)) answer: Mapped[float] mapped_column(Float) error: Mapped[float | None] mapped_column(Float, nullableTrue) score: Mapped[float | None] mapped_column(Float, nullableTrue) beat_ai: Mapped[bool | None] mapped_column(defaultNone) submitted_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow)这里的关键设计在于question_date使用日期作为唯一索引。每天一道题用日期检索非常高效也天然避免重复生成。PlayerAnswer记录玩家答案并在评分子阶段反填error、score和beat_ai。这样设计的好处是历史记录可以快速回看不需要重新算。5.2 AI Provider 封装文件路径app/ai_provider.pyimport os import json from typing import Any import httpx class AIProvider: 统一大模型 API 调用入口。 默认使用 OpenAI 兼容协议可通过环境变量切换。 def __init__(self, api_key: str | None None, base_url: str | None None, model: str | None None): self.api_key api_key or os.getenv(LLM_API_KEY, ) self.base_url base_url or os.getenv(LLM_BASE_URL, https://api.openai.com/v1) self.model model or os.getenv(LLM_MODEL, gpt-4o-mini) async def estimate(self, question: str) - tuple[float | None, str]: 调用模型估算答案。 返回 (预估值, 模型推理摘要)。 system_prompt ( 你是一个估算助手。用户会给你一个带有客观数值的问题。 请思考后给出一个最可能的数值不要做过多假设。 只输出 JSON{\value\: 数值, \reason\: \简要推理\} ) user_prompt f问题{question}\n请直接给出你估算的数值。 payload { model: self.model, messages: [ {role: system, content: system_prompt}, {role: user, content: user_prompt}, ], response_format: {type: json_object}, temperature: 0.2, } headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, } try: async with httpx.AsyncClient(timeout30) as client: resp await client.post( f{self.base_url}/chat/completions, headersheaders, jsonpayload, ) resp.raise_for_status() data resp.json() content data[choices][0][message][content] parsed json.loads(content) value float(parsed.get(value, 0)) reason parsed.get(reason, ) return value, reason except Exception as e: return None, fAI 调用失败: {e}这里有几个实际操作中很容易踩坑的地方必须检查response_format。对于“只输出数值”的需求直接把参数设为 JSON 模式能避免模型输出一堆解释。temperature要调低。估算任务需要稳定不建议超过 0.5。超时时间不能太短。部分模型在推理复杂问题时可能超过 10 秒建议 30 秒。不要在高频路径上同步调用。这里用的是async httpx在 FastAPI 中可以异步执行避免阻塞事件循环。更稳的方案是建模先输出一个区间再取中值。如果你接的模型支持 function calling也可以让模型输出{min: 100, max: 180}后取(min max) / 2。这样可以降低极端点的影响。5.3 评分逻辑文件路径app/scoring.pydef relative_error(answer: float, true_value: float) - float: if not true_value or true_value 0: return 2.0 # 没有真实值时直接打低分 return abs(answer - true_value) / true_value def score_from_error(error: float) - float: return round(100 * (1 - min(error, 2.0)), 1) def compute_rank(answers: list[float], true_value: float) - dict: 给定一组玩家答案和真实值计算 - 玩家中位数得分 - 中位数答案 if not answers: return {median_answer: None, median_score: 0.0} s sorted(answers) n len(s) median s[n // 2] if n % 2 1 else (s[n // 2 - 1] s[n // 2]) / 2 median_error relative_error(median, true_value) median_score score_from_error(median_error) return { median_answer: median, median_error: median_error, median_score: median_score, }为了让代码更好维护我单独抽出relative_error和score_from_error纯函数。任何人都能通过单元测试验证评分规则。5.4 每日任务调度文件路径app/scheduler.py这里使用 APScheduler 的 CronTrigger 实现每日任务。例子中我设每天 9:00 生成新题并调用 AI每天 21:00 锁卷并评分。from apscheduler.schedulers.asyncio import AsyncIOScheduler from datetime import date, datetime scheduler AsyncIOScheduler() async def publish_daily_question(): 每日 9:00 生成题目优先从题库选择当天题目并调用 AI 答案。 如果题目不存在则创建。 today date.today() async with SessionLocal() as session: q await get_or_create_question(session, today) if q.status DRAFT: ai_value, ai_reason await ai_provider.estimate(q.content) q.ai_answer ai_value q.ai_reasoning ai_reason q.status PUBLISHED await session.commit() # 实际发布动作推送、写入缓存等 async def lock_and_score(): 每日 21:00 锁定当前题目计算得分与胜负。 today date.today() async with SessionLocal() as session: q await get_today_question(session, today) if q and q.status PUBLISHED: rows (await session.execute( select(PlayerAnswer).where(PlayerAnswer.question_id q.id) )).scalars().all() if rows: group_result compute_rank([r.answer for r in rows], q.true_value or q.ai_answer) q.status SCORED await session.commit() def start_scheduler(): scheduler.add_job( publish_daily_question, triggercron, hour9, minute0, iddaily_publish, replace_existingTrue, ) scheduler.add_job( lock_and_score, triggercron, hour21, minute0, iddaily_score, replace_existingTrue, ) scheduler.start()这种“两个 cron 任务”的设计比“一个 cron 任务做完所有事”要可靠。因为题目发布和评分之间隔了 12 小时就算发布任务临时失败评分任务也不会误伤数据。关键坑点是时区。APScheduler 默认使用本机时区如果是部署到 Docker 容器要确保容器时区正确否则 21:00 实际可能是其他时间。建议在部署环境中统一设置为Asia/ShanghaiENV TZAsia/Shanghai5.5 FastAPI 入口与 API文件路径app/main.pyfrom datetime import date from fastapi import FastAPI, HTTPException, Depends from sqlalchemy import select from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession from pydantic import BaseModel from .models import Base, Question, PlayerAnswer from .scoring import relative_error, score_from_error from .scheduler import start_scheduler app FastAPI(titleEstimation Game API) # 这里以 SQLite 为例生产环境建议使用 PostgreSQL DATABASE_URL sqliteaiosqlite:///./estimation.db engine create_async_engine(DATABASE_URL) SessionLocal async_sessionmaker(engine, expire_on_commitFalse) app.on_event(startup) async def startup(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) start_scheduler() class AnswerIn(BaseModel): player_name: str answer: float app.get(/api/question/today) async def get_today_question(): today date.today() async with SessionLocal() as session: q (await session.execute( select(Question).where(Question.question_date today) )).scalar_one_or_none() if not q: raise HTTPException(status_code404, detail今日题目尚未发布) return { date: q.question_date.isoformat(), content: q.content, unit: q.unit, status: q.status, } app.post(/api/answer/submit) async def submit_answer(payload: AnswerIn): today date.today() async with SessionLocal() as session: q (await session.execute( select(Question).where(Question.question_date today) )).scalar_one_or_none() if not q: raise HTTPException(status_code404, detail今日题目不存在) if q.status ! PUBLISHED: raise HTTPException(status_code400, detail当前不在答题时间段) ans PlayerAnswer( question_idq.id, player_namepayload.player_name, answerpayload.answer, ) session.add(ans) await session.commit() return {ok: True, message: 答案已提交} app.get(/api/result/today) async def get_today_result(): today date.today() async with SessionLocal() as session: q (await session.execute( select(Question).where(Question.question_date today) )).scalar_one_or_none() if not q: raise HTTPException(status_code404, detail今日题目不存在) rows (await session.execute( select(PlayerAnswer).where(PlayerAnswer.question_id q.id) )).scalars().all() true_value q.true_value or q.ai_answer if not true_value: raise HTTPException(status_code400, detail今日真实值或 AI 答案未生成) ai_score score_from_error(relative_error(q.ai_answer, true_value)) # 玩家排名 player_results [] for r in rows: err relative_error(r.answer, true_value) score score_from_error(err) player_results.append({ player_name: r.player_name, answer: r.answer, error: round(err, 4), score: score, beat_ai: score ai_score, }) player_results.sort(keylambda x: x[score], reverseTrue) group_answers [r.answer for r in rows] avg_answer sum(group_answers) / len(group_answers) if group_answers else None avg_score score_from_error(relative_error(avg_answer, true_value)) if avg_answer else 0 return { date: today.isoformat(), question: q.content, ai_answer: q.ai_answer, ai_reasoning: q.ai_reasoning, ai_score: ai_score, group_avg_answer: avg_answer, group_avg_score: avg_score, crowd_beats_ai: avg_score ai_score, players: player_results[:20], }这一段把完整流程串起来。从启动时建表到每日调度再到用户提交和结果查询都能在这个文件里看到脉络。需要注意的地方前端即使只是简单页面也建议从/api/question/today获取题目而不是直接硬编码到 JS 里。提交接口要判断状态避免用户在 LOCKED 之后继续提交。如果没有true_value就把 AI 答案当作“基准值”用来评分。这会导致 AI 永远得满分不太公平所以更好的做法是提前在后台维护真实值或者用题目模板定义真实值来源。5.6 前端演示页面文件路径templates/index.html!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title每日估算人类 vs AI/title style body { font-family: PingFang SC, Microsoft YaHei, sans-serif; max-width: 720px; margin: 40px auto; padding: 0 16px; } .box { background: #f7f8fa; border-radius: 12px; padding: 24px; margin: 16px 0; } button { background: #2563eb; color: #fff; border: none; padding: 10px 18px; border-radius: 8px; font-size: 16px; cursor: pointer; } input { padding: 10px; border: 1px solid #ddd; border-radius: 8px; width: 240px; } .result { margin-top: 16px; white-space: pre-line; } /style /head body h1每日估算人类 vs AI/h1 div classbox idquestion-box p加载题目中……/p /div div classbox h3提交你的估算/h3 input idplayerName placeholder昵称 input idplayerAnswer typenumber placeholder你的答案 button onclicksubmitAnswer()提交/button /div div classbox h3今日结果/h3 div idresult-boxp结果尚未公布/p/div /div script async function loadQuestion() { const res await fetch(/api/question/today); if (res.ok) { const data await res.json(); document.getElementById(question-box).innerHTML pstrong${data.content}/strong/p p单位${data.unit}/p p状态${data.status}/p; } else { document.getElementById(question-box).innerHTML p题目尚未发布请稍后再来。/p; } } async function submitAnswer() { const name document.getElementById(playerName).value.trim(); const answer parseFloat(document.getElementById(playerAnswer).value); if (!name || isNaN(answer)) { alert(请填写昵称和有效数字); return; } const res await fetch(/api/answer/submit, { method: POST, headers: {Content-Type: application/json}, body: JSON.stringify({player_name: name, answer: answer}) }); const data await res.json(); alert(data.message || 提交失败); } async function loadResult() { const res await fetch(/api/result/today); if (!res.ok) return; const data await res.json(); let html ; html pAI 答案${data.ai_answer.toFixed(2)}得分${data.ai_score}/p; html p人群平均答案${data.group_avg_answer?.toFixed(2)}得分${data.group_avg_score}/p; html p${data.crowd_beats_ai ? 本日人群击败 AI : 本日 AI 胜过人群}/p; html ol data.players.slice(0, 5).map(p li${p.player_name}答案 ${p.answer}得分 ${p.score}/li ).join() /ol; document.getElementById(result-box).innerHTML html; } loadQuestion(); loadResult(); /script /body /html这个页面只做了最基础的三件事显示题目、提交答案、查看结果。没有引入前端框架没有构建工具直接放在 FastAPI 的 StaticFiles 下就能跑。6. 运行、测试与结果验证写完代码后启动开发服务uvicorn app.main:app --reload --port 8000启动后访问http://localhost:8000/docsFastAPI 会自动生成 Swagger 文档可以直接在页面上测试接口。手动验证步骤调GET /api/question/today看看今日题目是否存在。如果题目还是 DRAFT 状态可以手动触发一次发布任务或者等调度器运行。调POST /api/answer/submit用不同玩家名提交几个答案比如 25、30、35。把数据库中该题状态改成 SCORED或者手动执行评分逻辑。调GET /api/result/today确认返回的 AI 得分、人群平均得分、玩家排名符合预期。如果一切正常返回结果大概会长这样{ date: 2025-01-20, question: 一个标准篮球的周长是多少厘米, ai_answer: 76.5, ai_score: 87.2, group_avg_answer: 75.8, group_avg_score: 92.4, crowd_beats_ai: true, players: [ {player_name: Alice, answer: 75.0, error: 0.01, score: 98.7, beat_ai: true} ] }要验证调度任务可以在开发环境把时间调近一点scheduler.add_job( publish_daily_question, triggerdate, run_datedatetime.now() timedelta(seconds10), idtest_publish, )然后观察日志确认任务是否在指定时间触发。如果触发失败第一个排查点是AsyncIOScheduler是否在事件循环中启动。不要在和 FastAPI 同一个进程里用BackgroundScheduler否则会出现 asyncio 冲突。7. 常见问题与排查思路这类项目在开发中遇到的高频问题我整理成了一张表问题现象可能原因排查方式解决方案每天到点不出题调度器未启动或时区不对查看启动日志确认start_scheduler被调用检查容器时区在 startup 事件中启动调度器设置TZAsia/ShanghaiAI 返回值为 0 或明显异常大模型没有按 JSON 输出或 prompt 不明确直接打印 AI 原始响应在网页上单独测试一次完整生成调整 system prompt开启response_format: json_object设置temperature到 0.2 以下玩家提交答案后报 400当前状态不是 PUBLISHED或题目未创建查数据库里questions.status查前端传入日期确认题目状态机流转正常对状态修改操作增加手动管理工具评分为 0 或 NaNtrue_value为空或答案中混入了字符串检查questions.true_value检查PlayerAnswer.answer的类型评分前统一校验数据对缺失真实值的题目跳过评分同一玩家重复提交数据模型没有唯一约束查询player_answers表记录增加UniqueConstraint(question_id, player_name)定时评分把所有玩家分数都算成一样评分时用了同一份true_value但错误调用了list的引用检查代码块中是否多次使用同一个error变量用单元测试覆盖多人答案场景这里最容易被忽略的是“真实值”的获取。很多题目其实没有公开权威数值这时如果直接用 AI 答案当真实值就会变成“AI 自己出题自己评卷”游戏公平性大打折扣。建议维护一个人工审核后台或者在题目表里增加一个source字段用来记录真实值来源source: Mapped[str | None] mapped_column(String(256), nullableTrue)每次出题时运营者可以在发布前填入真实值来源比如某本参考书、某个统计网站或者自己计算。宁可少出也不要用虚假基准。8. 最佳实践与工程建议8.1 把 AI 调用做成可 mock 的模块在开发和测试阶段不能每次都真实调用大模型 API。建议为AIProvider增加一个 fake 实现或者用依赖注入替换if settings.LLM_PROVIDER fake: from .ai_provider import FakeAIProvider ai_provider FakeAIProvider() else: from .ai_provider import AIProvider ai_provider AIProvider()这样单元测试能快速跑通不会因为外部 API 故障导致测试失败。8.2 控制每日任务失败后的恢复逻辑调度器一旦错过任务比如服务器夜间重启第二天早上就不会自动出题。建议增加一个“catch-up”逻辑启动时检查今天是否已有 PUBLISHED 状态题目如果没有则立即执行发布任务。实现很简单app.on_event(startup) async def startup_catchup(): today date.today() async with SessionLocal() as session: q await get_today_question(session, today) if not q: await publish_daily_question()这比单纯依赖 cron 更健壮。8.3 评分公式要提前固化评分公式是游戏的核心规则一旦上线就不要轻易改。建议把公式写进配置或者至少在文档里注明版本。不要在数据库里只用score字段存结果却不记录公式版本。否则以后要改规则时历史分数无法对齐。一个可行的做法是在questions表增加scoring_version字段scoring_version: Mapped[str] mapped_column(String(16), defaultv1)8.4 防止刷榜和重复提交最简单的防刷策略提交答案前校验玩家昵称长度不允许大于 20 字符。每个玩家每天只能提交一次使用(question_id, player_name)唯一约束。限制提交时间窗口过期直接返回 400。如果有账号系统提交时用user_id而不是昵称作为唯一键。如果后续要做排行榜还需要考虑是不是允许 UU 号、IP 限流等。不过 MVP 阶段不需要过度设计。8.5 对 AI 答案做缓存虽然设计上每天只调用一次 AI但为了避免调试时重复调用浪费额度建议在模型中直接持久化ai_answer和ai_reasoning。在开发环境还可以增加一个.env配置ALWAYS_FETCH_AIFalse当这个值为 False且数据库中已有ai_answer时直接使用历史数据不再请求 API。8.6 日志与监控这类每日任务项目最重要的监控项有三个每天是否有新题目发布AI 调用是否成功评分任务是否有未处理异常建议至少接入一个简单的日志格式import logging logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(name)s %(message)s) logger logging.getLogger(estimation)在publish_daily_question和lock_and_score中加日志logger.info(question %s published, ai_answer%s, q.id, q.ai_answer)哪怕不接 Sentry有日志也能快速定位。8.7 数据备份每日题目和玩家答案量级不大但都是用户产出的数据建议数据库做每日备份。PostgreSQL 下最简单的方式pg_dump estimation_game backup_$(date %Y%m%d).sql配合 crontab 即可。9. 总结与后续学习方向这个项目真正训练的不是“让 AI 猜得更准”而是“如何把一个 AI 创意变成一个日更稳定的 Web 服务”。你会遇到状态管理、任务调度、外部 API 依赖、公平性设计、防刷策略等一系列问题。这些比调用一个 API 本身更能体现工程能力。后续如果你想继续扩展有四个方向值得尝试第一引入群组对战。让一个房间内的玩家答案均值和 AI 比而不是全站所有人混在一起比游戏性会更强。第二加入历史题库和难度曲线。每天出题可以按星期几区分不同难度比如周一是生活常识类周五是科技数据类。第三把 AI 的答案从“一个点”改成“一个置信区间”。这样可以把“距离”得分变成“区间是否包含真实值”的胜负判断更直观。第四倒过来让玩家预测 AI 答案。每天系统先展示 AI 的估算玩家要猜 AI 会给出什么数然后接近 AI 答案的人获胜。这个玩法成本更低也更容易引发社交传播。对于想深入学习的人建议从三块入手FastAPI 的异步任务和依赖注入、APScheduler 的 cron 底层原理、以及“群体智慧”在数据科学中的经典实验。这个方向看似小众其实把 AI 应用、数据工程和游戏化设计全部串起来了。如果你把这个项目完整部署上线你会发现它比很多“看上去酷炫”的 AI 玩具更具长期运营价值。建议收藏这篇文章然后从第 4 节的代码开始跑通一个最小版本。半小时内你就能看到第一个“人类 vs AI”的得分页面。