DIY监控仪表盘：使用CubicAPI统计DeepSeek资源利用率

随着AI模型的广泛应用，资源利用率的监控变得愈发重要。无论是本地部署还是云端服务，了解模型调用次数、计算资源消耗以及API响应情况，对于优化成本和提升性能都具有重要意义。

在本篇文章中，我们将探讨如何使用 CubicAPI 构建一个自定义的监控仪表盘，用于统计和展示 DeepSeek 模型的资源使用情况。我们将通过一个完整的代码示例，实现以下功能：

使用 DeepSeek API 发起请求；在每次调用后将关键指标（如 token 数量、调用时间、响应状态等）发送到 CubicAPI；使用 CubicAPI 的聚合能力生成资源使用报告；构建前端仪表盘展示统计数据。

---

前置知识与工具准备

所需技术栈

Python（用于编写 API 调用脚本）Flask（可选，用于构建本地 Web 服务）DeepSeek API（已申请并获取 API Key）CubicAPI（注册账号，创建数据源）

安装依赖

pip install requests flask

---

配置 DeepSeek API 调用

首先，我们需要编写一个简单的函数来调用 DeepSeek 提供的 API 接口，并记录相关参数。

示例：调用 DeepSeek 并返回 token 数量

import requestsimport timeDEEPSEEK_API_KEY = "your_deepseek_api_key"DEEPSEEK_URL = "https://api.deepseek.com/v1/chat/completions"def call_deepseek(prompt):    headers = {        "Authorization": f"Bearer {DEEPSEEK_API_KEY}",        "Content-Type": "application/json"    }    data = {        "model": "deepseek-chat",        "messages": [{"role": "user", "content": prompt}],        "temperature": 0.7    }    start_time = time.time()    response = requests.post(DEEPSEEK_URL, headers=headers, json=data)    end_time = time.time()    if response.status_code == 200:        resp_json = response.json()        input_tokens = resp_json['usage']['prompt_tokens']        output_tokens = resp_json['usage']['completion_tokens']        total_tokens = input_tokens + output_tokens        return {            "status": "success",            "input_tokens": input_tokens,            "output_tokens": output_tokens,            "total_tokens": total_tokens,            "duration": end_time - start_time        }    else:        return {            "status": "error",            "error_code": response.status_code,            "duration": end_time - start_time        }

---

集成 CubicAPI 上报资源使用情况

CubicAPI 是一个轻量级的时序数据收集平台，支持通过 HTTP 接口上传指标数据，并提供可视化面板。我们可以利用它来存储和展示我们的 DeepSeek 使用情况。

创建 CubicAPI 数据源

登录 CubicAPI 控制台，创建一个新的数据源，例如 deepseek_usage，并记下其写入密钥（Write Token）。

编写上报函数

