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

更新大志备注

Browse Source
This commit is contained in:
zhanggangyong
2026-01-16 10:38:17 +00:00
parent 9ceb72abf1
commit 6d164c838b
16 changed files with 1343 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/mcp-server/app/application.py
+16 -10
View File
@@ -46,21 +46,27 @@ def create_app() -> FastAPI:
if any(request.url.path.startswith(p) for p in skip_paths):
return await call_next(request)
# 认证逻辑：在独立的数据库会话中完成，不要包裹 call_next
principal = None
try:
async with AsyncSessionLocal() as session:
principal = await authenticate_request(request, session)
if principal:
request.state.principal = principal
elif request.url.path.startswith("/api"):
# Allow unauthenticated access for checklist placeholder APIs while keeping
# any provided principal for future auth-enabled endpoints.
request.state.principal = {}
response = await call_next(request)
return response
except Exception as e:
import logging
logging.getLogger(__name__).error(f"Auth middleware error: {e}")
logging.getLogger(__name__).error(f"Auth middleware error during authentication: {e}")
principal = None
# 设置 principal 到 request.state
if principal:
request.state.principal = principal
elif request.url.path.startswith("/api") or request.url.path.startswith("/agents"):
# Allow unauthenticated access for checklist placeholder APIs while keeping
# any provided principal for future auth-enabled endpoints.
request.state.principal = {}
return await call_next(request)
else:
request.state.principal = {}
# call_next 在数据库会话关闭后调用，避免长时间持有连接
return await call_next(request)
return app
services/mcp-server/app/billing.py
+27
View File
@@ -1,5 +1,32 @@
"""
计费与资源管理逻辑
EU（执行单元）计算说明：
=======================
系统中有两种EU计算方式：
1. Agent 运行时长计费（本模块）
- 计算公式：EU = ceil(duration_seconds / 10)
- 即：1 EU = 10秒运行时间，不足10秒按1 EU计算
- 适用于：平台Agent、自定义Agent的Pod运行时长
- 周期性统计：每小时更新一次计费记录（periodic_billing.py）
2. 模型 Token 计费（billing_webhook.py）
- 计算公式：EU = total_tokens * MODEL_EU_RATE[model_name]
- 不同模型有不同的转换率：
- gpt-4: 0.0001 EU/token (1000 tokens = 0.1 EU)
- gpt-3.5-turbo: 0.00005 EU/token (1000 tokens = 0.05 EU)
- claude: 0.0001 EU/token
- 默认: 0.0001 EU/token
- 适用于：LiteLLM模型调用的Token消耗
数据存储：
- Agent计费记录：agent_billing_records 表
- 模型计费记录：model_billing_records 表
- 用户余额：balances 表（Balance模型）
注意：User.eu_balance 和 User.balance 字段已废弃，请使用 Balance 表
"""
import math
services/mcp-server/app/periodic_billing.py
+16 -6
View File
@@ -2,14 +2,24 @@
周期性计费任务模块
功能：
1. 定时更新运行中 Agent 的 EU 消耗和成本
2. 周期性扣款（每小时）
3. 支持平台 Agent 和自定义 Agent
1. 定时更新运行中 Agent 的计费日志（agent_billing_records 表）
2. 增量扣款：只扣除上次计费后新增的费用，避免重复扣款
3. 余额不足时自动停止 Agent
计费规则：
- VM计算: 0.5 EU/hour
- 平台 Agent: 按模板固定价格计费
为什么需要周期性统计：
- Agent 可能长时间运行，需要定期更新计费记录以便用户查看实时消费
- 避免 Agent 结束回调失败导致计费记录不准确
- 及时发现余额不足的用户并停止其 Agent，防止欠费
EU计算（Agent运行时长）：
- 公式：EU = ceil(duration_seconds / 10)
- 即：1 EU = 10秒运行时间，不足10秒按1 EU计算
成本计算：
- 平台 Agent: 按模板固定价格计费（$/小时）
- 自定义 Agent: 按资源使用量（CPU/内存）计费
执行周期：每小时执行一次（BILLING_INTERVAL_SECONDS = 3600）
"""
import asyncio
services/mcp-server/app/routes/admin.py
+235 -93
View File
@@ -551,7 +551,7 @@ async def get_admin_dashboard_stats(
"totalTenants": total_tenants,
"totalAgents": total_agents,
"totalCalls": total_calls,
"totalRevenue": total_revenue,
"totalRevenue": round(total_revenue, 2),
# 总资源分配（平台端 + 自定义）
"totalAllocatedCpu": round(total_cpu, 2),
"totalAllocatedMemory": round(total_memory, 2),
@@ -1543,42 +1543,60 @@ async def allocate_channel_resources(
# LiteLLM 更新成功后，更新本地数据库
# 删除现有模型资源分配
# 分配模型资源（追加模式：只添加新模型，保留已有模型）
# 存储每个模型名称，而不是供应商 ID
# 这样渠道给租户分配模型时才能正确验证权限
from sqlalchemy import delete
await db.execute(
delete(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == channel_id,
ResourceAllocation.target_type == "channel",
ResourceAllocation.resource_type == "model"
for model_name in all_model_names:
# 检查是否已存在该模型的分配
existing_model_result = await db.execute(
select(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == channel_id,
ResourceAllocation.target_type == "channel",
ResourceAllocation.resource_type == "model",
ResourceAllocation.resource_id == model_name
)
)
)
)
# 分配模型资源（存储每个模型名称，而不是供应商 ID）
# 这样渠道给租户分配模型时才能正确验证权限
for model_name in all_model_names:
allocation = ResourceAllocation(
target_id=channel_id,
target_type="channel",
resource_type="model",
resource_id=model_name, # 存储模型名称（如 gpt-4o），而不是供应商 ID
)
db.add(allocation)
existing_model = existing_model_result.scalar_one_or_none()
if not existing_model:
# 只有不存在时才添加新分配
allocation = ResourceAllocation(
target_id=channel_id,
target_type="channel",
resource_type="model",
resource_id=model_name, # 存储模型名称（如 gpt-4o），而不是供应商 ID
)
db.add(allocation)
logger.info(f"追加模型分配给渠道 {channel.name}: model={model_name}")
else:
logger.info(f"模型 {model_name} 已分配给渠道 {channel.name}，跳过重复分配")
# Bug 修复：同时创建 ChannelProviderAccess 记录，确保渠道能看到分配的供应商
# 删除旧的供应商访问记录
await db.execute(
delete(ChannelProviderAccess).where(
ChannelProviderAccess.channel_id == channel_id
)
)
# 为每个供应商创建访问权限记录
# 追加模式：只添加新的供应商访问权限，保留已有的
user_id = principal.get("claims", {}).get("user_id") or principal.get("sub")
for provider_id in provider_ids:
try:
provider_uuid = uuid.UUID(provider_id)
# 检查是否已存在该供应商的访问权限
existing_access_result = await db.execute(
select(ChannelProviderAccess).where(
and_(
ChannelProviderAccess.channel_id == channel_id,
ChannelProviderAccess.provider_id == provider_uuid
)
)
)
existing_access = existing_access_result.scalar_one_or_none()
if existing_access:
# 已存在，跳过重复分配
logger.info(f"供应商 {provider_id} 的访问权限已存在于渠道 {channel.name}，跳过重复分配")
continue
# 查询供应商信息
provider_result = await db.execute(
select(ModelProvider).where(ModelProvider.id == provider_uuid)
@@ -1597,70 +1615,86 @@ async def allocate_channel_resources(
approved_at=datetime.utcnow(),
)
db.add(access)
logger.info(f"创建渠道 {channel.name} 的供应商访问权限: {provider.name}")
logger.info(f"追加渠道 {channel.name} 的供应商访问权限: {provider.name}")
except (ValueError, TypeError) as e:
logger.warning(f"无法为供应商 {provider_id} 创建访问权限: {e}")
# 删除现有 Agent 资源分配
await db.execute(
delete(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == channel_id,
ResourceAllocation.target_type == "channel",
ResourceAllocation.resource_type == "agent"
)
)
)
# 分配Agent资源
# 删除现有平台 Agent 配额
await db.execute(
delete(PlatformAgentQuota).where(
and_(
PlatformAgentQuota.target_id == channel_id,
PlatformAgentQuota.target_type == "channel"
)
)
)
# 分配Agent资源（追加模式：在现有配额基础上增加）
# 不再删除现有配额，改为追加
user_id_for_allocation = principal.get("claims", {}).get("user_id") or principal.get("sub")
for agent_alloc in req.agents:
# 1. 创建 ResourceAllocation 记录（兼容旧逻辑）
allocation = ResourceAllocation(
target_id=channel_id,
target_type="channel",
resource_type="agent",
resource_id=agent_alloc.agentId,
quantity=agent_alloc.quantity,
# 1. 查找现有的平台 Agent 配额
existing_quota_result = await db.execute(
select(PlatformAgentQuota).where(
and_(
PlatformAgentQuota.target_id == channel_id,
PlatformAgentQuota.target_type == "channel",
PlatformAgentQuota.template_name == agent_alloc.agentId
)
)
)
db.add(allocation)
existing_quota = existing_quota_result.scalar_one_or_none()
# 2. Bug 修复：同时创建 PlatformAgentQuota 记录，确保渠道能看到配额
quota = PlatformAgentQuota(
target_id=channel_id,
target_type="channel",
template_name=agent_alloc.agentId, # agentId 是模板名称
pod_quota=agent_alloc.quantity,
pod_used=0,
allocated_by=uuid.UUID(user_id_for_allocation) if user_id_for_allocation else None,
allocated_at=datetime.utcnow(),
)
db.add(quota)
if existing_quota:
# 追加模式：在现有配额基础上增加
existing_quota.pod_quota = (existing_quota.pod_quota or 0) + agent_alloc.quantity
existing_quota.allocated_at = datetime.utcnow()
if user_id_for_allocation:
existing_quota.allocated_by = uuid.UUID(user_id_for_allocation)
logger.info(
f"追加平台 Agent 配额给渠道 {channel.name}: "
f"template={agent_alloc.agentId}, 追加量={agent_alloc.quantity}, 新配额={existing_quota.pod_quota}"
)
else:
# 创建新配额记录
quota = PlatformAgentQuota(
target_id=channel_id,
target_type="channel",
template_name=agent_alloc.agentId, # agentId 是模板名称
pod_quota=agent_alloc.quantity,
pod_used=0,
allocated_by=uuid.UUID(user_id_for_allocation) if user_id_for_allocation else None,
allocated_at=datetime.utcnow(),
)
db.add(quota)
logger.info(
f"创建平台 Agent 配额给渠道 {channel.name}: "
f"template={agent_alloc.agentId}, quota={agent_alloc.quantity}"
)
logger.info(
f"分配平台 Agent 配额给渠道 {channel.name}: "
f"template={agent_alloc.agentId}, quota={agent_alloc.quantity}"
# 2. 更新或创建 ResourceAllocation 记录（兼容旧逻辑）
# 先查找是否已存在
existing_alloc_result = await db.execute(
select(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == channel_id,
ResourceAllocation.target_type == "channel",
ResourceAllocation.resource_type == "agent",
ResourceAllocation.resource_id == agent_alloc.agentId
)
)
)
existing_alloc = existing_alloc_result.scalar_one_or_none()
if existing_alloc:
# 追加模式
existing_alloc.quantity = (existing_alloc.quantity or 0) + agent_alloc.quantity
else:
# 创建新记录
allocation = ResourceAllocation(
target_id=channel_id,
target_type="channel",
resource_type="agent",
resource_id=agent_alloc.agentId,
quantity=agent_alloc.quantity,
)
db.add(allocation)
# 更新自定义Agent资源（旧格式，保留兼容）
# Bug 修复：customAgentResources 也需要创建 ChannelCustomAgentQuota 记录
# 追加模式：在现有配额基础上增加
if req.customAgentResources:
channel.custom_agent_cpu = req.customAgentResources.cpu
channel.custom_agent_memory = req.customAgentResources.memory
# 同时更新新格式字段
channel.custom_agent_cpu_quota = req.customAgentResources.cpu
channel.custom_agent_memory_quota = req.customAgentResources.memory
# 查找或创建渠道配额记录
channel_quota_result = await db.execute(
select(ChannelCustomAgentQuota).where(
@@ -1670,9 +1704,16 @@ async def allocate_channel_resources(
channel_quota = channel_quota_result.scalar_one_or_none()
if channel_quota:
# 更新现有配额
channel_quota.cpu_quota = req.customAgentResources.cpu
channel_quota.memory_quota = req.customAgentResources.memory
# 追加模式：在现有配额基础上增加
new_cpu_quota = float(channel_quota.cpu_quota or 0) + req.customAgentResources.cpu
new_memory_quota = float(channel_quota.memory_quota or 0) + req.customAgentResources.memory
channel_quota.cpu_quota = new_cpu_quota
channel_quota.memory_quota = new_memory_quota
# 同步更新渠道表字段
channel.custom_agent_cpu = new_cpu_quota
channel.custom_agent_memory = new_memory_quota
channel.custom_agent_cpu_quota = new_cpu_quota
channel.custom_agent_memory_quota = new_memory_quota
else:
# 创建新配额记录
channel_quota = ChannelCustomAgentQuota(
@@ -1683,13 +1724,15 @@ async def allocate_channel_resources(
memory_allocated=0,
)
db.add(channel_quota)
# 同步更新渠道表字段
channel.custom_agent_cpu = req.customAgentResources.cpu
channel.custom_agent_memory = req.customAgentResources.memory
channel.custom_agent_cpu_quota = req.customAgentResources.cpu
channel.custom_agent_memory_quota = req.customAgentResources.memory
# 更新自定义 Agent 配额（新格式）
# 追加模式：在现有配额基础上增加
if req.customAgentQuota:
# 更新渠道表中的配额字段
channel.custom_agent_cpu_quota = req.customAgentQuota.cpuQuota
channel.custom_agent_memory_quota = req.customAgentQuota.memoryQuota
# 查找或创建渠道配额记录
channel_quota_result = await db.execute(
select(ChannelCustomAgentQuota).where(
@@ -1699,9 +1742,14 @@ async def allocate_channel_resources(
channel_quota = channel_quota_result.scalar_one_or_none()
if channel_quota:
# 更新现有配额
channel_quota.cpu_quota = req.customAgentQuota.cpuQuota
channel_quota.memory_quota = req.customAgentQuota.memoryQuota
# 追加模式：在现有配额基础上增加
new_cpu_quota = float(channel_quota.cpu_quota or 0) + req.customAgentQuota.cpuQuota
new_memory_quota = float(channel_quota.memory_quota or 0) + req.customAgentQuota.memoryQuota
channel_quota.cpu_quota = new_cpu_quota
channel_quota.memory_quota = new_memory_quota
# 更新渠道表中的配额字段
channel.custom_agent_cpu_quota = new_cpu_quota
channel.custom_agent_memory_quota = new_memory_quota
else:
# 创建新配额记录
channel_quota = ChannelCustomAgentQuota(
@@ -1712,6 +1760,9 @@ async def allocate_channel_resources(
memory_allocated=0,
)
db.add(channel_quota)
# 更新渠道表中的配额字段
channel.custom_agent_cpu_quota = req.customAgentQuota.cpuQuota
channel.custom_agent_memory_quota = req.customAgentQuota.memoryQuota
channel.channel_credit = req.channelCredit
@@ -3063,8 +3114,20 @@ async def review_provider_application(
application.review_reason = req.reason
application.reviewed_at = datetime.utcnow()
# 如果批准，创建渠道供应商授权记录
# 如果批准，创建渠道供应商授权记录并更新 LiteLLM team
if req.approved:
# 获取渠道信息
channel_result = await db.execute(
select(Channel).where(Channel.id == application.channel_id)
)
channel = channel_result.scalar_one_or_none()
if not channel:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="渠道不存在"
)
# 检查是否已存在授权
existing_result = await db.execute(
select(ChannelProviderAccess).where(
@@ -3095,6 +3158,77 @@ async def review_provider_application(
approved_at=datetime.utcnow(),
)
db.add(access)
# 更新 LiteLLM team 的 models 列表（添加该供应商的所有模型）
if channel.litellm_team_id and provider.supported_models:
try:
from app.litellm_client import get_litellm_client, LiteLLMClientError
litellm_client = get_litellm_client()
# 获取渠道当前已分配的模型列表
current_models_result = await db.execute(
select(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == application.channel_id,
ResourceAllocation.target_type == "channel",
ResourceAllocation.resource_type == "model"
)
)
)
current_models = {alloc.resource_id for alloc in current_models_result.scalars().all()}
# 添加供应商的模型（去重）
new_models = set(provider.supported_models)
all_models = list(current_models | new_models)
# 更新 LiteLLM team
await litellm_client.update_team(
team_id=channel.litellm_team_id,
models=all_models,
metadata={
"channel_id": str(channel.id),
"channel_name": channel.name,
"channel_email": channel.email,
"models_count": len(all_models),
"provider_ids": [str(application.provider_id)],
}
)
logger.info(f"渠道 {channel.name} 的 LiteLLM Team 模型列表已更新（批准供应商申请）: {all_models}")
# 同时更新 ResourceAllocation 记录（追加模式）
for model_name in new_models:
if model_name not in current_models:
allocation = ResourceAllocation(
target_id=application.channel_id,
target_type="channel",
resource_type="model",
resource_id=model_name,
)
db.add(allocation)
logger.info(f"追加模型分配给渠道 {channel.name}: model={model_name}（来自供应商申请）")
except LiteLLMClientError as e:
logger.error(f"更新渠道 {channel.name} 的 LiteLLM Team 失败（批准供应商申请）: {e}")
await db.rollback()
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"LiteLLM Team 更新失败，供应商申请审批已取消: {str(e)}"
)
except Exception as e:
logger.error(f"更新渠道 {channel.name} 的 LiteLLM 连接失败（批准供应商申请）: {e}")
await db.rollback()
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=f"无法连接 LiteLLM Gateway，供应商申请审批已取消: {str(e)}"
)
elif not channel.litellm_team_id:
# 渠道没有关联的 LiteLLM Team，无法批准申请
await db.rollback()
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="渠道没有关联的 LiteLLM Team，无法批准供应商申请。请先重新创建渠道或联系管理员"
)
await db.commit()
@@ -3860,12 +3994,16 @@ async def allocate_platform_agent_to_channel(
quota = quota_result.scalar_one_or_none()
if quota:
# 更新现有配额
old_quota = quota.pod_quota
quota.pod_quota = pod_quota
# 追加模式：在现有配额基础上增加
old_quota = quota.pod_quota or 0
quota.pod_quota = old_quota + pod_quota
quota.allocated_by = user_id
quota.allocated_at = datetime.utcnow()
message = f"平台 Agent 配额已更新（{old_quota} -> {pod_quota}）"
message = f"平台 Agent 配额已追加（{old_quota} + {pod_quota} = {quota.pod_quota}）"
logger.info(
f"追加平台 Agent 配额给渠道 {channel.name}: "
f"template={template_name}, 追加量={pod_quota}, 新配额={quota.pod_quota}"
)
else:
# 创建新配额记录
quota = PlatformAgentQuota(
@@ -3879,6 +4017,10 @@ async def allocate_platform_agent_to_channel(
)
db.add(quota)
message = "平台 Agent 配额分配成功"
logger.info(
f"创建平台 Agent 配额给渠道 {channel.name}: "
f"template={template_name}, quota={pod_quota}"
)
await db.commit()
services/mcp-server/app/routes/auth.py
+30 -1
View File
@@ -4,7 +4,7 @@
from datetime import timedelta
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy import select
from sqlalchemy import select, and_
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.exc import IntegrityError
import secrets
@@ -602,6 +602,32 @@ async def register(req: UserCreate, db: AsyncSession = Depends(get_db)):
for template in platform_templates:
template_name = template.template
# 获取渠道的配额记录（需要同步更新 pod_used）
channel_quota_result = await db.execute(
select(PlatformAgentQuota)
.where(
and_(
PlatformAgentQuota.target_id == TAIJI_CHANNEL_ID,
PlatformAgentQuota.target_type == "channel",
PlatformAgentQuota.template_name == template_name
)
)
.with_for_update() # 行锁
)
channel_quota = channel_quota_result.scalar_one_or_none()
# 如果渠道没有该模板的配额，跳过
if not channel_quota:
logger.warning(f"渠道没有平台 Agent '{template_name}' 的配额，跳过分配")
continue
# 检查渠道剩余配额是否足够
remaining = channel_quota.pod_quota - (channel_quota.pod_used or 0)
if remaining < 1:
logger.warning(f"渠道平台 Agent '{template_name}' 配额不足，跳过分配")
continue
# 创建平台 Agent 配额记录
platform_quota = PlatformAgentQuota(
target_id=user_id,
@@ -614,6 +640,9 @@ async def register(req: UserCreate, db: AsyncSession = Depends(get_db)):
)
db.add(platform_quota)
# Bug 修复：更新渠道的 pod_used（分配给租户的配额视为渠道已使用的配额）
channel_quota.pod_used = (channel_quota.pod_used or 0) + 1
# 同时创建 ResourceAllocation 记录（兼容旧逻辑）
allocation = ResourceAllocation(
target_id=user_id,
services/mcp-server/app/routes/billing_webhook.py
+15
View File
@@ -1,6 +1,21 @@
"""
LiteLLM Callback Webhook路由
接收LiteLLM的实时Token计费数据
此模块处理两种计费回调：
1. LiteLLM 模型调用回调（/litellm-callback）
- 数据来源：LiteLLM 的 success_callback
- EU计算：基于Token数量和模型类型
- 公式：EU = total_tokens * MODEL_EU_RATE[model_name]
- 存储表：model_billing_records
2. Agent Manager 回调（/agent-callback）
- 数据来源：Agent Manager 的运行结束通知
- EU计算：基于Pod运行时长（调用 app.billing.calculate_eu）
- 公式：EU = ceil(duration_seconds / 10)
- 存储表：agent_billing_records
- 注意：周期性计费任务会预先扣款，此回调只扣除增量部分
"""
import json
import logging
services/mcp-server/app/routes/channel.py
+131 -83
View File
@@ -330,27 +330,37 @@ async def allocate_tenant_resources(
# 获取租户所属渠道（用于验证配额）
tenant_channel_id = tenant.channel_id
# 删除现有资源分配
await db.execute(
select(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == tenant_id,
ResourceAllocation.target_type == "tenant"
)
)
)
# 追加模式：不再删除现有资源分配，而是在原有基础上追加
from sqlalchemy import delete
# 分配平台 Agent 资源（更新 PlatformAgentQuota）
for agent_alloc in req.agents:
# 1. 创建 ResourceAllocation 记录（兼容旧逻辑）
allocation = ResourceAllocation(
target_id=tenant_id,
target_type="tenant",
resource_type="agent",
resource_id=agent_alloc.agentId,
quantity=agent_alloc.quantity,
# 1. 追加模式：查找或创建 ResourceAllocation 记录
existing_alloc_result = await db.execute(
select(ResourceAllocation).where(
and_(
ResourceAllocation.target_id == tenant_id,
ResourceAllocation.target_type == "tenant",
ResourceAllocation.resource_type == "agent",
ResourceAllocation.resource_id == agent_alloc.agentId
)
)
)
db.add(allocation)
existing_alloc = existing_alloc_result.scalar_one_or_none()
if existing_alloc:
# 追加模式：在现有数量基础上增加
existing_alloc.quantity = (existing_alloc.quantity or 0) + agent_alloc.quantity
else:
# 创建新的分配记录
allocation = ResourceAllocation(
target_id=tenant_id,
target_type="tenant",
resource_type="agent",
resource_id=agent_alloc.agentId,
quantity=agent_alloc.quantity,
)
db.add(allocation)
# 2. 更新 PlatformAgentQuota（平台 Agent 配额管理）
# agentId 在这里是模板名称（如 echo_agent）
@@ -402,21 +412,22 @@ async def allocate_tenant_resources(
tenant_quota = tenant_quota_result.scalar_one_or_none()
current_tenant_quota = tenant_quota.pod_quota if tenant_quota else 0
# 计算配额增量
quota_delta = agent_alloc.quantity - current_tenant_quota
# 追加模式：quantity 表示要追加的配额数量
quota_delta = agent_alloc.quantity # 追加的配额量
new_quota = current_tenant_quota + quota_delta # 新的总配额
# 检查增量是否超过剩余配额
# 检查追加量是否超过剩余配额
remaining = channel_quota.pod_quota - other_quota - current_tenant_quota
if quota_delta > remaining:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"平台 Agent '{template_name}' 配额超出渠道剩余配额。渠道剩余: {remaining}，请求增量: {quota_delta}"
detail=f"平台 Agent '{template_name}' 配额超出渠道剩余配额。渠道剩余: {remaining}，请求追加: {quota_delta}"
)
# 更新或创建租户配额记录
if tenant_quota:
# 更新现有配额
tenant_quota.pod_quota = agent_alloc.quantity
# 追加模式：在现有配额基础上增加
tenant_quota.pod_quota = new_quota
tenant_quota.allocated_at = datetime.utcnow()
else:
# 创建新配额记录
@@ -424,7 +435,7 @@ async def allocate_tenant_resources(
target_id=tenant_id,
target_type="tenant",
template_name=template_name,
pod_quota=agent_alloc.quantity,
pod_quota=new_quota,
pod_used=0,
allocated_by=None, # 渠道管理员分配
allocated_at=datetime.utcnow(),
@@ -435,8 +446,8 @@ async def allocate_tenant_resources(
channel_quota.pod_used = (channel_quota.pod_used or 0) + quota_delta
logger.info(
f"分配平台 Agent 配额: template={template_name}, tenant={tenant_id}, "
f"quota={agent_alloc.quantity}, channel_pod_used={channel_quota.pod_used}"
f"追加平台 Agent 配额: template={template_name}, tenant={tenant_id}, "
f"追加量={quota_delta}, 新配额={new_quota}, channel_pod_used={channel_quota.pod_used}"
)
else:
# 渠道没有该平台 Agent 的配额，记录警告但不阻止操作
@@ -598,6 +609,26 @@ async def allocate_tenant_resources(
detail="渠道没有自定义 Agent 配额，请先向管理员申请"
)
# 查找租户现有配额记录
tenant_quota_result = await db.execute(
select(TenantCustomAgentQuota).where(
TenantCustomAgentQuota.tenant_id == tenant_id
)
)
tenant_quota = tenant_quota_result.scalar_one_or_none()
# 获取租户当前配额
current_tenant_cpu = float(tenant_quota.cpu_quota) if tenant_quota else 0
current_tenant_memory = float(tenant_quota.memory_quota) if tenant_quota else 0
# 追加模式：请求值表示要追加的量
cpu_delta = req.customAgentQuota.cpuQuota
memory_delta = req.customAgentQuota.memoryQuota
# 计算新配额
new_cpu_quota = current_tenant_cpu + cpu_delta
new_memory_quota = current_tenant_memory + memory_delta
# 计算渠道已分配给其他租户的配额
other_tenants_quota_result = await db.execute(
select(
@@ -617,61 +648,47 @@ async def allocate_tenant_resources(
other_cpu = float(other_quota.total_cpu or 0) if other_quota else 0
other_memory = float(other_quota.total_memory or 0) if other_quota else 0
# 检查是否超过渠道配额
remaining_cpu = float(channel_quota.cpu_quota) - other_cpu
remaining_memory = float(channel_quota.memory_quota) - other_memory
# 检查追加量是否超过渠道剩余配额
# 渠道剩余 = 渠道配额 - 其他租户已分配 - 当前租户已分配
remaining_cpu = float(channel_quota.cpu_quota) - other_cpu - current_tenant_cpu
remaining_memory = float(channel_quota.memory_quota) - other_memory - current_tenant_memory
if req.customAgentQuota.cpuQuota > remaining_cpu:
if cpu_delta > remaining_cpu:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"CPU 配额超出渠道剩余配额。渠道剩余: {remaining_cpu:.2f} 核，请求: {req.customAgentQuota.cpuQuota:.2f} 核"
detail=f"CPU 配额超出渠道剩余配额。渠道剩余: {remaining_cpu:.2f} 核，请求追加: {cpu_delta:.2f} 核"
)
if req.customAgentQuota.memoryQuota > remaining_memory:
if memory_delta > remaining_memory:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"内存配额超出渠道剩余配额。渠道剩余: {remaining_memory:.2f} GB，请求: {req.customAgentQuota.memoryQuota:.2f} GB"
)
# 查找或创建租户配额记录
tenant_quota_result = await db.execute(
select(TenantCustomAgentQuota).where(
TenantCustomAgentQuota.tenant_id == tenant_id
)
)
tenant_quota = tenant_quota_result.scalar_one_or_none()
# 检查当前使用量是否超过新配额
current_cpu_used = float(tenant_quota.cpu_used) if tenant_quota else 0
current_memory_used = float(tenant_quota.memory_used) if tenant_quota else 0
if req.customAgentQuota.cpuQuota < current_cpu_used:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"CPU 配额不能低于当前使用量。当前使用: {current_cpu_used:.2f} 核，请求设置: {req.customAgentQuota.cpuQuota:.2f} 核"
)
if req.customAgentQuota.memoryQuota < current_memory_used:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"内存配额不能低于当前使用量。当前使用: {current_memory_used:.2f} GB，请求设置: {req.customAgentQuota.memoryQuota:.2f} GB"
detail=f"内存配额超出渠道剩余配额。渠道剩余: {remaining_memory:.2f} GB，请求追加: {memory_delta:.2f} GB"
)
if tenant_quota:
# 更新现有配额
tenant_quota.cpu_quota = req.customAgentQuota.cpuQuota
tenant_quota.memory_quota = req.customAgentQuota.memoryQuota
# 追加模式：在现有配额基础上增加
tenant_quota.cpu_quota = new_cpu_quota
tenant_quota.memory_quota = new_memory_quota
logger.info(
f"追加自定义 Agent 配额: tenant={tenant_id}, "
f"CPU 追加量={cpu_delta}, 新配额={new_cpu_quota}, "
f"内存 追加量={memory_delta}, 新配额={new_memory_quota}"
)
else:
# 创建新配额记录
tenant_quota = TenantCustomAgentQuota(
tenant_id=tenant_id,
cpu_quota=req.customAgentQuota.cpuQuota,
memory_quota=req.customAgentQuota.memoryQuota,
cpu_quota=new_cpu_quota,
memory_quota=new_memory_quota,
cpu_used=0,
memory_used=0,
agent_count=0,
)
db.add(tenant_quota)
logger.info(
f"创建自定义 Agent 配额: tenant={tenant_id}, "
f"CPU={new_cpu_quota}, 内存={new_memory_quota}"
)
# Bug 修复：在查询之前 flush，确保数据库能看到刚才的更改
await db.flush()
@@ -2864,6 +2881,24 @@ async def list_available_platform_agents(
)
quota_map = {q.template_name: q for q in quota_result.scalars().all()}
# 获取渠道下每个模板已分配给租户的配额总和（与 allocate_tenant_resources 保持一致）
tenant_quota_result = await db.execute(
select(
PlatformAgentQuota.template_name,
func.sum(PlatformAgentQuota.pod_quota).label("total_allocated")
)
.select_from(PlatformAgentQuota)
.join(User, PlatformAgentQuota.target_id == User.id)
.where(
and_(
User.channel_id == channel_id,
PlatformAgentQuota.target_type == "tenant"
)
)
.group_by(PlatformAgentQuota.template_name)
)
tenant_allocated_map = {row.template_name: row.total_allocated or 0 for row in tenant_quota_result.all()}
# 获取待审批的申请
pending_result = await db.execute(
select(ResourceApplication).where(
@@ -2879,12 +2914,15 @@ async def list_available_platform_agents(
data = []
for template_name, template in platform_templates.items():
quota = quota_map.get(template_name)
# 使用查询租户配额总和的方式计算剩余配额（与 allocate_tenant_resources 一致）
tenant_allocated = tenant_allocated_map.get(template_name, 0)
pod_remaining = (quota.pod_quota - tenant_allocated) if quota else 0
item = {
**template,
"hasAccess": quota is not None,
"podQuota": quota.pod_quota if quota else 0,
"podUsed": quota.pod_used if quota else 0,
"podRemaining": (quota.pod_quota - quota.pod_used) if quota else 0,
"podUsed": tenant_allocated, # 已分配给租户的配额总和
"podRemaining": pod_remaining,
"pendingApplication": template_name in pending_apps,
}
data.append(item)
@@ -3208,14 +3246,6 @@ async def allocate_platform_agent_to_tenant(
)
other_quota = other_tenants_quota_result.scalar() or 0
# 检查是否超过渠道配额
remaining = channel_quota.pod_quota - other_quota
if req.podQuota > remaining:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"配额超出渠道剩余配额。渠道剩余: {remaining}，请求: {req.podQuota}"
)
# 查找或创建租户配额记录（使用行锁防止并发更新）
tenant_quota_result = await db.execute(
select(PlatformAgentQuota)
@@ -3230,16 +3260,30 @@ async def allocate_platform_agent_to_tenant(
)
tenant_quota = tenant_quota_result.scalar_one_or_none()
# 计算配额变化量（用于更新渠道的 pod_used）
old_tenant_quota = tenant_quota.pod_quota if tenant_quota else 0
old_pod_used = tenant_quota.pod_used if tenant_quota else 0
quota_delta = req.podQuota - old_tenant_quota
# 追加模式：req.podQuota 表示要追加的配额量
current_tenant_quota = tenant_quota.pod_quota if tenant_quota else 0
quota_delta = req.podQuota # 追加量
new_quota = current_tenant_quota + quota_delta # 新的总配额
# 计算需要新启动的 Pod 数量
pods_to_start = req.podQuota - old_pod_used
# 检查追加量是否超过渠道剩余配额
# 渠道剩余 = 渠道配额 - 其他租户已分配 - 当前租户已分配
remaining = channel_quota.pod_quota - other_quota - current_tenant_quota
if quota_delta > remaining:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"配额超出渠道剩余配额。渠道剩余: {remaining}，请求追加: {quota_delta}"
)
old_pod_used = tenant_quota.pod_used if tenant_quota else 0
if tenant_quota:
tenant_quota.pod_quota = req.podQuota
# 追加模式：在现有配额基础上增加
tenant_quota.pod_quota = new_quota
tenant_quota.allocated_at = datetime.utcnow()
logger.info(
f"追加平台 Agent 配额: template={req.templateName}, tenant={tenant_id}, "
f"追加量={quota_delta}, 新配额={new_quota}"
)
else:
# 注意：allocated_by 设置为 None，因为渠道管理员的 JWT sub 字段是渠道 ID 而非用户 ID
# 如果需要记录分配人，应该在 JWT 中添加实际的用户 ID 字段
@@ -3247,12 +3291,16 @@ async def allocate_platform_agent_to_tenant(
target_id=tenant_id,
target_type="tenant",
template_name=req.templateName,
pod_quota=req.podQuota,
pod_quota=new_quota,
pod_used=0,
allocated_by=None, # 渠道管理员的 sub 是渠道 ID，不是用户 ID
allocated_at=datetime.utcnow(),
)
db.add(tenant_quota)
logger.info(
f"创建平台 Agent 配额: template={req.templateName}, tenant={tenant_id}, "
f"配额={new_quota}"
)
# Bug 修复：更新渠道的 pod_used（分配给租户的配额视为渠道已使用的配额）
channel_quota.pod_used = (channel_quota.pod_used or 0) + quota_delta
@@ -3272,11 +3320,11 @@ async def allocate_platform_agent_to_tenant(
"tenantName": tenant.name,
"templateName": req.templateName,
"templateDisplayName": template.get("displayName", req.templateName),
"podQuota": req.podQuota,
"podQuota": new_quota,
"podUsed": tenant_quota.pod_used if tenant_quota else 0,
"podRemaining": req.podQuota - (tenant_quota.pod_used if tenant_quota else 0),
"podRemaining": new_quota - (tenant_quota.pod_used if tenant_quota else 0),
},
message=f"平台 Agent 配额分配成功，租户可在租户端部署"
message=f"平台 Agent 配额追加成功（追加量: {quota_delta}），租户可在租户端部署"
)
services/mcp-server/app/routes/platform_agent_quota.py
+23 -6
View File
@@ -462,34 +462,51 @@ async def allocate_platform_agent_to_tenant(
)
existing_quota = existing_quota_result.scalar_one_or_none()
# 计算配额增量
# 追加模式：request.podQuota 表示要追加的配额量
current_quota = existing_quota.pod_quota if existing_quota else 0
quota_delta = request.podQuota - current_quota
quota_delta = request.podQuota # 追加量
new_quota = current_quota + quota_delta # 新的总配额
# 使用渠道的 pod_used 检查剩余配额（pod_used 表示已分配给租户的配额）
if channel_quota.pod_used + quota_delta > channel_quota.pod_quota:
raise HTTPException(
status_code=400,
detail=f"配额不足，渠道配额: {channel_quota.pod_quota}，已分配: {channel_quota.pod_used}，请求增量: {quota_delta}"
detail=f"配额不足，渠道配额: {channel_quota.pod_quota}，已分配: {channel_quota.pod_used}，请求追加: {quota_delta}"
)
# 更新或创建租户配额
if existing_quota:
# 更新现有配额
existing_quota.pod_quota = request.podQuota
# 追加模式：在现有配额基础上增加
existing_quota.pod_quota = new_quota
existing_quota.allocated_at = datetime.utcnow()
quota = existing_quota
logger.info(
"追加平台Agent配额",
tenant_id=tenant_id,
template=request.templateName,
quota_delta=quota_delta,
new_quota=new_quota
)
else:
# 创建新配额
quota = PlatformAgentQuota(
target_id=tenant_uuid,
target_type="tenant",
template_name=request.templateName,
pod_quota=request.podQuota,
pod_quota=new_quota,
pod_used=0,
allocated_by=uuid.UUID(_get_user_id(current_user))
)
db.add(quota)
logger.info(
"创建平台Agent配额",
tenant_id=tenant_id,
template=request.templateName,
quota=new_quota
)
# Bug 修复：更新渠道的 pod_used（分配给租户的配额视为渠道已使用的配额）
channel_quota.pod_used = (channel_quota.pod_used or 0) + quota_delta
# 立即启动 Pod（平台 Agent 分配时立即启动）
try:
services/mcp-server/app/routes/resource_monitoring.py
+17 -6
View File
@@ -14,7 +14,7 @@ from database import get_db
from app.auth import require_role
from models import (
User, Agent, ResourceUsage, BillingRecord,
AgentTrace
AgentTrace, AgentBillingRecord, ModelBillingRecord
)
@@ -99,12 +99,23 @@ async def get_platform_overview(
# 总活跃 Agent 数
active_agents = platform_agents_count + custom_agents_count
# 5. 本月总EU消费
month_eu_result = await db.execute(
select(func.sum(ResourceUsage.eu_consumed))
.where(ResourceUsage.created_at >= month_start)
# 5. 本月总EU消费（合并 Agent运行时间 + 模型Token 两种计费）
# 5.1 Agent 运行时间计费的 EU（从 agent_billing_records 表）
agent_eu_result = await db.execute(
select(func.sum(AgentBillingRecord.eu_consumed))
.where(AgentBillingRecord.period_start >= month_start)
)
month_total_eu = month_eu_result.scalar() or 0
agent_eu = agent_eu_result.scalar() or 0
# 5.2 模型 Token 计费的 EU（从 model_billing_records 表）
model_eu_result = await db.execute(
select(func.sum(ModelBillingRecord.eu_consumed))
.where(ModelBillingRecord.created_at >= month_start)
)
model_eu = model_eu_result.scalar() or 0
# 5.3 合并两种 EU 消耗
month_total_eu = float(agent_eu or 0) + float(model_eu or 0)
return {
"success": True,
services/mcp-server/app/routes/user.py
+220 -2
View File
@@ -2304,13 +2304,66 @@ async def deploy_platform_agent(
detail=f"您没有使用 {req.agentType} 的权限，请联系渠道管理员分配配额"
)
# 检查 Pod 配额
# 检查 Pod 配额（租户配额）
if quota.pod_used >= quota.pod_quota:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Pod 配额已用完。配额: {quota.pod_quota}，已使用: {quota.pod_used}"
)
# 获取用户所属渠道 ID
user_result = await db.execute(
select(User).where(User.id == user_id)
)
user = user_result.scalar_one_or_none()
if not user or not user.channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取用户渠道信息"
)
user_channel_id = user.channel_id
# 检查渠道配额（渠道层面的总使用量不能超过渠道配额）
channel_quota_result = await db.execute(
select(PlatformAgentQuota)
.where(
and_(
PlatformAgentQuota.target_id == user_channel_id,
PlatformAgentQuota.target_type == "channel",
PlatformAgentQuota.template_name == req.agentType
)
)
)
channel_quota = channel_quota_result.scalar_one_or_none()
if not channel_quota:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"渠道没有 {req.agentType} 的配额"
)
# 查询渠道下所有租户的实际 pod_used 总和
channel_total_used_result = await db.execute(
select(func.sum(PlatformAgentQuota.pod_used).label("total"))
.select_from(PlatformAgentQuota)
.join(User, PlatformAgentQuota.target_id == User.id)
.where(
and_(
User.channel_id == user_channel_id,
PlatformAgentQuota.target_type == "tenant",
PlatformAgentQuota.template_name == req.agentType
)
)
)
channel_total_used = channel_total_used_result.scalar() or 0
# 检查渠道层面的总使用量 + 1 是否超过渠道配额
if channel_total_used + 1 > channel_quota.pod_quota:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"渠道 {req.agentType} 配额已满。渠道配额: {channel_quota.pod_quota}，当前总使用: {channel_total_used}"
)
# 检查用户余额（使用 Balance 表）
balance_result = await db.execute(
select(Balance).where(Balance.user_id == user_id)
@@ -2496,13 +2549,66 @@ async def use_platform_agent(
detail=f"您没有使用 {req.agentType} 的权限，请联系渠道管理员分配配额"
)
# 检查 Pod 配额
# 检查 Pod 配额（租户配额）
if quota.pod_used >= quota.pod_quota:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Pod 配额已用完。配额: {quota.pod_quota}，已使用: {quota.pod_used}"
)
# 获取用户所属渠道 ID
user_result = await db.execute(
select(User).where(User.id == user_id)
)
user = user_result.scalar_one_or_none()
if not user or not user.channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取用户渠道信息"
)
user_channel_id = user.channel_id
# 检查渠道配额（渠道层面的总使用量不能超过渠道配额）
channel_quota_result = await db.execute(
select(PlatformAgentQuota)
.where(
and_(
PlatformAgentQuota.target_id == user_channel_id,
PlatformAgentQuota.target_type == "channel",
PlatformAgentQuota.template_name == req.agentType
)
)
)
channel_quota = channel_quota_result.scalar_one_or_none()
if not channel_quota:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"渠道没有 {req.agentType} 的配额"
)
# 查询渠道下所有租户的实际 pod_used 总和
channel_total_used_result = await db.execute(
select(func.sum(PlatformAgentQuota.pod_used).label("total"))
.select_from(PlatformAgentQuota)
.join(User, PlatformAgentQuota.target_id == User.id)
.where(
and_(
User.channel_id == user_channel_id,
PlatformAgentQuota.target_type == "tenant",
PlatformAgentQuota.template_name == req.agentType
)
)
)
channel_total_used = channel_total_used_result.scalar() or 0
# 检查渠道层面的总使用量 + 1 是否超过渠道配额
if channel_total_used + 1 > channel_quota.pod_quota:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"渠道 {req.agentType} 配额已满。渠道配额: {channel_quota.pod_quota}，当前总使用: {channel_total_used}"
)
try:
client = get_agent_manager_client()
@@ -2521,6 +2627,7 @@ async def use_platform_agent(
)
# 如果有环境变量，使用 create_agent；否则使用 create_platform_agent
env_vars = {} # use_platform_agent 不支持环境变量注入
if env_vars:
result = await client.create_agent(
name=instance_name,
@@ -2824,6 +2931,63 @@ async def create_custom_agent(
detail=f"内存配额不足。剩余: {remaining_memory:.2f} GB，请求: {memory_request:.2f} GB"
)
# 获取用户所属渠道 ID
user_result = await db.execute(
select(User).where(User.id == user_id)
)
user = user_result.scalar_one_or_none()
if not user or not user.channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取用户渠道信息"
)
user_channel_id = user.channel_id
# 检查渠道配额（渠道层面的总使用量不能超过渠道配额）
from models import ChannelCustomAgentQuota
channel_quota_result = await db.execute(
select(ChannelCustomAgentQuota)
.where(ChannelCustomAgentQuota.channel_id == user_channel_id)
)
channel_custom_quota = channel_quota_result.scalar_one_or_none()
if not channel_custom_quota:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="渠道没有自定义 Agent 配额"
)
# 查询渠道下所有租户的实际 CPU/内存使用量总和
channel_usage_result = await db.execute(
select(
func.sum(TenantCustomAgentQuota.cpu_used).label("total_cpu"),
func.sum(TenantCustomAgentQuota.memory_used).label("total_memory")
)
.select_from(TenantCustomAgentQuota)
.join(User, TenantCustomAgentQuota.tenant_id == User.id)
.where(User.channel_id == user_channel_id)
)
channel_usage = channel_usage_result.one()
channel_total_cpu_used = float(channel_usage.total_cpu or 0)
channel_total_memory_used = float(channel_usage.total_memory or 0)
channel_cpu_quota = float(channel_custom_quota.cpu_quota or 0)
channel_memory_quota = float(channel_custom_quota.memory_quota or 0)
# 检查渠道层面的 CPU 使用量 + 请求量是否超过渠道配额
if channel_total_cpu_used + cpu_request > channel_cpu_quota:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"渠道 CPU 配额已满。渠道配额: {channel_cpu_quota:.2f} 核，当前总使用: {channel_total_cpu_used:.2f} 核，请求: {cpu_request:.2f} 核"
)
# 检查渠道层面的内存使用量 + 请求量是否超过渠道配额
if channel_total_memory_used + memory_request > channel_memory_quota:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"渠道内存配额已满。渠道配额: {channel_memory_quota:.2f} GB，当前总使用: {channel_total_memory_used:.2f} GB，请求: {memory_request:.2f} GB"
)
# 验证框架模板类型
framework_template = req.frameworkTemplate or "MCP"
allowed_frameworks = ["A2A", "langchain", "MCP"]
@@ -3362,6 +3526,60 @@ async def scale_custom_agent_api(
detail=f"内存配额不足。剩余: {remaining_memory:.2f} GB，需要增加: {memory_delta:.2f} GB"
)
# 检查渠道配额（扩容时需要检查）
if cpu_delta > 0 or memory_delta > 0:
# 获取用户所属渠道 ID
user_result = await db.execute(
select(User).where(User.id == user_id)
)
user = user_result.scalar_one_or_none()
if not user or not user.channel_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无法获取用户渠道信息"
)
user_channel_id = user.channel_id
# 获取渠道配额
from models import ChannelCustomAgentQuota
channel_quota_result = await db.execute(
select(ChannelCustomAgentQuota)
.where(ChannelCustomAgentQuota.channel_id == user_channel_id)
)
channel_custom_quota = channel_quota_result.scalar_one_or_none()
if channel_custom_quota:
# 查询渠道下所有租户的实际使用量总和
channel_usage_result = await db.execute(
select(
func.sum(TenantCustomAgentQuota.cpu_used).label("total_cpu"),
func.sum(TenantCustomAgentQuota.memory_used).label("total_memory")
)
.select_from(TenantCustomAgentQuota)
.join(User, TenantCustomAgentQuota.tenant_id == User.id)
.where(User.channel_id == user_channel_id)
)
channel_usage = channel_usage_result.one()
channel_total_cpu = float(channel_usage.total_cpu or 0)
channel_total_memory = float(channel_usage.total_memory or 0)
channel_cpu_quota = float(channel_custom_quota.cpu_quota or 0)
channel_memory_quota = float(channel_custom_quota.memory_quota or 0)
# 检查渠道 CPU 配额
if cpu_delta > 0 and channel_total_cpu + cpu_delta > channel_cpu_quota:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"渠道 CPU 配额已满。渠道配额: {channel_cpu_quota:.2f} 核，当前总使用: {channel_total_cpu:.2f} 核"
)
# 检查渠道内存配额
if memory_delta > 0 and channel_total_memory + memory_delta > channel_memory_quota:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"渠道内存配额已满。渠道配额: {channel_memory_quota:.2f} GB，当前总使用: {channel_total_memory:.2f} GB"
)
try:
client = get_agent_manager_client()
services/mcp-server/monitoring.py
+35 -18
View File
@@ -695,11 +695,12 @@ class SystemMonitor:
try:
# Agent运行时长计费 - 独立捕获异常
# 注意：agent_billing_records 表使用 cost 列，没有 eu_consumed 列
# EU计算：基于Agent运行时长，1 EU = 10秒
try:
agent_result = await session.execute(
text("""
SELECT
COALESCE(SUM(eu_consumed), 0) as total_eu,
COALESCE(SUM(cost), 0) as total_cost,
COUNT(*) as total_calls
FROM agent_billing_records
@@ -732,20 +733,36 @@ class SystemMonitor:
logger.warning(f"查询model_billing_records失败: {model_error}")
# 不要回滚，继续使用同一个session
# 每小时消耗 - 仅查询存在的表
# 每小时消耗 - 合并两个计费表的数据
try:
# 尝试查询model_billing_records的每小时数据
# 使用 UNION ALL 合并 agent_billing_records 和 model_billing_records 的每小时数据
hourly_result = await session.execute(
text("""
SELECT
DATE_TRUNC('hour', created_at) as hour,
COALESCE(SUM(eu_consumed), 0) as eu,
COALESCE(SUM(total_cost), 0) as cost,
COUNT(*) as calls
FROM model_billing_records
WHERE tenant_id = :tenant_id
AND created_at > NOW() - INTERVAL '24 hours'
GROUP BY DATE_TRUNC('hour', created_at)
SELECT hour, SUM(eu) as eu, SUM(cost) as cost, SUM(calls) as calls
FROM (
-- Agent 运行时长计费
SELECT
DATE_TRUNC('hour', start_time) as hour,
COALESCE(eu_consumed, 0) as eu,
COALESCE(cost, 0) as cost,
1 as calls
FROM agent_billing_records
WHERE user_id = :tenant_id
AND start_time > NOW() - INTERVAL '24 hours'
UNION ALL
-- 模型 Token 计费
SELECT
DATE_TRUNC('hour', created_at) as hour,
COALESCE(eu_consumed, 0) as eu,
COALESCE(total_cost, 0) as cost,
1 as calls
FROM model_billing_records
WHERE tenant_id = :tenant_id
AND created_at > NOW() - INTERVAL '24 hours'
) combined
GROUP BY hour
ORDER BY hour
"""),
{"tenant_id": tenant_id}
@@ -756,17 +773,17 @@ class SystemMonitor:
# 不要回滚，继续使用同一个session
# 合并数据
# agent_row: (total_cost, total_calls)
# model_row: (total_eu, total_cost, total_calls)
agent_cost = float(agent_row[0]) if agent_row else 0
agent_calls = int(agent_row[1]) if agent_row else 0
# agent_row: (total_eu, total_cost, total_calls) - Agent运行时长计费
# model_row: (total_eu, total_cost, total_calls) - 模型Token计费
agent_eu = float(agent_row[0]) if agent_row else 0
agent_cost = float(agent_row[1]) if agent_row else 0
agent_calls = int(agent_row[2]) if agent_row else 0
model_eu = float(model_row[0]) if model_row else 0
model_cost = float(model_row[1]) if model_row else 0
model_calls = int(model_row[2]) if model_row else 0
# Agent计费使用cost换算EU（假设1 EU = 1 cost单位）
agent_eu = agent_cost
# EU消耗 = Agent运行时长EU + 模型Token EU
total_eu = agent_eu + model_eu
total_cost = agent_cost + model_cost
total_calls = agent_calls + model_calls