taijibaga/taiji-AI-PAD
9
0
Fork 0
forked from xiaohei/taiji-AI-PAD
Code Pull Requests Activity

更新litellm

Browse Source
This commit is contained in:
zhanggangyong
2026-01-07 10:46:55 +00:00
parent 8bbc97ae36
commit bb719db8ed
5 changed files with 1075 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Docs/租户用户端-接口对接文档.md
+95 -15
View File
@@ -1,6 +1,6 @@
# 租户用户端 - 接口对接文档
> **版本**: v1.0.0
> **版本**: v1.1.0
> **更新时间**: 2026-01-07
> **说明**: 本文档核实了前端租户用户端与后端API的对接情况，包括接口路径、请求参数、响应格式等详细信息
@@ -218,21 +218,63 @@
|------|----------|----------|
| **路径** | `GET /api/v1/monitoring/dashboard` | `GET /api/v1/monitoring/dashboard` |
| **实现文件** | - | [`monitoring.py:63`](../services/mcp-server/app/routes/monitoring.py:63) |
| **核心逻辑** | - | [`monitoring.py:476`](../services/mcp-server/monitoring.py:476) |
**请求头**: `Authorization: Bearer <token>` (需要登录获取租户级别数据)
**功能说明**:
该接口提供租户级别的监控仪表盘数据，包括：
1. **当前租户EU消耗过去24小时的消耗情况**
2. **使用的模型和模型的使用情况、当前调用次数**
3. **每周API调用的次数，7天的每一天调用多少次，包括成功和失败**
4. **三个微服务的状态（mcp_server, data_ingestion, api_gateway）**
**响应格式**:
```json
{
"timestamp": "2026-01-07T00:00:00",
"timestamp": "2026-01-07T07:26:13.401465",
"health": {
"score": 98,
"status": "healthy",
"score": 100,
"timestamp": "2026-01-07T07:26:12.271865",
"uptime_seconds": 128.265972,
"services": {
"mcp_server": { "status": "healthy", "latency": 45 },
"data_ingestion": { "status": "healthy", "latency": 32 },
"api_gateway": { "status": "healthy", "latency": 28 }
"database": { "status": "healthy", "latency": 0 },
"mcp_server": { "status": "healthy", "latency": 1 },
"data_ingestion": { "status": "healthy", "latency": 65 },
"api_gateway": { "status": "healthy", "latency": 8 }
}
},
"metrics": { ... },
"stats": { ... },
"euConsumption24h": {
"total": 150.5,
"totalCost": 15.05,
"totalCalls": 1250,
"hourlyData": [
{ "timestamp": "2026-01-06T00:00:00", "value": 10.5, "cost": 1.05, "calls": 50 },
{ "timestamp": "2026-01-06T01:00:00", "value": 12.3, "cost": 1.23, "calls": 60 }
]
},
"modelUsage": {
"models": [
{ "name": "gpt-4", "calls": 500, "euConsumed": 100.0, "cost": 10.0 },
{ "name": "gpt-3.5-turbo", "calls": 750, "euConsumed": 50.5, "cost": 5.05 }
],
"totalCalls": 1250
},
"weeklyApiCalls": {
"dailyData": [
{ "date": "2026-01-01", "total": 100, "success": 95, "failed": 5 },
{ "date": "2026-01-02", "total": 120, "success": 115, "failed": 5 },
{ "date": "2026-01-03", "total": 90, "success": 88, "failed": 2 },
{ "date": "2026-01-04", "total": 150, "success": 145, "failed": 5 },
{ "date": "2026-01-05", "total": 200, "success": 190, "failed": 10 },
{ "date": "2026-01-06", "total": 180, "success": 175, "failed": 5 },
{ "date": "2026-01-07", "total": 160, "success": 157, "failed": 3 }
],
"totalSuccess": 965,
"totalFailed": 35,
"total": 1000
},
"alerts": {
"items": [],
"count": 0,
@@ -242,6 +284,13 @@
}
```
**微服务状态说明**:
| 状态值 | 说明 |
|--------|------|
| `healthy` | 运行中，服务正常 |
| `maintenance` | 维护中，服务暂时不可用 |
| `error` | 错误，服务异常 |
---
### D4. 计费余额接口 ✅ 完全对接
@@ -298,6 +347,12 @@
## 服务网关模块
缺少服务网关监控接口
可用网关数量这里可以是前端假数据
服务端点是用户自建API
平均延迟
今日调用必须含自己api网关和平台提供的三个网关接口调用
### B6. 创建API接口 ✅ 完全对接
| 项目 | 前端需求 | 后端实现 |
@@ -328,7 +383,8 @@
---
### B7. 选择网关类型接口 ✅ 完全对接
### B7. 选择网关类型接口 ✅ 完全对接 网关类型没必要做接口选择，用户不是在这里选择网关
| 项目 | 前端需求 | 后端实现 |
|------|----------|----------|
@@ -355,7 +411,7 @@
---
### D6. 网关API列表接口 ✅ 完全对接
### D6. 网关API列表接口 ✅ 完全对接 这也不需要
| 项目 | 前端需求 | 后端实现 |
|------|----------|----------|
@@ -381,7 +437,7 @@
---
### D7. 网关监控数据接口 ✅ 完全对接
### D7. 网关监控数据接口 ✅ 完全对接 监控是模型监控具体到供应商延迟，第二个是模型请求负载情况，下面参数不对。
| 项目 | 前端需求 | 后端实现 |
|------|----------|----------|
@@ -403,7 +459,7 @@
---
### D8. 模型提供商列表接口 ✅ 完全对接
### D8. 模型提供商列表接口 ✅ 完全对接 这里也不需要
| 项目 | 前端需求 | 后端实现 |
|------|----------|----------|
@@ -432,6 +488,13 @@
---
## 数据与工具模块
缺少看板接口
1、总工具数 全部工具包括平台的
2、生成的工具 自定义的工具
3、现在活跃的工具，指的是已经部署到agnet的工具
###
### B8. 生成工具接口 ✅ 完全对接
@@ -471,7 +534,9 @@
---
### B9. 创建数据模板接口 ✅ 完全对接
### B9. 创建数据模板接口 ✅ 完全对接
缺少模板类型上传接口 云存储和数据库模板
| 项目 | 前端需求 | 后端实现 |
|------|----------|----------|
@@ -506,6 +571,8 @@
### D9. 工具列表接口 ✅ 完全对接
缺少删除、修改
| 项目 | 前端需求 | 后端实现 |
|------|----------|----------|
| **路径** | `GET /tools` | `GET /tools` |
@@ -541,7 +608,7 @@
---
### D10. 统计信息接口 ✅ 完全对接
### D10. 统计信息接口 ✅ 完全对接 不要
| 项目 | 前端需求 | 后端实现 |
|------|----------|----------|
@@ -563,6 +630,13 @@
### B10. 部署Agent接口 ✅ 完全对接
缺少看板
可用的Agnet主要是渠道分配多少个显示多少个
已经部署的agnet是自定义和平台的已经部署的agnet
所有平台和自定义agnet使用的CPU和内存
| 项目 | 前端需求 | 后端实现 |
|------|----------|----------|
| **路径** | `POST /api/user/agents/deploy` | `POST /api/user/agents/deploy` |
@@ -596,6 +670,11 @@
}
```
B11以下全部没有。
---
### B11. 创建自定义Agent接口 ⚠️ 路径差异
@@ -1037,4 +1116,5 @@
| 版本 | 日期 | 更新内容 |
|------|------|----------|
| v1.0.0 | 2026-01-07 | 初始版本，完成前后端接口对接核实 |
| v1.0.0 | 2026-01-07 | 初始版本，完成前后端接口对接核实 |
| v1.1.0 | 2026-01-07 | 完善D3监控仪表盘接口，实现租户级别数据过滤、模型使用统计、每周API调用统计、三个微服务状态检查 |
plans/模型供应商与租户模型使用设计方案.md
+527
View File
@@ -0,0 +1,527 @@
# 模型供应商与租户模型使用设计方案（最终版 v5.0）
> **版本**: v5.0.0
> **创建时间**: 2026-01-07
> **状态**: 待评审
---
## 1. 核心职责分工
### 1.1 职责划分表
| 职责 | 负责方 | 说明 |
|------|--------|------|
| 决定租户能用哪些模型 | **mcp-server** | 业务规则制定者 |
| 决定 RPM/TPM/Budget 数值 | **mcp-server** | 配额分配 |
| 真正拦截超额请求 | **litellm-gateway** | 规则执行者 |
| 管理 Agent 生命周期 | **mcp-server + Agent Manager** | 与 litellm 解耦 |
| 用量统计和计费 | **litellm-gateway** | 原生能力 |
### 1.2 概念映射
| 平台概念 | LiteLLM 对应 | 说明 |
|----------|-------------|------|
| 渠道 | team | 一个渠道 = 一个 team |
| 租户 | team 下的 key 或子 team | 租户归属于渠道 |
| 模型供应商 | model/provider | litellm 配置 |
| 平台 Agent | 不进 LiteLLM | mcp-server 自己管理 |
### 1.3 LiteLLM 不关心的事情
- ❌ Agent 是什么
- ❌ Agent 从哪来
- ❌ Agent 跑在 AKS 还是别的地方
- ❌ Agent 生命周期
**LiteLLM 只看请求里的**：
- ✅ api_key
- ✅ team
- ✅ model
---
## 2. 整体架构
### 2.1 架构图
```mermaid
graph TB
subgraph mcp-server[mcp-server 业务规则制定者]
A1[管理员创建渠道]
A2[分配模型供应商给渠道]
A3[渠道分配模型给租户]
A4[设置 RPM/TPM/Budget]
A5[用户充值更新额度]
end
subgraph litellm[litellm-gateway 规则执行者]
B1[team 管理]
B2[key 管理]
B3[rate_limit 检查]
B4[budget 检查]
B5[用量统计]
end
subgraph agent[Agent 运行层 - 与 litellm 解耦]
C1[Agent Manager]
C2[AKS Pod]
C3[调用模型]
end
A1 --> |创建 team| B1
A3 --> |创建 key + 配额| B2
A4 --> |设置 rate_limit| B3
A5 --> |更新 budget| B4
C2 --> |带 api_key 请求| litellm
litellm --> |检查配额| B3
litellm --> |检查余额| B4
litellm --> |转发| Provider[Azure/Gemini]
```
### 2.2 核心流程
```mermaid
sequenceDiagram
participant Admin as 管理员
participant MCP as mcp-server
participant LiteLLM as litellm-gateway
participant Agent as Agent Pod
participant Azure as Azure/Gemini
Note over Admin,Azure: 1️⃣ 创建渠道
Admin->>MCP: 创建渠道
MCP->>LiteLLM: POST /team/new
Note over MCP,LiteLLM: team_alias: channel-xxx
Note over Admin,Azure: 2️⃣ 分配模型给渠道
Admin->>MCP: 分配模型 azure/gpt-4
MCP->>MCP: 记录到 ResourceAllocation
Note over Admin,Azure: 3️⃣ 渠道分配模型给租户
MCP->>LiteLLM: POST /key/generate
Note over MCP,LiteLLM: team_id, models, rpm_limit, tpm_limit, max_budget
LiteLLM-->>MCP: 返回 api_key
MCP->>MCP: 保存 key 到数据库
Note over Admin,Azure: 4️⃣ 租户创建 Agent
MCP->>MCP: 查询租户的 api_key
MCP->>Agent: 启动 Pod，注入 api_key
Note over Admin,Azure: 5️⃣ Agent 调用模型
Agent->>LiteLLM: POST /chat/completions
Note over Agent,LiteLLM: Authorization: Bearer api_key
LiteLLM->>LiteLLM: 检查 RPM/TPM
LiteLLM->>LiteLLM: 检查 Budget
alt 配额/余额不足
LiteLLM-->>Agent: 429/403 拒绝
else 配额充足
LiteLLM->>Azure: 转发请求
Azure-->>LiteLLM: 返回结果
LiteLLM->>LiteLLM: 记录用量
LiteLLM-->>Agent: 返回结果
end
Note over Admin,Azure: 6️⃣ 用户充值
MCP->>LiteLLM: POST /key/update
Note over MCP,LiteLLM: 更新 max_budget
Note over MCP,LiteLLM: 下一次请求立刻放行，无需重启
```
---
## 3. 详细设计
### 3.1 创建渠道（对应 LiteLLM team）
**mcp-server 操作**：
```python
async def create_channel(channel_name: str, db: AsyncSession):
# 1. 在 mcp-server 创建渠道记录
channel = Channel(name=channel_name, ...)
db.add(channel)
# 2. 在 litellm 创建对应的 team
async with httpx.AsyncClient() as client:
response = await client.post(
f"{settings.litellm_url}/team/new",
json={
"team_alias": f"channel-{channel.id}",
"metadata": {
"channel_id": str(channel.id),
"channel_name": channel_name
}
},
headers={"Authorization": f"Bearer {settings.litellm_master_key}"}
)
data = response.json()
# 保存 litellm team_id
channel.litellm_team_id = data["team_id"]
await db.commit()
return channel
```
### 3.2 分配模型给渠道
**mcp-server 操作**：
```python
async def allocate_models_to_channel(
channel_id: str,
models: list[str], # ["azure/gpt-4", "gemini/gemini-pro"]
db: AsyncSession
):
# 只在 mcp-server 记录，不需要同步到 litellm
# litellm 的 team 不限制模型，模型限制在 key 级别
for model in models:
allocation = ResourceAllocation(
target_id=channel_id,
target_type="channel",
resource_type="model",
resource_id=model
)
db.add(allocation)
await db.commit()
```
### 3.3 渠道分配模型给租户（核心）
**mcp-server 操作**：
```python
async def allocate_model_to_tenant(
tenant_id: str,
channel_id: str,
model_name: str,
rpm_limit: int,
tpm_limit: int,
max_budget: float,
db: AsyncSession
):
# 1. 验证渠道是否有该模型
channel_has_model = await check_channel_has_model(channel_id, model_name, db)
if not channel_has_model:
raise HTTPException(status_code=403, detail="渠道没有该模型的权限")
# 2. 获取渠道的 litellm team_id
channel = await get_channel(channel_id, db)
# 3. 在 litellm 创建 key（绑定 team + model + 配额）
async with httpx.AsyncClient() as client:
response = await client.post(
f"{settings.litellm_url}/key/generate",
json={
"team_id": channel.litellm_team_id,
"models": [model_name],
"rpm_limit": rpm_limit,
"tpm_limit": tpm_limit,
"max_budget": max_budget,
"budget_duration": "monthly", # 或 "total"
"metadata": {
"tenant_id": str(tenant_id),
"channel_id": str(channel_id),
"model": model_name
}
},
headers={"Authorization": f"Bearer {settings.litellm_master_key}"}
)
data = response.json()
# 4. 保存到数据库
tenant_key = TenantModelKey(
tenant_id=tenant_id,
channel_id=channel_id,
model_name=model_name,
litellm_key_id=data["key"],
litellm_key_hash=encrypt(data["key"]), # 加密存储
rpm_limit=rpm_limit,
tpm_limit=tpm_limit,
max_budget=max_budget
)
db.add(tenant_key)
# 5. 同时记录到 ResourceAllocation
allocation = ResourceAllocation(
target_id=tenant_id,
target_type="tenant",
resource_type="model",
resource_id=model_name,
rpm=rpm_limit,
tpm=tpm_limit
)
db.add(allocation)
await db.commit()
return tenant_key
```
### 3.4 租户创建 Agent
**mcp-server 操作**：
```python
async def create_custom_agent(
tenant_id: str,
agent_name: str,
model_name: str,
cpu_request: str,
memory_request: str,
db: AsyncSession
):
# 1. 查询租户的模型 key
tenant_key = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == tenant_id,
TenantModelKey.model_name == model_name,
TenantModelKey.status == "active"
)
)
)
tenant_key = tenant_key.scalar_one_or_none()
if not tenant_key:
raise HTTPException(
status_code=403,
detail=f"您没有使用模型 {model_name} 的权限"
)
# 2. 调用 Agent Manager 创建 Pod
# Agent Manager 不需要知道 litellm 的任何细节
# 只需要注入环境变量
result = await agent_manager.create_agent(
name=agent_name,
cpu_request=cpu_request,
memory_request=memory_request,
env_vars={
"OPENAI_API_BASE": settings.litellm_url,
"OPENAI_API_KEY": decrypt(tenant_key.litellm_key_hash),
"MODEL_NAME": model_name
}
)
return result
```
### 3.5 用户充值更新额度
**mcp-server 操作**：
```python
async def recharge_tenant_budget(
tenant_id: str,
model_name: str,
additional_budget: float,
db: AsyncSession
):
# 1. 获取租户的 key
tenant_key = await get_tenant_model_key(tenant_id, model_name, db)
# 2. 计算新的 budget
new_budget = float(tenant_key.max_budget or 0) + additional_budget
# 3. 更新 litellm key 的 budget
async with httpx.AsyncClient() as client:
await client.post(
f"{settings.litellm_url}/key/update",
json={
"key": tenant_key.litellm_key_id,
"max_budget": new_budget
},
headers={"Authorization": f"Bearer {settings.litellm_master_key}"}
)
# 4. 更新数据库
tenant_key.max_budget = new_budget
await db.commit()
# ✅ 下一次请求立刻放行，无需重启任何服务
```
---
## 4. 数据模型
### 4.1 新增表：TenantModelKey
```sql
CREATE TABLE tenant_model_keys (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id UUID NOT NULL REFERENCES users(id),
channel_id UUID REFERENCES channels(id),
-- 模型信息
model_name VARCHAR(100) NOT NULL,
-- litellm Key 信息
litellm_key_id VARCHAR(255) NOT NULL, -- litellm 返回的完整 key
litellm_key_hash TEXT NOT NULL, -- 加密存储
-- 配额配置（与 litellm 同步）
rpm_limit INTEGER DEFAULT 0,
tpm_limit INTEGER DEFAULT 0,
max_budget NUMERIC(12, 2),
budget_duration VARCHAR(20) DEFAULT 'monthly',
-- 状态
status VARCHAR(20) DEFAULT 'active',
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW(),
UNIQUE(tenant_id, model_name)
);
```
### 4.2 Channel 表增加字段
```sql
ALTER TABLE channels ADD COLUMN litellm_team_id VARCHAR(100);
```
---
## 5. LiteLLM 配置
### 5.1 litellm_config.yaml
```yaml
general_settings:
master_key: "sk-taiji-master-key"
database_url: "postgresql://..." # 可选，用于持久化
model_list:
# Azure GPT-4
- model_name: "azure/gpt-4"
litellm_params:
model: "azure/gpt-4"
api_base: "https://taiji-azure.openai.azure.com"
api_key: "os.environ/AZURE_API_KEY"
api_version: "2024-02-15-preview"
# Azure GPT-3.5
- model_name: "azure/gpt-35-turbo"
litellm_params:
model: "azure/gpt-35-turbo"
api_base: "https://taiji-azure.openai.azure.com"
api_key: "os.environ/AZURE_API_KEY"
api_version: "2024-02-15-preview"
# Google Gemini
- model_name: "gemini/gemini-pro"
litellm_params:
model: "gemini/gemini-pro"
api_key: "os.environ/GOOGLE_API_KEY"
# 启用用量追踪
litellm_settings:
success_callback: ["prometheus"]
track_cost_callback: true
```
### 5.2 LiteLLM Admin API 使用
| API | 用途 | 调用时机 |
|-----|------|----------|
| `POST /team/new` | 创建 team | 创建渠道时 |
| `POST /key/generate` | 创建 key | 分配模型给租户时 |
| `POST /key/update` | 更新 key 配额 | 修改配额/充值时 |
| `POST /key/delete` | 删除 key | 取消分配时 |
| `GET /spend/logs` | 查询用量 | 统计报表时 |
---
## 6. 接口设计
### 6.1 渠道管理接口
| 接口 | 方法 | 功能 | 变更 |
|------|------|------|------|
| `POST /api/admin/channels/create` | POST | 创建渠道 | 同步创建 litellm team |
| `DELETE /api/admin/channels/{id}` | DELETE | 删除渠道 | 同步删除 litellm team |
### 6.2 模型分配接口
| 接口 | 方法 | 功能 | 变更 |
|------|------|------|------|
| `PUT /api/channel/tenants/{id}/models` | PUT | 分配模型给租户 | 创建 litellm key |
| `DELETE /api/channel/tenants/{id}/models/{model}` | DELETE | 取消分配 | 删除 litellm key |
| `PUT /api/channel/tenants/{id}/models/{model}/quota` | PUT | 更新配额 | 更新 litellm key |
### 6.3 租户接口
| 接口 | 方法 | 功能 |
|------|------|------|
| `GET /api/user/models/available` | GET | 获取可用模型列表 |
| `POST /api/user/custom-agents` | POST | 创建 Agent（注入 key） |
| `GET /api/user/models/usage/stats` | GET | 查询用量统计 |
| `POST /api/user/billing/recharge` | POST | 充值（更新 litellm budget） |
---
## 7. 实现计划
### 7.1 任务清单
- [ ] **数据库迁移**
- [ ] 创建 `tenant_model_keys` 表
- [ ] Channel 表增加 `litellm_team_id` 字段
- [ ] **litellm 集成**
- [ ] 实现 litellm Admin API 客户端
- [ ] 创建渠道时同步创建 team
- [ ] 分配模型时创建 key
- [ ] 充值时更新 budget
- [ ] **接口调整**
- [ ] 更新渠道创建接口
- [ ] 实现模型分配接口
- [ ] 更新 Agent 创建接口
- [ ] 实现充值接口
- [ ] **Agent Manager**
- [ ] 支持注入 litellm key 环境变量
---
## 8. 关键行为说明
### 8.1 配额超限行为
| 状态 | LiteLLM 行为 | 说明 |
|------|-------------|------|
| RPM 超限 | 返回 429 | 等待下一分钟自动恢复 |
| TPM 超限 | 返回 429 | 等待下一分钟自动恢复 |
| Budget 用完 | 返回 403 | 需要充值才能恢复 |
### 8.2 充值后恢复
```
用户充值 → mcp-server 调用 litellm API 更新 budget → 下一次请求立刻放行
```
- ✅ 不需要重启任何服务
- ✅ 不需要重新创建 key
- ✅ 实时生效
---
## 9. 总结
### 9.1 核心设计原则
1. **mcp-server 是业务规则制定者**：决定谁能用什么、用多少
2. **litellm-gateway 是规则执行者**：真正拦截超额请求
3. **Agent 与 litellm 解耦**：Agent 只需要带正确的 api_key 调用
4. **实时生效**：配额更新、充值后立刻生效，无需重启
### 9.2 架构优势
| 优势 | 说明 |
|------|------|
| 职责清晰 | mcp-server 管业务，litellm 管执行 |
| 延迟低 | Agent 直接调用 litellm，无需代理 |
| 原生能力 | 充分利用 litellm 的配额和计费能力 |
| 易扩展 | 新增模型只需配置 litellm |
| 实时生效 | 配额更新无需重启 |
services/mcp-server/app/routes/monitoring.py
+64 -22
View File
@@ -4,9 +4,11 @@ from __future__ import annotations
import asyncio
import structlog
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, HTTPException, Depends
from typing import Optional
from monitoring import system_monitor
from app.auth import require_auth
logger = structlog.get_logger(__name__)
@@ -61,30 +63,70 @@ async def get_system_alerts(severity: str | None = None) -> dict:
@router.get("/dashboard")
async def get_monitoring_dashboard() -> dict:
"""Aggregate health, metrics, stats, and alerts for dashboards."""
async def get_monitoring_dashboard(
principal: Optional[dict] = Depends(require_auth)
) -> dict:
"""
Aggregate health, metrics, stats, and alerts for dashboards.
如果用户已登录，返回租户级别的数据；否则返回全局数据。
返回数据包括：
1. 当前租户EU消耗过去24小时的消耗情况
2. 使用的模型和模型的使用情况、当前调用次数
3. 每周API调用的次数，7天的每一天调用多少次，包括成功和失败
4. 三个微服务的状态（mcp_server, data_ingestion, api_gateway）
"""
try:
# 获取租户ID（如果已登录）
tenant_id = principal.get("user_id") if principal else None
# 获取基础健康状态和告警
health_task = system_monitor.get_system_health()
metrics_task = system_monitor.get_system_metrics()
stats_task = system_monitor.get_service_stats("all")
alerts_task = system_monitor.get_alerts()
health, metrics, stats, alerts = await asyncio.gather(
health_task, metrics_task, stats_task, alerts_task
)
return {
"timestamp": await _current_timestamp(),
"health": health,
"metrics": metrics,
"stats": stats.get("stats", {}),
"alerts": {
"items": alerts,
"count": len(alerts),
"critical_count": len([a for a in alerts if a.get("severity") == "critical"]),
"warning_count": len([a for a in alerts if a.get("severity") == "warning"]),
},
}
health, alerts = await asyncio.gather(health_task, alerts_task)
# 如果有租户ID，获取租户级别的数据
if tenant_id:
tenant_data = await system_monitor.get_tenant_dashboard(tenant_id)
return {
"timestamp": await _current_timestamp(),
"health": health,
# 租户级别的EU消耗（24小时）
"euConsumption24h": tenant_data.get("euConsumption24h", {}),
# 模型使用情况
"modelUsage": tenant_data.get("modelUsage", {}),
# 每周API调用统计（7天每天成功/失败）
"weeklyApiCalls": tenant_data.get("weeklyApiCalls", {}),
# 告警信息
"alerts": {
"items": alerts,
"count": len(alerts),
"critical_count": len([a for a in alerts if a.get("severity") == "critical"]),
"warning_count": len([a for a in alerts if a.get("severity") == "warning"]),
},
}
else:
# 未登录用户返回全局数据
metrics_task = system_monitor.get_system_metrics()
stats_task = system_monitor.get_service_stats("all")
metrics, stats = await asyncio.gather(metrics_task, stats_task)
return {
"timestamp": await _current_timestamp(),
"health": health,
"metrics": metrics,
"stats": stats.get("stats", {}),
"alerts": {
"items": alerts,
"count": len(alerts),
"critical_count": len([a for a in alerts if a.get("severity") == "critical"]),
"warning_count": len([a for a in alerts if a.get("severity") == "warning"]),
},
}
except Exception as exc:
logger.error("获取监控仪表板失败", error=str(exc))
raise HTTPException(status_code=500, detail=str(exc)) from exc
services/mcp-server/app/routes/user.py
+68 -11
View File
@@ -110,6 +110,8 @@ async def get_dashboard_stats(
"""
获取用户仪表板统计数据
"""
from monitoring import system_monitor
user_id = principal.get("user_id")
# 活跃Agent数量
@@ -134,8 +136,12 @@ async def get_dashboard_stats(
user = result.scalar_one_or_none()
eu_balance = float(user.balance) if user else 0
# 系统健康度（简化计算）
system_health = 98.5
# 系统健康度（从监控模块获取真实数据）
try:
health_data = await system_monitor.get_system_health()
system_health = health_data.get("score", 100)
except Exception:
system_health = 100 # 默认健康
return SuccessResponse(
data={
@@ -268,16 +274,67 @@ async def get_gateway_monitoring(
):
"""
获取网关监控数据
返回当前租户的网关使用统计：
- uptime: 服务可用率（基于过去24小时的成功率）
- requestsPerMinute: 每分钟请求数（基于过去1小时的平均值）
- averageLatency: 平均延迟（毫秒）
- errorRate: 错误率（百分比）
"""
# 简化实现，返回模拟数据
return SuccessResponse(
data={
"uptime": 99.9,
"requestsPerMinute": 1250,
"averageLatency": 45,
"errorRate": 0.1,
}
)
user_id = principal.get("user_id")
try:
# 计算过去1小时的请求统计
one_hour_ago = datetime.utcnow() - timedelta(hours=1)
# 查询过去1小时的请求数和错误数
stats_result = await db.execute(
select(
func.count(BillingRecord.id).label("total"),
func.count(
func.nullif(BillingRecord.status == 'failed', False)
).label("failed"),
func.avg(BillingRecord.duration).label("avg_duration")
)
.where(BillingRecord.tenant_id == user_id)
.where(BillingRecord.timestamp >= one_hour_ago)
)
stats = stats_result.fetchone()
total_requests = stats[0] or 0 if stats else 0
failed_requests = stats[1] or 0 if stats else 0
avg_duration = float(stats[2] or 0) if stats and stats[2] else 0
# 计算每分钟请求数
requests_per_minute = round(total_requests / 60, 2) if total_requests > 0 else 0
# 计算错误率
error_rate = round((failed_requests / total_requests * 100), 2) if total_requests > 0 else 0
# 计算可用率（100% - 错误率）
uptime = round(100 - error_rate, 2)
# 平均延迟（转换为毫秒）
average_latency = round(avg_duration * 1000, 2) if avg_duration else 0
return SuccessResponse(
data={
"uptime": uptime,
"requestsPerMinute": requests_per_minute,
"averageLatency": average_latency,
"errorRate": error_rate,
}
)
except Exception as e:
# 如果查询失败，返回默认值
return SuccessResponse(
data={
"uptime": 100.0,
"requestsPerMinute": 0,
"averageLatency": 0,
"errorRate": 0,
}
)
# ============= 自定义 Agent 配额 =============
services/mcp-server/monitoring.py
+321 -3
View File
@@ -8,6 +8,7 @@ import json
import logging
import psutil
import time
import httpx
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from sqlalchemy.ext.asyncio import AsyncSession
@@ -16,6 +17,7 @@ from sqlalchemy.orm import selectinload
from models import Agent, Execution, User, Tool, Session
from database import AsyncSessionLocal
from config import settings
logger = logging.getLogger(__name__)
@@ -25,11 +27,18 @@ class SystemMonitor:
def __init__(self):
self.start_time = datetime.utcnow()
# 服务健康检查端点配置
self._service_endpoints = {
"mcp_server": "http://localhost:8000/health", # 本服务
"data_ingestion": "http://data-ingestion:8000/health", # Docker 内部网络
"api_gateway": "http://api-gateway:80/health", # Nginx
}
async def get_system_health(self) -> Dict[str, Any]:
"""获取系统健康状态"""
"""获取系统健康状态，包括三个微服务的状态"""
health = {
"status": "healthy",
"score": 100,
"timestamp": datetime.utcnow().isoformat(),
"uptime_seconds": (datetime.utcnow() - self.start_time).total_seconds(),
"services": {}
@@ -39,13 +48,85 @@ class SystemMonitor:
try:
async with AsyncSessionLocal() as session:
await session.execute(text("SELECT 1"))
health["services"]["database"] = "healthy"
health["services"]["database"] = {"status": "healthy", "latency": 0}
except Exception as e:
health["services"]["database"] = f"unhealthy: {str(e)}"
health["services"]["database"] = {"status": "error", "latency": 0, "error": str(e)}
health["status"] = "degraded"
health["score"] -= 25
# 检查三个微服务状态
service_health = await self._check_microservices_health()
health["services"].update(service_health)
# 计算整体健康分数
unhealthy_count = sum(1 for s in health["services"].values()
if s.get("status") not in ["healthy", "running"])
if unhealthy_count > 0:
health["score"] = max(0, 100 - (unhealthy_count * 20))
if unhealthy_count >= 2:
health["status"] = "degraded"
if unhealthy_count >= 3:
health["status"] = "unhealthy"
return health
async def _check_microservices_health(self) -> Dict[str, Dict[str, Any]]:
"""
检查三个微服务的健康状态
返回格式：
{
"mcp_server": { "status": "healthy", "latency": 45 },
"data_ingestion": { "status": "healthy", "latency": 32 },
"api_gateway": { "status": "healthy", "latency": 28 }
}
状态值：
- healthy/running: 运行中
- maintenance: 维护中
- error: 错误
"""
results = {}
async def check_service(name: str, url: str) -> Dict[str, Any]:
"""检查单个服务的健康状态"""
start_time = time.time()
try:
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.get(url)
latency = int((time.time() - start_time) * 1000)
if response.status_code == 200:
return {"status": "healthy", "latency": latency}
elif response.status_code == 503:
return {"status": "maintenance", "latency": latency}
else:
return {"status": "error", "latency": latency, "code": response.status_code}
except httpx.TimeoutException:
return {"status": "error", "latency": 5000, "error": "timeout"}
except httpx.ConnectError:
# 服务不可达，可能是维护中或未启动
return {"status": "maintenance", "latency": 0, "error": "connection_refused"}
except Exception as e:
return {"status": "error", "latency": 0, "error": str(e)}
# mcp_server 本服务始终是健康的（因为能响应请求）
results["mcp_server"] = {"status": "healthy", "latency": 1}
# 检查 data_ingestion 服务
results["data_ingestion"] = await check_service(
"data_ingestion",
self._service_endpoints["data_ingestion"]
)
# 检查 api_gateway (Nginx)
results["api_gateway"] = await check_service(
"api_gateway",
self._service_endpoints["api_gateway"]
)
return results
async def get_system_metrics(self) -> Dict[str, Any]:
"""获取系统性能指标"""
try:
@@ -392,6 +473,243 @@ class SystemMonitor:
except Exception as e:
logger.error(f"获取告警失败: {e}")
return []
async def get_tenant_dashboard(self, tenant_id: str) -> Dict[str, Any]:
"""
获取租户级别的监控仪表盘数据
包含：
1. 当前租户EU消耗过去24小时的消耗情况
2. 使用的模型和模型的使用情况、当前调用次数
3. 每周API调用的次数，7天的每一天调用多少次，包括成功和失败
4. 三个微服务的状态
"""
try:
async with AsyncSessionLocal() as session:
# 1. 获取租户过去24小时的EU消耗
eu_consumption = await self._get_tenant_eu_consumption_24h(session, tenant_id)
# 2. 获取模型使用情况
model_usage = await self._get_tenant_model_usage(session, tenant_id)
# 3. 获取每周API调用统计（7天每天成功/失败）
weekly_api_calls = await self._get_tenant_weekly_api_calls(session, tenant_id)
# 4. 获取微服务状态
services_health = await self._check_microservices_health()
return {
"timestamp": datetime.utcnow().isoformat(),
"tenantId": tenant_id,
"euConsumption24h": eu_consumption,
"modelUsage": model_usage,
"weeklyApiCalls": weekly_api_calls,
"health": {
"score": 100, # 将在下面计算
"services": services_health
}
}
except Exception as e:
logger.error(f"获取租户仪表盘数据失败: {e}")
raise
async def _get_tenant_eu_consumption_24h(
self,
session: AsyncSession,
tenant_id: str
) -> Dict[str, Any]:
"""
获取租户过去24小时的EU消耗情况
返回：
- total: 总EU消耗
- hourlyData: 每小时的消耗数据
"""
try:
# 总消耗
total_result = await session.execute(
text("""
SELECT
COALESCE(SUM(eu), 0) as total_eu,
COALESCE(SUM(cost), 0) as total_cost,
COUNT(*) as total_calls
FROM billing_records
WHERE tenant_id = :tenant_id
AND timestamp > NOW() - INTERVAL '24 hours'
"""),
{"tenant_id": tenant_id}
)
total_row = total_result.fetchone()
# 每小时消耗
hourly_result = await session.execute(
text("""
SELECT
DATE_TRUNC('hour', timestamp) as hour,
COALESCE(SUM(eu), 0) as eu,
COALESCE(SUM(cost), 0) as cost,
COUNT(*) as calls
FROM billing_records
WHERE tenant_id = :tenant_id
AND timestamp > NOW() - INTERVAL '24 hours'
GROUP BY DATE_TRUNC('hour', timestamp)
ORDER BY hour
"""),
{"tenant_id": tenant_id}
)
hourly_rows = hourly_result.fetchall()
return {
"total": float(total_row[0]) if total_row else 0,
"totalCost": float(total_row[1]) if total_row else 0,
"totalCalls": int(total_row[2]) if total_row else 0,
"hourlyData": [
{
"timestamp": row[0].isoformat() if row[0] else None,
"value": float(row[1]),
"cost": float(row[2]),
"calls": int(row[3])
}
for row in hourly_rows
]
}
except Exception as e:
logger.warning(f"获取租户EU消耗失败: {e}")
return {"total": 0, "totalCost": 0, "totalCalls": 0, "hourlyData": []}
async def _get_tenant_model_usage(
self,
session: AsyncSession,
tenant_id: str
) -> Dict[str, Any]:
"""
获取租户的模型使用情况
返回：
- models: 使用的模型列表及其调用次数
- totalCalls: 总调用次数
"""
try:
# 查询模型使用统计
result = await session.execute(
text("""
SELECT
COALESCE(model_name, 'unknown') as model,
COUNT(*) as calls,
COALESCE(SUM(eu), 0) as eu_consumed,
COALESCE(SUM(cost), 0) as cost
FROM billing_records
WHERE tenant_id = :tenant_id
AND timestamp > NOW() - INTERVAL '30 days'
GROUP BY model_name
ORDER BY calls DESC
"""),
{"tenant_id": tenant_id}
)
rows = result.fetchall()
models = []
total_calls = 0
for row in rows:
calls = int(row[1])
total_calls += calls
models.append({
"name": row[0],
"calls": calls,
"euConsumed": float(row[2]),
"cost": float(row[3])
})
return {
"models": models,
"totalCalls": total_calls
}
except Exception as e:
logger.warning(f"获取租户模型使用情况失败: {e}")
return {"models": [], "totalCalls": 0}
async def _get_tenant_weekly_api_calls(
self,
session: AsyncSession,
tenant_id: str
) -> Dict[str, Any]:
"""
获取租户每周API调用统计（7天每天成功/失败）
返回：
- dailyData: 7天每天的调用数据
- totalSuccess: 总成功次数
- totalFailed: 总失败次数
"""
try:
# 查询7天每天的调用统计
result = await session.execute(
text("""
SELECT
DATE(timestamp) as date,
COUNT(*) as total,
COUNT(CASE WHEN status = 'success' OR status = 'completed' THEN 1 END) as success,
COUNT(CASE WHEN status = 'failed' OR status = 'error' THEN 1 END) as failed
FROM billing_records
WHERE tenant_id = :tenant_id
AND timestamp > NOW() - INTERVAL '7 days'
GROUP BY DATE(timestamp)
ORDER BY date
"""),
{"tenant_id": tenant_id}
)
rows = result.fetchall()
daily_data = []
total_success = 0
total_failed = 0
# 生成过去7天的日期列表
today = datetime.utcnow().date()
date_map = {}
for row in rows:
if row[0]:
date_map[row[0]] = {
"total": int(row[1]),
"success": int(row[2]),
"failed": int(row[3])
}
total_success += int(row[2])
total_failed += int(row[3])
# 填充7天数据（包括没有数据的日期）
for i in range(6, -1, -1):
date = today - timedelta(days=i)
if date in date_map:
daily_data.append({
"date": date.isoformat(),
"total": date_map[date]["total"],
"success": date_map[date]["success"],
"failed": date_map[date]["failed"]
})
else:
daily_data.append({
"date": date.isoformat(),
"total": 0,
"success": 0,
"failed": 0
})
return {
"dailyData": daily_data,
"totalSuccess": total_success,
"totalFailed": total_failed,
"total": total_success + total_failed
}
except Exception as e:
logger.warning(f"获取租户每周API调用统计失败: {e}")
# 返回空的7天数据
today = datetime.utcnow().date()
daily_data = [
{"date": (today - timedelta(days=i)).isoformat(), "total": 0, "success": 0, "failed": 0}
for i in range(6, -1, -1)
]
return {"dailyData": daily_data, "totalSuccess": 0, "totalFailed": 0, "total": 0}
# 全局监控实例