CUBIC_WRITE_TOKEN = "your_cubic_write_token"CUBIC_INGEST_URL = "https://ingest.cubic.to/v1/write"def send_to_cubic(data):    payload = {        "series": [            {                "name": "deepseek_usage",                "columns": ["time", "input_tokens", "output_tokens", "total_tokens", "duration", "status"],                "points": [                    [                        int(time.time() * 1e9),  # 时间戳（纳秒）                        data.get("input_tokens", 0),                        data.get("output_tokens", 0),                        data.get("total_tokens", 0),                        round(data["duration"], 3),                        data["status"]                    ]                ]            }        ]    }    headers = {        "Authorization": f"Bearer {CUBIC_WRITE_TOKEN}",        "Content-Type": "application/json"    }    response = requests.post(CUBIC_INGEST_URL, json=payload, headers=headers)    return response.status_code == 204

整合调用逻辑

def deepseek_query_and_log(prompt):    result = call_deepseek(prompt)    print("Call Result:", result)    success = send_to_cubic(result)    if success:        print("✅ Usage data sent to CubicAPI.")    else:        print("❌ Failed to send data to CubicAPI.")# 示例调用if __name__ == "__main__":    deepseek_query_and_log("请总结一下机器学习的基本概念。")

---

配置 CubicAPI 查询与可视化

在 Cubic 控制台中，你可以创建一个新面板，并使用如下查询语句来分析 DeepSeek 的资源使用情况：

查询示例：每小时总 token 数

SELECT sum(total_tokens) FROM "deepseek_usage" GROUP BY time(1h)

查询示例：平均调用延迟

SELECT mean(duration) FROM "deepseek_usage" GROUP BY time(5m)

查询示例：错误率统计

SELECT count(*) / total() * 100 AS error_rate FROM "deepseek_usage" WHERE status='error' GROUP BY time(1h)

你还可以设置报警规则，比如当错误率超过 5% 或单次调用耗时超过 10 秒时触发通知。

---

构建简易仪表盘（Flask + Bootstrap）

为了更直观地查看统计结果，我们可以通过 Flask 构建一个本地仪表盘页面，从 CubicAPI 获取数据并渲染成图表。

安装额外依赖

pip install jinja2 matplotlib

Flask 应用代码

from flask import Flask, render_templateimport requestsimport matplotlib.pyplot as pltimport base64from io import BytesIOimport timeCUBIC_READ_TOKEN = "your_cubic_read_token"CUBIC_QUERY_URL = "https://query.cubic.to/v1/query"app = Flask(__name__)def query_cubic(query):    headers = {        "Authorization": f"Bearer {CUBIC_READ_TOKEN}"    }    data = {"query": query}    response = requests.post(CUBIC_QUERY_URL, json=data, headers=headers)    return response.json()@app.route('/')def dashboard():    # 查询 token 使用趋势    token_data = query_cubic('SELECT sum(total_tokens) FROM "deepseek_usage" GROUP BY time(1h)')    times = []    tokens = []    for row in token_data[0]['series'][0]['values']:        timestamp = row[0] // 1_000_000_000  # 转换为秒        times.append(time.strftime('%H:%M', time.localtime(timestamp)))        tokens.append(row[1])    # 绘图    plt.figure(figsize=(10, 4))    plt.plot(times, tokens, marker='o')    plt.title("Total Tokens per Hour")    plt.xlabel("Time")    plt.ylabel("Tokens")    plt.grid(True)    img = BytesIO()    plt.savefig(img, format='png')    img.seek(0)    plot_url = base64.b64encode(img.getvalue()).decode()    plt.close()    return render_template('dashboard.html', plot_url=plot_url)if __name__ == '__main__':    app.run(debug=True)

HTML 模板（templates/dashboard.html）

<!DOCTYPE html><html lang="en"><head>    <meta charset="UTF-8">    <title>DeepSeek Usage Dashboard</title></head><body>    <h1>DeepSeek 资源使用监控</h1>    <img src="data:image/png;base64,{{ plot_url }}" alt="Token Usage Chart"></body></html>

运行 Flask 后访问 http://localhost:5000 即可看到实时绘制的图表。

---

扩展建议

多模型支持：可在调用时加入 model 字段，便于区分不同模型的使用情况。用户身份识别：若系统有多个用户，可以添加 user_id 标签进行分组统计。自动定时任务：使用 cron job 定期执行测试调用，持续产生样本数据。接入 Grafana：Cubic 支持 InfluxDB 兼容协议，可直接接入 Grafana 做高级可视化。

---

通过本文，我们成功实现了使用 CubicAPI 构建一个定制化的 DeepSeek 资源监控仪表盘。整个过程涵盖了 API 调用、数据采集、时序数据库写入、查询分析及前端展示等多个环节，具备良好的扩展性和实用性。

如果你正在管理一个 AI 服务或希望更好地控制模型调用成本，这种 DIY 监控方案无疑是一个低成本、高灵活性的选择。

---

📌 GitHub 示例项目地址（假设）
你可以将上述代码整理为一个开源项目，例如：
https://github.com/yourname/deepseek-monitor-dashboard

欢迎 Fork & Star！如有问题也欢迎留言交流。

---

作者：AI运维工程师 | 日期：2025年4月5日