1. 生产环境没告警我用 AI 把监控平台跑起来了应用监控平台这个词听起来挺唬人说白了就是你的服务挂了、接口变慢了、错误率飙了它能第一时间告诉你而不是等用户来投诉。我之前的处境很典型——几个小服务跑在云主机上日志靠tail -f出问题靠用户截图半夜被电话叫醒是常态。想上 Prometheus Grafana 那套吧配置量大、组件多对个人开发者来说维护成本比业务本身还高。后来我换了个思路不追求大而全只做「上报 → 存储 → 看板 → 告警」这条最小闭环采集端用 Python FastAPI看板用 Vue 3接口和告警逻辑交给 AI 生成我负责提需求和验收。整个过程我几乎没手写业务代码但每一步都能跑通、能验证。这篇文章就把这套可复制的做法拆开讲包括统一 Key 怎么配、FastAPI 采集路由怎么写、Vue 3 看板怎么起、以及一次从上报到告警触发的完整验证。适合谁看个人开发者、运维新手、想给自己项目加一层轻量监控但不想引入重型栈的人。你不需要精通前端也不需要会写复杂后端跟着步骤走就行。核心检索词就三个AI 应用监控平台、Python FastAPI 采集、Vue 3 看板。下面所有配置和命令我都实测过直接抄能跑。先说清楚这套平台的边界避免你期待错位。它不做分布式链路追踪不做 APM 级别的火焰图它解决的是「我的服务现在活着吗、错误多不多、响应慢不慢、出问题能不能通知我」这四个问题。数据模型也很简单一条上报记录包含应用名、指标名、数值、时间戳、标签。告警规则就是「某应用某指标在 N 分钟内超过阈值就触发」。简单但够用而且生产环境跑到现在没掉过链子。我踩过的坑主要集中在前端联调和告警去重上一开始看板每 5 秒全量刷新数据一多浏览器就卡告警没做冷却同一个错误一分钟发了二十条通知。这些后面都会讲到怎么处理。你如果只想先跑起来按顺序做就行细节排障放在第 5 节。2. TaoToken 统一 Key 前置准备一个 Key 管住所有模型调用这套平台里 AI 参与的部分有两块一是生成 FastAPI 接口和告警判断逻辑二是在运行时用模型做告警内容的归纳比如把一堆错误日志压成一句人话。如果每接一个模型就换一套 Key、换一个 Base URL配置会散得到处都是。所以我用 TaoToken 做统一入口一个 Key、一个 Base URL模型 ID 按需切换。TaoToken 在这里的角色是模型调用的统一网关你拿到一个 API Key把 Base URL 指向https://taotoken.net/api然后在请求里指定 Model ID 就行。对监控平台来说好处是告警归纳、日志摘要、接口生成这几处调用可以共用同一套凭证和客户端封装换模型只改一个字符串。前置准备分三步。第一步去控制台创建 API Key。打开 https://taotoken.net/console 登录后在 API Keys 页面新建一个复制出来保存好这个 Key 只显示一次。第二步确认你要用的模型 ID。不同任务对模型要求不一样生成代码用能力强的告警文本归纳用响应快的具体可用列表在文档里查 https://taotoken.net/doc 。第三步把 Key 写进环境变量别硬编码进代码。# Linux / macOS写进 ~/.bashrc 或部署时的环境变量 export TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api# Windows PowerShell $env:TAOTOKEN_API_KEYsk-你的Key $env:TAOTOKEN_BASE_URLhttps://taotoken.net/api这里有个关键点Base URL 是https://taotoken.net/api不要自己加/v1之类的后缀具体路径由 SDK 或请求体决定。如果你用的是 OpenAI 兼容的客户端通常只需要把base_url指过去、api_key填上其余不变。注意Key 属于敏感凭证不要提交到 Git不要写进前端代码。前端永远不直接调模型所有模型调用都放在 FastAPI 后端前端只调你自己的接口。我建议在项目根目录建一个.env文件用python-dotenv加载这样本地和线上配置一致# .env TAOTOKEN_API_KEYsk-你的Key TAOTOKEN_BASE_URLhttps://taotoken.net/api ALERT_MODEL你的模型ID然后在代码里统一读环境变量。这样做的另一个好处是以后要换模型或换 Key只改.env不用翻代码。对于监控平台这种要长期跑的服务配置和代码分离是基本纪律。如果你还想让 AI 帮你写代码、做 Agent 类的长期任务可以了解下 Coding Plan它更适合持续性的编码场景https://taotoken.net/coding-plan 。单纯做模型对话验证的话用模型对话页就行https://taotoken.net/models 。先把 Key 和 Base URL 这两样准备好后面所有步骤都依赖它们。3. 可复制配置FastAPI 采集路由 Vue 3 看板启动这一节是全文的核心给你能直接抄的配置和代码。整体结构后端 FastAPI 提供/report上报接口和/metrics查询接口数据先落 SQLite个人项目够用想换 PostgreSQL 改连接串即可前端 Vue 3 用 Vite 起定时拉/metrics渲染看板告警逻辑在后端上报时判断阈值触发就调模型归纳并推送。先看后端项目结构api-monitor/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── db.py # SQLite 连接与建表 │ ├── models.py # 数据模型 │ ├── alert.py # 告警判断 模型归纳 │ └── llm.py # TaoToken 客户端封装 ├── .env └── requirements.txtrequirements.txtfastapi0.115.0 uvicorn[standard]0.30.6 httpx0.27.2 python-dotenv1.0.1 pydantic2.9.2app/llm.py统一封装 TaoToken 调用所有模型请求都走这里import os import httpx from dotenv import load_dotenv load_dotenv() API_KEY os.getenv(TAOTOKEN_API_KEY) BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) MODEL_ID os.getenv(ALERT_MODEL) async def summarize_alert(raw_text: str) - str: 把原始告警文本交给模型归纳成一句话 url f{BASE_URL}/v1/chat/completions headers { Authorization: fBearer {API_KEY}, Content-Type: application/json, } payload { model: MODEL_ID, messages: [ {role: system, content: 你是运维助手把告警压缩成一句中文保留应用名和关键数值。}, {role: user, content: raw_text}, ], temperature: 0.2, } async with httpx.AsyncClient(timeout20) as client: resp await client.post(url, headersheaders, jsonpayload) resp.raise_for_status() data resp.json() return data[choices][0][message][content].strip()app/db.py建表并暴露连接import sqlite3 DB_PATH monitor.db def init_db(): conn sqlite3.connect(DB_PATH) conn.execute( CREATE TABLE IF NOT EXISTS metrics ( id INTEGER PRIMARY KEY AUTOINCREMENT, app TEXT NOT NULL, metric TEXT NOT NULL, value REAL NOT NULL, ts INTEGER NOT NULL ) ) conn.execute( CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, app TEXT NOT NULL, metric TEXT NOT NULL, value REAL NOT NULL, message TEXT, ts INTEGER NOT NULL ) ) conn.commit() conn.close() def get_conn(): return sqlite3.connect(DB_PATH)app/alert.py阈值判断加冷却去重避免告警风暴import time from app.llm import summarize_alert # 规则应用 - 指标 - 阈值 RULES { order-service: {error_rate: 5.0, latency_ms: 800.0}, user-service: {error_rate: 3.0}, } # 冷却同一应用同一指标 60 秒内只告警一次 _last_alert {} COOLDOWN 60 async def check_and_alert(app: str, metric: str, value: float) - str | None: rule RULES.get(app, {}) threshold rule.get(metric) if threshold is None or value threshold: return None key f{app}:{metric} now int(time.time()) if now - _last_alert.get(key, 0) COOLDOWN: return None _last_alert[key] now raw f应用 {app} 的指标 {metric} 当前值 {value}超过阈值 {threshold} message await summarize_alert(raw) return messageapp/main.py采集路由和查询路由import time from fastapi import FastAPI from pydantic import BaseModel from app.db import init_db, get_conn from app.alert import check_and_alert app FastAPI(titleAPI Monitor) init_db() class ReportIn(BaseModel): app: str metric: str value: float app.post(/report) async def report(item: ReportIn): ts int(time.time()) conn get_conn() conn.execute( INSERT INTO metrics (app, metric, value, ts) VALUES (?, ?, ?, ?), (item.app, item.metric, item.value, ts), ) conn.commit() conn.close() alert_msg await check_and_alert(item.app, item.metric, item.value) if alert_msg: conn get_conn() conn.execute( INSERT INTO alerts (app, metric, value, message, ts) VALUES (?, ?, ?, ?, ?), (item.app, item.metric, item.value, alert_msg, ts), ) conn.commit() conn.close() return {ok: True, alert: alert_msg} return {ok: True} app.get(/metrics) def metrics(app: str | None None, limit: int 100): conn get_conn() if app: rows conn.execute( SELECT app, metric, value, ts FROM metrics WHERE app? ORDER BY ts DESC LIMIT ?, (app, limit), ).fetchall() else: rows conn.execute( SELECT app, metric, value, ts FROM metrics ORDER BY ts DESC LIMIT ?, (limit,), ).fetchall() conn.close() return [{app: r[0], metric: r[1], value: r[2], ts: r[3]} for r in rows] app.get(/alerts) def alerts(limit: int 50): conn get_conn() rows conn.execute( SELECT app, metric, value, message, ts FROM alerts ORDER BY ts DESC LIMIT ?, (limit,), ).fetchall() conn.close() return [{app: r[0], metric: r[1], value: r[2], message: r[3], ts: r[4]} for r in rows]启动后端pip install -r requirements.txt uvicorn app.main:app --host 0.0.0.0 --port 8000前端用 Vite 起 Vue 3npm create vitelatest monitor-web -- --template vue cd monitor-web npm install npm install axios npm run devsrc/App.vue里拉数据渲染核心逻辑script setup import { ref, onMounted, onUnmounted } from vue import axios from axios const metrics ref([]) const alerts ref([]) let timer null async function load() { const m await axios.get(http://127.0.0.1:8000/metrics?limit50) metrics.value m.data const a await axios.get(http://127.0.0.1:8000/alerts?limit20) alerts.value a.data } onMounted(() { load() timer setInterval(load, 5000) }) onUnmounted(() clearInterval(timer)) /script template div classboard h2指标看板/h2 table trth应用/thth指标/thth数值/thth时间/th/tr tr v-for(m, i) in metrics :keyi td{{ m.app }}/tdtd{{ m.metric }}/td td{{ m.value }}/td td{{ new Date(m.ts * 1000).toLocaleTimeString() }}/td /tr /table h2告警记录/h2 ul li v-for(a, i) in alerts :keyi [{{ a.app }}] {{ a.message }} /li /ul /div /template生产部署时前端npm run build出静态文件用 Nginx 托管后端用 systemd 或 supervisor 常驻。这套配置我实测能直接跑SQLite 单文件备份就是复制文件对个人项目非常友好。4. 验证请求从上报到告警触发的完整动作配置写完必须验证不然你不知道是接口通了还是模型没调上。这一节给你一条完整的验证链路照着敲就能看到告警被触发。第一步确认后端起来了。访问http://127.0.0.1:8000/docs能看到 FastAPI 自动生成的接口文档说明服务正常。第二步上报一条正常数据验证采集路由curl -X POST http://127.0.0.1:8000/report \ -H Content-Type: application/json \ -d {app:order-service,metric:latency_ms,value:120}预期返回{ok:true}没有 alert 字段因为 120 没超过阈值 800。第三步上报一条超阈值数据触发告警curl -X POST http://127.0.0.1:8000/report \ -H Content-Type: application/json \ -d {app:order-service,metric:latency_ms,value:1500}预期返回类似{ok: true, alert: order-service 的 latency_ms 达到 1500超过阈值 800请检查服务响应。}这条 alert 文本就是模型归纳出来的说明 TaoToken 调用链路通了。如果这里返回的 alert 是空的或者报错去第 5 节对照排查。第四步查询告警记录确认落库curl http://127.0.0.1:8000/alerts?limit5能看到刚才那条告警包含 app、metric、value、message、ts。第五步打开前端看板http://127.0.0.1:51735 秒内应该能看到指标表格里出现刚才两条数据告警列表里出现那条归纳后的告警。如果前端没数据先看浏览器控制台有没有跨域报错后端加 CORS 中间件即可from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins[http://127.0.0.1:5173], allow_methods[*], allow_headers[*], )第六步验证冷却去重。连续快速上报三次超阈值数据for i in 1 2 3; do curl -s -X POST http://127.0.0.1:8000/report \ -H Content-Type: application/json \ -d {app:order-service,metric:latency_ms,value:2000} echo done预期只有第一次返回 alert后两次返回{ok:true}因为 60 秒冷却生效。这一步很关键生产环境没有冷却的话一个持续故障会把你通知渠道刷爆。到这一步整条链路验证完毕上报 → 存储 → 阈值判断 → 模型归纳 → 落库 → 看板展示。你可以把上报逻辑接到真实业务里比如在 FastAPI 中间件里统计每个请求的耗时和状态码定时上报或者在定时任务里采集 CPU、内存。采集端就是一个 HTTP 请求任何语言都能发。5. 常见报错排查401、local proxy failed、reading choices、OAuth这一节按真实报错来都是我实际遇到过的。你对照错误信息找对应条目。401 Unauthorized。最常见原因是 Key 没读到或写错。检查三点.env里TAOTOKEN_API_KEY是否以sk-开头且没有多余空格代码里load_dotenv()是否在读取环境变量之前执行部署到服务器时环境变量是否真的注入了systemd要在 service 文件里写Environment或EnvironmentFile。还有一种情况是 Key 被复制时带了换行用echo $TAOTOKEN_API_KEY | wc -c看长度对不对。local proxy failed / connection refused。这个报错通常出现在请求根本没发出去的时候说明本机网络层或客户端配置有问题。检查BASE_URL是不是写成了https://taotoken.net/api有没有多写路径检查本机是否能正常访问外网如果你在容器里跑确认容器网络正常。注意不要配置任何来路不明的网络工具直接用系统默认网络即可。reading choices of undefined。这个报错说明响应体里没有choices字段通常是请求体格式不对或模型 ID 写错。检查payload里model字段是不是你实际可用的模型 IDmessages是不是标准格式。可以先把响应体打印出来看resp await client.post(url, headersheaders, jsonpayload) print(resp.status_code, resp.text)如果返回的是错误 JSON里面一般会写明原因比如模型不存在、参数不合法。把resp.text贴给 AI 让它帮你判断比盲猜快得多。OAuth / authentication 相关报错。如果你用的是某些 CLI 工具比如 Claude Code 这类它可能走的是 OAuth 登录流程而不是 API Key。这时候要确认工具支持用 Base URL API Key 的方式接入把三件套配全Base URL 填https://taotoken.net/apiKey 填你的sk-KeyModel ID 填你要用的模型。三者缺一不可只填两个通常会报认证失败。具体接入方式看文档 https://taotoken.net/doc 。告警不触发。先确认阈值规则RULES里有没有对应应用和指标键名要完全一致再确认上报的 value 确实大于阈值最后看冷却时间可能上一次告警还在 60 秒内。调试时把COOLDOWN临时改成 0。前端跨域。浏览器控制台报 CORS 相关错误按第 4 节的 CORS 中间件配置allow_origins填你前端实际地址。生产环境前端和后端同域的话用 Nginx 反代/api到后端就不需要 CORS 了。数据不落库。检查monitor.db文件是否有写权限容器部署时注意挂载卷否则重启数据就没了。SQLite 并发写有限个人项目够用量大了换 PostgreSQL改db.py里的连接即可。排查的通用心法先看 HTTP 状态码再看响应体原文最后看服务端日志。90% 的问题在响应体里就写清楚了别急着改代码。6. 把监控接进真实业务下一步怎么走平台跑起来只是开始真正有价值的是把它接进你的业务。最省事的做法是在 FastAPI 里加一个中间件统计每个请求的耗时和状态码按分钟聚合后上报import time import httpx from fastapi import Request app.middleware(http) async def collect_metrics(request: Request, call_next): start time.time() response await call_next(request) cost_ms (time.time() - start) * 1000 async with httpx.AsyncClient() as client: await client.post(http://127.0.0.1:8000/report, json{ app: order-service, metric: latency_ms, value: cost_ms, }) return response注意别把上报接口自己也监控了否则会递归。加个路径判断跳过/report。告警通知渠道可以扩展邮件用smtplib企业微信或钉钉用 webhook短信接云厂商 SDK。模型归纳出来的那句话直接作为通知正文比原始数值可读性好很多。你还可以让模型根据告警内容给出初步排查建议比如「latency_ms 飙升通常是数据库慢查询或下游超时建议先看慢日志」。看板可以加趋势图用 ECharts 按时间轴画折线比表格直观。数据量大了给metrics表的ts字段加索引查询会快很多。长期来看如果你想让 AI 持续帮你迭代这个平台——加功能、改 UI、修 bug——用 Coding Plan 会比单次对话更顺上下文能保持住https://taotoken.net/coding-plan 。需要新建更多 Key 或管理配额去控制台https://taotoken.net/console 。接入细节和参数说明都在文档里https://taotoken.net/doc 。想先验证模型输出效果用模型对话页试几句https://taotoken.net/models 。这套东西最大的价值不是技术多复杂而是它真的能跑在生产环境、真的会在出问题时提醒你。我把它跑起来之后半夜被叫醒的次数明显少了因为很多问题在变成事故之前就收到了告警。你先按第 3 节的配置跑通再按第 4 节验证一遍剩下的就是接到你自己的业务里慢慢加规则、加渠道。
零代码上线生产：用 TaoToken 统一 Key 搭一个 AI 应用监控平台
1. 生产环境没告警我用 AI 把监控平台跑起来了应用监控平台这个词听起来挺唬人说白了就是你的服务挂了、接口变慢了、错误率飙了它能第一时间告诉你而不是等用户来投诉。我之前的处境很典型——几个小服务跑在云主机上日志靠tail -f出问题靠用户截图半夜被电话叫醒是常态。想上 Prometheus Grafana 那套吧配置量大、组件多对个人开发者来说维护成本比业务本身还高。后来我换了个思路不追求大而全只做「上报 → 存储 → 看板 → 告警」这条最小闭环采集端用 Python FastAPI看板用 Vue 3接口和告警逻辑交给 AI 生成我负责提需求和验收。整个过程我几乎没手写业务代码但每一步都能跑通、能验证。这篇文章就把这套可复制的做法拆开讲包括统一 Key 怎么配、FastAPI 采集路由怎么写、Vue 3 看板怎么起、以及一次从上报到告警触发的完整验证。适合谁看个人开发者、运维新手、想给自己项目加一层轻量监控但不想引入重型栈的人。你不需要精通前端也不需要会写复杂后端跟着步骤走就行。核心检索词就三个AI 应用监控平台、Python FastAPI 采集、Vue 3 看板。下面所有配置和命令我都实测过直接抄能跑。先说清楚这套平台的边界避免你期待错位。它不做分布式链路追踪不做 APM 级别的火焰图它解决的是「我的服务现在活着吗、错误多不多、响应慢不慢、出问题能不能通知我」这四个问题。数据模型也很简单一条上报记录包含应用名、指标名、数值、时间戳、标签。告警规则就是「某应用某指标在 N 分钟内超过阈值就触发」。简单但够用而且生产环境跑到现在没掉过链子。我踩过的坑主要集中在前端联调和告警去重上一开始看板每 5 秒全量刷新数据一多浏览器就卡告警没做冷却同一个错误一分钟发了二十条通知。这些后面都会讲到怎么处理。你如果只想先跑起来按顺序做就行细节排障放在第 5 节。2. TaoToken 统一 Key 前置准备一个 Key 管住所有模型调用这套平台里 AI 参与的部分有两块一是生成 FastAPI 接口和告警判断逻辑二是在运行时用模型做告警内容的归纳比如把一堆错误日志压成一句人话。如果每接一个模型就换一套 Key、换一个 Base URL配置会散得到处都是。所以我用 TaoToken 做统一入口一个 Key、一个 Base URL模型 ID 按需切换。TaoToken 在这里的角色是模型调用的统一网关你拿到一个 API Key把 Base URL 指向https://taotoken.net/api然后在请求里指定 Model ID 就行。对监控平台来说好处是告警归纳、日志摘要、接口生成这几处调用可以共用同一套凭证和客户端封装换模型只改一个字符串。前置准备分三步。第一步去控制台创建 API Key。打开 https://taotoken.net/console 登录后在 API Keys 页面新建一个复制出来保存好这个 Key 只显示一次。第二步确认你要用的模型 ID。不同任务对模型要求不一样生成代码用能力强的告警文本归纳用响应快的具体可用列表在文档里查 https://taotoken.net/doc 。第三步把 Key 写进环境变量别硬编码进代码。# Linux / macOS写进 ~/.bashrc 或部署时的环境变量 export TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api# Windows PowerShell $env:TAOTOKEN_API_KEYsk-你的Key $env:TAOTOKEN_BASE_URLhttps://taotoken.net/api这里有个关键点Base URL 是https://taotoken.net/api不要自己加/v1之类的后缀具体路径由 SDK 或请求体决定。如果你用的是 OpenAI 兼容的客户端通常只需要把base_url指过去、api_key填上其余不变。注意Key 属于敏感凭证不要提交到 Git不要写进前端代码。前端永远不直接调模型所有模型调用都放在 FastAPI 后端前端只调你自己的接口。我建议在项目根目录建一个.env文件用python-dotenv加载这样本地和线上配置一致# .env TAOTOKEN_API_KEYsk-你的Key TAOTOKEN_BASE_URLhttps://taotoken.net/api ALERT_MODEL你的模型ID然后在代码里统一读环境变量。这样做的另一个好处是以后要换模型或换 Key只改.env不用翻代码。对于监控平台这种要长期跑的服务配置和代码分离是基本纪律。如果你还想让 AI 帮你写代码、做 Agent 类的长期任务可以了解下 Coding Plan它更适合持续性的编码场景https://taotoken.net/coding-plan 。单纯做模型对话验证的话用模型对话页就行https://taotoken.net/models 。先把 Key 和 Base URL 这两样准备好后面所有步骤都依赖它们。3. 可复制配置FastAPI 采集路由 Vue 3 看板启动这一节是全文的核心给你能直接抄的配置和代码。整体结构后端 FastAPI 提供/report上报接口和/metrics查询接口数据先落 SQLite个人项目够用想换 PostgreSQL 改连接串即可前端 Vue 3 用 Vite 起定时拉/metrics渲染看板告警逻辑在后端上报时判断阈值触发就调模型归纳并推送。先看后端项目结构api-monitor/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── db.py # SQLite 连接与建表 │ ├── models.py # 数据模型 │ ├── alert.py # 告警判断 模型归纳 │ └── llm.py # TaoToken 客户端封装 ├── .env └── requirements.txtrequirements.txtfastapi0.115.0 uvicorn[standard]0.30.6 httpx0.27.2 python-dotenv1.0.1 pydantic2.9.2app/llm.py统一封装 TaoToken 调用所有模型请求都走这里import os import httpx from dotenv import load_dotenv load_dotenv() API_KEY os.getenv(TAOTOKEN_API_KEY) BASE_URL os.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) MODEL_ID os.getenv(ALERT_MODEL) async def summarize_alert(raw_text: str) - str: 把原始告警文本交给模型归纳成一句话 url f{BASE_URL}/v1/chat/completions headers { Authorization: fBearer {API_KEY}, Content-Type: application/json, } payload { model: MODEL_ID, messages: [ {role: system, content: 你是运维助手把告警压缩成一句中文保留应用名和关键数值。}, {role: user, content: raw_text}, ], temperature: 0.2, } async with httpx.AsyncClient(timeout20) as client: resp await client.post(url, headersheaders, jsonpayload) resp.raise_for_status() data resp.json() return data[choices][0][message][content].strip()app/db.py建表并暴露连接import sqlite3 DB_PATH monitor.db def init_db(): conn sqlite3.connect(DB_PATH) conn.execute( CREATE TABLE IF NOT EXISTS metrics ( id INTEGER PRIMARY KEY AUTOINCREMENT, app TEXT NOT NULL, metric TEXT NOT NULL, value REAL NOT NULL, ts INTEGER NOT NULL ) ) conn.execute( CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, app TEXT NOT NULL, metric TEXT NOT NULL, value REAL NOT NULL, message TEXT, ts INTEGER NOT NULL ) ) conn.commit() conn.close() def get_conn(): return sqlite3.connect(DB_PATH)app/alert.py阈值判断加冷却去重避免告警风暴import time from app.llm import summarize_alert # 规则应用 - 指标 - 阈值 RULES { order-service: {error_rate: 5.0, latency_ms: 800.0}, user-service: {error_rate: 3.0}, } # 冷却同一应用同一指标 60 秒内只告警一次 _last_alert {} COOLDOWN 60 async def check_and_alert(app: str, metric: str, value: float) - str | None: rule RULES.get(app, {}) threshold rule.get(metric) if threshold is None or value threshold: return None key f{app}:{metric} now int(time.time()) if now - _last_alert.get(key, 0) COOLDOWN: return None _last_alert[key] now raw f应用 {app} 的指标 {metric} 当前值 {value}超过阈值 {threshold} message await summarize_alert(raw) return messageapp/main.py采集路由和查询路由import time from fastapi import FastAPI from pydantic import BaseModel from app.db import init_db, get_conn from app.alert import check_and_alert app FastAPI(titleAPI Monitor) init_db() class ReportIn(BaseModel): app: str metric: str value: float app.post(/report) async def report(item: ReportIn): ts int(time.time()) conn get_conn() conn.execute( INSERT INTO metrics (app, metric, value, ts) VALUES (?, ?, ?, ?), (item.app, item.metric, item.value, ts), ) conn.commit() conn.close() alert_msg await check_and_alert(item.app, item.metric, item.value) if alert_msg: conn get_conn() conn.execute( INSERT INTO alerts (app, metric, value, message, ts) VALUES (?, ?, ?, ?, ?), (item.app, item.metric, item.value, alert_msg, ts), ) conn.commit() conn.close() return {ok: True, alert: alert_msg} return {ok: True} app.get(/metrics) def metrics(app: str | None None, limit: int 100): conn get_conn() if app: rows conn.execute( SELECT app, metric, value, ts FROM metrics WHERE app? ORDER BY ts DESC LIMIT ?, (app, limit), ).fetchall() else: rows conn.execute( SELECT app, metric, value, ts FROM metrics ORDER BY ts DESC LIMIT ?, (limit,), ).fetchall() conn.close() return [{app: r[0], metric: r[1], value: r[2], ts: r[3]} for r in rows] app.get(/alerts) def alerts(limit: int 50): conn get_conn() rows conn.execute( SELECT app, metric, value, message, ts FROM alerts ORDER BY ts DESC LIMIT ?, (limit,), ).fetchall() conn.close() return [{app: r[0], metric: r[1], value: r[2], message: r[3], ts: r[4]} for r in rows]启动后端pip install -r requirements.txt uvicorn app.main:app --host 0.0.0.0 --port 8000前端用 Vite 起 Vue 3npm create vitelatest monitor-web -- --template vue cd monitor-web npm install npm install axios npm run devsrc/App.vue里拉数据渲染核心逻辑script setup import { ref, onMounted, onUnmounted } from vue import axios from axios const metrics ref([]) const alerts ref([]) let timer null async function load() { const m await axios.get(http://127.0.0.1:8000/metrics?limit50) metrics.value m.data const a await axios.get(http://127.0.0.1:8000/alerts?limit20) alerts.value a.data } onMounted(() { load() timer setInterval(load, 5000) }) onUnmounted(() clearInterval(timer)) /script template div classboard h2指标看板/h2 table trth应用/thth指标/thth数值/thth时间/th/tr tr v-for(m, i) in metrics :keyi td{{ m.app }}/tdtd{{ m.metric }}/td td{{ m.value }}/td td{{ new Date(m.ts * 1000).toLocaleTimeString() }}/td /tr /table h2告警记录/h2 ul li v-for(a, i) in alerts :keyi [{{ a.app }}] {{ a.message }} /li /ul /div /template生产部署时前端npm run build出静态文件用 Nginx 托管后端用 systemd 或 supervisor 常驻。这套配置我实测能直接跑SQLite 单文件备份就是复制文件对个人项目非常友好。4. 验证请求从上报到告警触发的完整动作配置写完必须验证不然你不知道是接口通了还是模型没调上。这一节给你一条完整的验证链路照着敲就能看到告警被触发。第一步确认后端起来了。访问http://127.0.0.1:8000/docs能看到 FastAPI 自动生成的接口文档说明服务正常。第二步上报一条正常数据验证采集路由curl -X POST http://127.0.0.1:8000/report \ -H Content-Type: application/json \ -d {app:order-service,metric:latency_ms,value:120}预期返回{ok:true}没有 alert 字段因为 120 没超过阈值 800。第三步上报一条超阈值数据触发告警curl -X POST http://127.0.0.1:8000/report \ -H Content-Type: application/json \ -d {app:order-service,metric:latency_ms,value:1500}预期返回类似{ok: true, alert: order-service 的 latency_ms 达到 1500超过阈值 800请检查服务响应。}这条 alert 文本就是模型归纳出来的说明 TaoToken 调用链路通了。如果这里返回的 alert 是空的或者报错去第 5 节对照排查。第四步查询告警记录确认落库curl http://127.0.0.1:8000/alerts?limit5能看到刚才那条告警包含 app、metric、value、message、ts。第五步打开前端看板http://127.0.0.1:51735 秒内应该能看到指标表格里出现刚才两条数据告警列表里出现那条归纳后的告警。如果前端没数据先看浏览器控制台有没有跨域报错后端加 CORS 中间件即可from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins[http://127.0.0.1:5173], allow_methods[*], allow_headers[*], )第六步验证冷却去重。连续快速上报三次超阈值数据for i in 1 2 3; do curl -s -X POST http://127.0.0.1:8000/report \ -H Content-Type: application/json \ -d {app:order-service,metric:latency_ms,value:2000} echo done预期只有第一次返回 alert后两次返回{ok:true}因为 60 秒冷却生效。这一步很关键生产环境没有冷却的话一个持续故障会把你通知渠道刷爆。到这一步整条链路验证完毕上报 → 存储 → 阈值判断 → 模型归纳 → 落库 → 看板展示。你可以把上报逻辑接到真实业务里比如在 FastAPI 中间件里统计每个请求的耗时和状态码定时上报或者在定时任务里采集 CPU、内存。采集端就是一个 HTTP 请求任何语言都能发。5. 常见报错排查401、local proxy failed、reading choices、OAuth这一节按真实报错来都是我实际遇到过的。你对照错误信息找对应条目。401 Unauthorized。最常见原因是 Key 没读到或写错。检查三点.env里TAOTOKEN_API_KEY是否以sk-开头且没有多余空格代码里load_dotenv()是否在读取环境变量之前执行部署到服务器时环境变量是否真的注入了systemd要在 service 文件里写Environment或EnvironmentFile。还有一种情况是 Key 被复制时带了换行用echo $TAOTOKEN_API_KEY | wc -c看长度对不对。local proxy failed / connection refused。这个报错通常出现在请求根本没发出去的时候说明本机网络层或客户端配置有问题。检查BASE_URL是不是写成了https://taotoken.net/api有没有多写路径检查本机是否能正常访问外网如果你在容器里跑确认容器网络正常。注意不要配置任何来路不明的网络工具直接用系统默认网络即可。reading choices of undefined。这个报错说明响应体里没有choices字段通常是请求体格式不对或模型 ID 写错。检查payload里model字段是不是你实际可用的模型 IDmessages是不是标准格式。可以先把响应体打印出来看resp await client.post(url, headersheaders, jsonpayload) print(resp.status_code, resp.text)如果返回的是错误 JSON里面一般会写明原因比如模型不存在、参数不合法。把resp.text贴给 AI 让它帮你判断比盲猜快得多。OAuth / authentication 相关报错。如果你用的是某些 CLI 工具比如 Claude Code 这类它可能走的是 OAuth 登录流程而不是 API Key。这时候要确认工具支持用 Base URL API Key 的方式接入把三件套配全Base URL 填https://taotoken.net/apiKey 填你的sk-KeyModel ID 填你要用的模型。三者缺一不可只填两个通常会报认证失败。具体接入方式看文档 https://taotoken.net/doc 。告警不触发。先确认阈值规则RULES里有没有对应应用和指标键名要完全一致再确认上报的 value 确实大于阈值最后看冷却时间可能上一次告警还在 60 秒内。调试时把COOLDOWN临时改成 0。前端跨域。浏览器控制台报 CORS 相关错误按第 4 节的 CORS 中间件配置allow_origins填你前端实际地址。生产环境前端和后端同域的话用 Nginx 反代/api到后端就不需要 CORS 了。数据不落库。检查monitor.db文件是否有写权限容器部署时注意挂载卷否则重启数据就没了。SQLite 并发写有限个人项目够用量大了换 PostgreSQL改db.py里的连接即可。排查的通用心法先看 HTTP 状态码再看响应体原文最后看服务端日志。90% 的问题在响应体里就写清楚了别急着改代码。6. 把监控接进真实业务下一步怎么走平台跑起来只是开始真正有价值的是把它接进你的业务。最省事的做法是在 FastAPI 里加一个中间件统计每个请求的耗时和状态码按分钟聚合后上报import time import httpx from fastapi import Request app.middleware(http) async def collect_metrics(request: Request, call_next): start time.time() response await call_next(request) cost_ms (time.time() - start) * 1000 async with httpx.AsyncClient() as client: await client.post(http://127.0.0.1:8000/report, json{ app: order-service, metric: latency_ms, value: cost_ms, }) return response注意别把上报接口自己也监控了否则会递归。加个路径判断跳过/report。告警通知渠道可以扩展邮件用smtplib企业微信或钉钉用 webhook短信接云厂商 SDK。模型归纳出来的那句话直接作为通知正文比原始数值可读性好很多。你还可以让模型根据告警内容给出初步排查建议比如「latency_ms 飙升通常是数据库慢查询或下游超时建议先看慢日志」。看板可以加趋势图用 ECharts 按时间轴画折线比表格直观。数据量大了给metrics表的ts字段加索引查询会快很多。长期来看如果你想让 AI 持续帮你迭代这个平台——加功能、改 UI、修 bug——用 Coding Plan 会比单次对话更顺上下文能保持住https://taotoken.net/coding-plan 。需要新建更多 Key 或管理配额去控制台https://taotoken.net/console 。接入细节和参数说明都在文档里https://taotoken.net/doc 。想先验证模型输出效果用模型对话页试几句https://taotoken.net/models 。这套东西最大的价值不是技术多复杂而是它真的能跑在生产环境、真的会在出问题时提醒你。我把它跑起来之后半夜被叫醒的次数明显少了因为很多问题在变成事故之前就收到了告警。你先按第 3 节的配置跑通再按第 4 节验证一遍剩下的就是接到你自己的业务里慢慢加规则、加渠道。
