Files
taiji-AI-PAD/services/mcp-server/app/routes/user.py
T
2026-02-02 14:08:13 +00:00

4135 lines
141 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
用户侧平台API路由
"""
import logging
from datetime import datetime, timedelta
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import select, func, and_, desc
from sqlalchemy.ext.asyncio import AsyncSession
import uuid
from database import get_db
from models import (
User, Agent, Tool, GatewayAPI,
Workflow, BillingRecord, RechargeRecord, TenantCustomAgentQuota,
PlatformAgentQuota, AgentBillingRecord, TenantModelKey, ModelBillingRecord, Balance,
PlatformAgentTemplateConfig
)
from app.billing import (
get_agent_billing_stats,
calculate_platform_agent_cost,
calculate_agent_cost_by_resources,
deduct_balance,
_parse_cpu_to_cores,
_parse_memory_to_gb,
)
from app.auth import require_auth
from app.schemas import (
SuccessResponse,
DashboardStats,
AgentActivity,
GatewaySelectRequest,
CreateAPIRequest,
PlatformAgentInfo,
DeployAgentRequest,
CreateWorkflowRequest,
BalanceInfo,
RechargeRequest,
RechargeResponse,
BillingHistoryQuery,
BillingHistoryResponse,
ExportResponse,
UsePlatformAgentRequest,
CreateCustomAgentRequest,
ScaleCustomAgentRequest,
)
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/user", tags=["用户侧平台"])
def _calculate_eu(duration_seconds: int) -> int:
"""计算EU:1 EU = 10秒,不足10秒按1 EU计算"""
import math
return math.ceil(duration_seconds / 10)
def _parse_cpu(cpu_str: str) -> float:
"""
解析 CPU 字符串为核心数
支持格式:
- "100m" -> 0.1 核
- "1" -> 1 核
- "1.5" -> 1.5 核
"""
if not cpu_str:
return 0.0
cpu_str = cpu_str.strip().lower()
if cpu_str.endswith("m"):
return float(cpu_str[:-1]) / 1000
else:
return float(cpu_str)
def _parse_memory(memory_str: str) -> float:
"""
解析内存字符串为 GB
支持格式:
- "128Mi" -> 0.125 GB
- "1Gi" -> 1 GB
- "512M" -> 0.5 GB
- "2G" -> 2 GB
"""
if not memory_str:
return 0.0
memory_str = memory_str.strip()
# 处理 Kubernetes 格式
if memory_str.endswith("Gi"):
return float(memory_str[:-2])
elif memory_str.endswith("Mi"):
return float(memory_str[:-2]) / 1024
elif memory_str.endswith("Ki"):
return float(memory_str[:-2]) / (1024 * 1024)
# 处理简化格式
elif memory_str.endswith("G"):
return float(memory_str[:-1])
elif memory_str.endswith("M"):
return float(memory_str[:-1]) / 1024
elif memory_str.endswith("K"):
return float(memory_str[:-1]) / (1024 * 1024)
else:
# 假设是字节
return float(memory_str) / (1024 * 1024 * 1024)
@router.get("/tools/stats", response_model=SuccessResponse)
async def get_tools_stats(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取工具统计数据(用于数据与工具页面)
统计指标:
- totalTools: 用户创建的工具总数(内置工具 + 外部数据工具)
- generatedTools: 外部数据工具数量(已生成的工具)
- activeTools: 活跃的外部数据工具数量(状态为 active)
"""
from models import ExternalDataTool
user_id = principal.get("user_id")
# 1. 统计内置工具数量(Tool 表)
result = await db.execute(
select(func.count(Tool.id))
.where(Tool.owner_id == user_id)
)
builtin_tools_count = result.scalar() or 0
# 2. 统计外部数据工具数量(ExternalDataTool 表)
result = await db.execute(
select(func.count(ExternalDataTool.id))
.where(ExternalDataTool.owner_id == user_id)
)
external_tools_count = result.scalar() or 0
# 3. 统计活跃的外部数据工具(状态为 active)
result = await db.execute(
select(func.count(ExternalDataTool.id))
.where(
and_(
ExternalDataTool.owner_id == user_id,
ExternalDataTool.status == "active"
)
)
)
active_external_tools = result.scalar() or 0
# 总工具数 = 内置工具 + 外部数据工具
total_tools = builtin_tools_count + external_tools_count
return SuccessResponse(
data={
"totalTools": total_tools,
"generatedTools": external_tools_count, # 外部数据工具(已生成)
"activeTools": active_external_tools, # 活跃的外部数据工具
}
)
@router.get("/tools", response_model=SuccessResponse)
async def get_user_tools(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取用户创建的所有工具
返回用户创建的工具列表,包括:
- 工具ID、名称、描述
- 模板名称(如 mysql_agent、postgresql_agent)
- 环境变量配置(敏感信息如密码会脱敏)
- 工具类型、类别
- 创建时间、更新时间
- 是否激活、是否公开
"""
user_id = principal.get("user_id")
# 查询用户创建的所有工具
result = await db.execute(
select(Tool)
.where(Tool.owner_id == user_id)
.order_by(desc(Tool.created_at))
)
tools = result.scalars().all()
# 敏感字段列表(需要脱敏的环境变量键)
SENSITIVE_KEYS = [
"PASSWORD", "SECRET", "KEY", "TOKEN", "CREDENTIAL",
"MYSQL_PASSWORD", "POSTGRES_PASSWORD", "OPENAI_API_KEY",
"API_KEY", "AUTH_TOKEN"
]
def mask_sensitive_value(key: str, value: str) -> str:
"""对敏感值进行脱敏处理"""
key_upper = key.upper()
for sensitive in SENSITIVE_KEYS:
if sensitive in key_upper:
if len(value) > 4:
return value[:2] + "*" * (len(value) - 4) + value[-2:]
return "*" * len(value)
return value
def mask_env_config(env_config: dict) -> dict:
"""脱敏环境变量配置"""
if not env_config:
return {}
return {
k: mask_sensitive_value(k, str(v)) if v else v
for k, v in env_config.items()
}
# 构造返回数据
tools_data = []
for tool in tools:
tools_data.append({
"id": str(tool.id),
"name": tool.name,
"description": tool.description,
"template": tool.template, # 新增:模板名称
"envConfig": mask_env_config(tool.env_config) if tool.env_config else {}, # 新增:脱敏后的环境变量配置
"type": tool.category, # category作为type返回给前端
"category": tool.category,
"endpoint": tool.endpoint,
"method": tool.method,
"created_at": tool.created_at.isoformat() if tool.created_at else None,
"updated_at": tool.updated_at.isoformat() if tool.updated_at else None,
"is_active": tool.is_active,
"is_public": tool.is_public,
})
return SuccessResponse(
data={"tools": tools_data}
)
@router.get("/dashboard/stats", response_model=SuccessResponse)
async def get_dashboard_stats(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取用户仪表板统计数据
"""
from monitoring import system_monitor
user_id = principal.get("user_id")
# 活跃Agent数量
result = await db.execute(
select(func.count(Agent.id))
.where(Agent.owner_id == user_id)
.where(Agent.status == "active")
)
active_agents = result.scalar() or 0
# 总请求数(统计模型调用次数)
result = await db.execute(
select(func.count(ModelBillingRecord.id))
.where(ModelBillingRecord.tenant_id == user_id)
)
total_requests = result.scalar() or 0
# 24小时内的请求数(统计模型调用次数)
time_24h_ago = datetime.utcnow() - timedelta(hours=24)
result = await db.execute(
select(func.count(ModelBillingRecord.id))
.where(ModelBillingRecord.tenant_id == user_id)
.where(ModelBillingRecord.created_at >= time_24h_ago)
)
requests_24h = result.scalar() or 0
# EU余额 - 从 Balance 表查询
result = await db.execute(
select(Balance).where(Balance.user_id == user_id)
)
balance = result.scalar_one_or_none()
if balance is None:
# 如果余额记录不存在,创建一个新的(初始余额为0)
balance = Balance(user_id=user_id, eu_balance=0.0)
db.add(balance)
await db.commit()
await db.refresh(balance)
eu_balance = float(balance.eu_balance)
# 系统健康度(从监控模块获取真实数据)
try:
health_data = await system_monitor.get_system_health()
system_health = health_data.get("score", 100)
except Exception:
system_health = 100 # 默认健康
return SuccessResponse(
data={
"activeAgents": active_agents,
"totalRequests": total_requests,
"requests24h": requests_24h,
"euBalance": eu_balance,
"systemHealth": system_health,
}
)
@router.get("/agents/activity", response_model=SuccessResponse)
async def get_agent_activity(
period: str = Query("7d", pattern="^(7d|30d|90d)$"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取Agent活动数据
"""
user_id = principal.get("user_id")
# 计算时间范围
days_map = {"7d": 7, "30d": 30, "90d": 90}
days = days_map[period]
start_date = datetime.utcnow() - timedelta(days=days)
# 查询计费记录
result = await db.execute(
select(
func.date(BillingRecord.timestamp).label("date"),
BillingRecord.agent_name,
func.count(BillingRecord.id).label("requests")
)
.where(BillingRecord.tenant_id == user_id)
.where(BillingRecord.timestamp >= start_date)
.group_by(func.date(BillingRecord.timestamp), BillingRecord.agent_name)
.order_by(func.date(BillingRecord.timestamp))
)
data = []
for row in result.all():
data.append({
"date": row.date.isoformat() if row.date else "",
"agentName": row.agent_name,
"requests": row.requests,
})
return SuccessResponse(data={"data": data})
@router.get("/dashboard/billing-overview", response_model=SuccessResponse)
async def get_billing_overview(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取计费和资源使用综合仪表板数据
返回:
- 本月费用统计(当前消费、平均每日费用、预计月底费用)
- 与上月对比
- EU消费历史(时间序列图表数据)
- 费用明细(按类别分组)
- 资源使用情况(CPU、内存、存储、API调用的配额和使用量)
"""
user_id = principal.get("user_id")
now = datetime.utcnow()
# ========== 1. 计算本月和上月的时间范围 ==========
month_start = datetime(now.year, now.month, 1)
# 计算上月时间范围
if now.month == 1:
last_month_start = datetime(now.year - 1, 12, 1)
last_month_end = datetime(now.year, 1, 1)
else:
last_month_start = datetime(now.year, now.month - 1, 1)
last_month_end = month_start
# 计算本月总天数
if now.month == 12:
next_month = datetime(now.year + 1, 1, 1)
else:
next_month = datetime(now.year, now.month + 1, 1)
total_days = (next_month - month_start).days
days_passed = now.day
# ========== 2. 查询余额信息 ==========
balance_result = await db.execute(
select(Balance).where(Balance.user_id == user_id)
)
balance = balance_result.scalar_one_or_none()
if balance is None:
# 如果余额记录不存在,创建一个新的(初始余额为0)
balance = Balance(user_id=user_id, eu_balance=0.0)
db.add(balance)
await db.commit()
await db.refresh(balance)
# ========== 3. 查询本月消费 (EU消费) ==========
# 3.1 从BillingRecord表查询(Agent计费)
month_eu_result = await db.execute(
select(func.sum(BillingRecord.eu))
.where(and_(
BillingRecord.tenant_id == user_id,
BillingRecord.timestamp >= month_start
))
)
month_eu_agent = float(month_eu_result.scalar() or 0)
month_cost_agent_result = await db.execute(
select(func.sum(BillingRecord.cost))
.where(and_(
BillingRecord.tenant_id == user_id,
BillingRecord.timestamp >= month_start
))
)
month_cost_agent = float(month_cost_agent_result.scalar() or 0)
# 3.2 从ModelBillingRecord表查询(模型调用计费)
month_eu_model_result = await db.execute(
select(func.sum(ModelBillingRecord.eu_consumed))
.where(and_(
ModelBillingRecord.tenant_id == user_id,
ModelBillingRecord.created_at >= month_start
))
)
month_eu_model = float(month_eu_model_result.scalar() or 0)
month_cost_model_result = await db.execute(
select(func.sum(ModelBillingRecord.total_cost))
.where(and_(
ModelBillingRecord.tenant_id == user_id,
ModelBillingRecord.created_at >= month_start
))
)
month_cost_model = float(month_cost_model_result.scalar() or 0)
# 3.3 汇总本月总消费
month_eu_consumed = month_eu_agent + month_eu_model
month_cost = month_cost_agent + month_cost_model
# ========== 4. 查询上月消费 ==========
# 4.1 从BillingRecord表查询
last_month_eu_agent_result = await db.execute(
select(func.sum(BillingRecord.eu))
.where(and_(
BillingRecord.tenant_id == user_id,
BillingRecord.timestamp >= last_month_start,
BillingRecord.timestamp < last_month_end
))
)
last_month_eu_agent = float(last_month_eu_agent_result.scalar() or 0)
last_month_cost_agent_result = await db.execute(
select(func.sum(BillingRecord.cost))
.where(and_(
BillingRecord.tenant_id == user_id,
BillingRecord.timestamp >= last_month_start,
BillingRecord.timestamp < last_month_end
))
)
last_month_cost_agent = float(last_month_cost_agent_result.scalar() or 0)
# 4.2 从ModelBillingRecord表查询
last_month_eu_model_result = await db.execute(
select(func.sum(ModelBillingRecord.eu_consumed))
.where(and_(
ModelBillingRecord.tenant_id == user_id,
ModelBillingRecord.created_at >= last_month_start,
ModelBillingRecord.created_at < last_month_end
))
)
last_month_eu_model = float(last_month_eu_model_result.scalar() or 0)
last_month_cost_model_result = await db.execute(
select(func.sum(ModelBillingRecord.total_cost))
.where(and_(
ModelBillingRecord.tenant_id == user_id,
ModelBillingRecord.created_at >= last_month_start,
ModelBillingRecord.created_at < last_month_end
))
)
last_month_cost_model = float(last_month_cost_model_result.scalar() or 0)
# 4.3 汇总上月总消费
last_month_eu_consumed = last_month_eu_agent + last_month_eu_model
last_month_cost = last_month_cost_agent + last_month_cost_model
# ========== 5. 计算平均每日费用和预测 ==========
avg_daily_cost = month_cost / days_passed if days_passed > 0 else 0
predicted_monthly_cost = avg_daily_cost * total_days
# 计算与上月对比
cost_comparison = 0
if last_month_cost > 0:
cost_comparison = ((month_cost - last_month_cost) / last_month_cost) * 100
elif month_cost > 0:
cost_comparison = 100 # 上月无消费,本月有消费
# ========== 6. 查询EU消费历史(最近30天,按日期聚合) ==========
history_start = now - timedelta(days=30)
# 6.1 从BillingRecord表查询Agent计费历史
agent_history_result = await db.execute(
select(
func.date(BillingRecord.timestamp).label("date"),
func.sum(BillingRecord.eu).label("eu_consumed"),
func.sum(BillingRecord.cost).label("cost"),
func.count(BillingRecord.id).label("calls")
)
.where(and_(
BillingRecord.tenant_id == user_id,
BillingRecord.timestamp >= history_start
))
.group_by(func.date(BillingRecord.timestamp))
)
# 6.2 从ModelBillingRecord表查询模型调用计费历史
model_history_result = await db.execute(
select(
func.date(ModelBillingRecord.created_at).label("date"),
func.sum(ModelBillingRecord.eu_consumed).label("eu_consumed"),
func.sum(ModelBillingRecord.total_cost).label("cost"),
func.count(ModelBillingRecord.id).label("calls")
)
.where(and_(
ModelBillingRecord.tenant_id == user_id,
ModelBillingRecord.created_at >= history_start
))
.group_by(func.date(ModelBillingRecord.created_at))
)
# 6.3 合并两个表的历史数据
history_dict = {}
for row in agent_history_result.all():
date_str = row.date.isoformat() if row.date else ""
history_dict[date_str] = {
"euConsumed": float(row.eu_consumed or 0),
"cost": float(row.cost or 0),
"calls": int(row.calls or 0)
}
for row in model_history_result.all():
date_str = row.date.isoformat() if row.date else ""
if date_str in history_dict:
history_dict[date_str]["euConsumed"] += float(row.eu_consumed or 0)
history_dict[date_str]["cost"] += float(row.cost or 0)
history_dict[date_str]["calls"] += int(row.calls or 0)
else:
history_dict[date_str] = {
"euConsumed": float(row.eu_consumed or 0),
"cost": float(row.cost or 0),
"calls": int(row.calls or 0)
}
# 6.4 转换为列表并排序
eu_history = []
for date_str in sorted(history_dict.keys()):
data = history_dict[date_str]
eu_history.append({
"date": date_str,
"euConsumed": round(data["euConsumed"], 2),
"cost": round(data["cost"], 2),
"calls": data["calls"]
})
# ========== 7. 查询费用明细(按类别分组) ==========
# 7.1 查询平台Agent费用
platform_agent_result = await db.execute(
select(func.sum(BillingRecord.cost))
.join(Agent, BillingRecord.agent_name == Agent.name)
.where(and_(
BillingRecord.tenant_id == user_id,
BillingRecord.timestamp >= month_start,
Agent.type == "platform"
))
)
platform_agent_cost = float(platform_agent_result.scalar() or 0)
# 7.2 查询自定义Agent费用
custom_agent_result = await db.execute(
select(func.sum(BillingRecord.cost))
.join(Agent, BillingRecord.agent_name == Agent.name)
.where(and_(
BillingRecord.tenant_id == user_id,
BillingRecord.timestamp >= month_start,
Agent.type == "custom"
))
)
custom_agent_cost = float(custom_agent_result.scalar() or 0)
# 7.3 查询模型API费用(从ModelBillingRecord表)- 已在前面查询过,直接使用
model_api_cost = month_cost_model
# 7.4 计算未分类费用(总费用 - 已分类费用)
categorized_cost = platform_agent_cost + custom_agent_cost + model_api_cost
other_cost = max(0, month_cost - categorized_cost)
# 构建费用明细数组
cost_categories = []
total_category_cost = month_cost if month_cost > 0 else 1 # 避免除零
if platform_agent_cost > 0 or month_cost == 0:
cost_categories.append({
"category": "platform_agent",
"name": "平台Agent",
"cost": round(platform_agent_cost, 2),
"percentage": round(platform_agent_cost / total_category_cost * 100, 1) if month_cost > 0 else 0
})
if custom_agent_cost > 0 or month_cost == 0:
cost_categories.append({
"category": "custom_agent",
"name": "自定义Agent",
"cost": round(custom_agent_cost, 2),
"percentage": round(custom_agent_cost / total_category_cost * 100, 1) if month_cost > 0 else 0
})
if model_api_cost > 0 or month_cost == 0:
cost_categories.append({
"category": "model_api",
"name": "模型API",
"cost": round(model_api_cost, 2),
"percentage": round(model_api_cost / total_category_cost * 100, 1) if month_cost > 0 else 0
})
if other_cost > 0:
cost_categories.append({
"category": "other",
"name": "其他",
"cost": round(other_cost, 2),
"percentage": round(other_cost / total_category_cost * 100, 1) if month_cost > 0 else 0
})
# ========== 8. 查询资源配额和使用情况 ==========
# 8.1 查询自定义Agent配额
quota_result = await db.execute(
select(TenantCustomAgentQuota)
.where(TenantCustomAgentQuota.tenant_id == user_id)
)
quota = quota_result.scalar_one_or_none()
if not quota:
# 创建默认配额
quota = TenantCustomAgentQuota(
tenant_id=user_id,
cpu_quota=10.0,
memory_quota=20.0,
cpu_used=0.0,
memory_used=0.0
)
db.add(quota)
await db.commit()
await db.refresh(quota)
cpu_used = float(quota.cpu_used or 0)
cpu_quota = float(quota.cpu_quota or 10.0)
memory_used = float(quota.memory_used or 0)
memory_quota = float(quota.memory_quota or 20.0)
# 8.2 查询本月API调用次数(包含Agent调用和模型调用)
agent_calls_result = await db.execute(
select(func.count(BillingRecord.id))
.where(and_(
BillingRecord.tenant_id == user_id,
BillingRecord.timestamp >= month_start
))
)
agent_calls = int(agent_calls_result.scalar() or 0)
model_calls_result = await db.execute(
select(func.count(ModelBillingRecord.id))
.where(and_(
ModelBillingRecord.tenant_id == user_id,
ModelBillingRecord.created_at >= month_start
))
)
model_calls = int(model_calls_result.scalar() or 0)
api_calls = agent_calls + model_calls
# 8.3 API调用限制(从用户表获取或使用默认值)
user_result = await db.execute(
select(User).where(User.id == user_id)
)
user = user_result.scalar_one_or_none()
api_limit = 100000 # 默认值
if user and hasattr(user, 'monthly_api_limit'):
api_limit = getattr(user, 'monthly_api_limit', 100000)
# 8.4 存储使用情况(暂时使用模拟数据)
storage_used = 0
storage_limit = 1000 # GB
# 构建资源使用数组
resources = [
{
"type": "cpu",
"name": "CPU",
"used": round(cpu_used, 2),
"limit": round(cpu_quota, 2),
"unit": "核",
"percentage": round(cpu_used / cpu_quota * 100, 1) if cpu_quota > 0 else 0
},
{
"type": "memory",
"name": "内存",
"used": round(memory_used, 2),
"limit": round(memory_quota, 2),
"unit": "GB",
"percentage": round(memory_used / memory_quota * 100, 1) if memory_quota > 0 else 0
},
{
"type": "storage",
"name": "存储",
"used": storage_used,
"limit": storage_limit,
"unit": "GB",
"percentage": round(storage_used / storage_limit * 100, 1) if storage_limit > 0 else 0
},
{
"type": "api_calls",
"name": "API调用",
"used": api_calls,
"limit": api_limit,
"unit": "次",
"percentage": round(api_calls / api_limit * 100, 1) if api_limit > 0 else 0
}
]
# ========== 9. 构建完整响应 ==========
return SuccessResponse(
data={
# 本月费用统计
"currentMonth": {
"spent": round(month_cost, 2),
"euConsumed": round(month_eu_consumed, 2),
"avgDailySpent": round(avg_daily_cost, 2),
"predictedTotal": round(predicted_monthly_cost, 2),
"daysElapsed": days_passed,
"totalDays": total_days,
"currency": "CNY"
},
# 上月费用
"lastMonth": {
"spent": round(last_month_cost, 2),
"euConsumed": round(last_month_eu_consumed, 2)
},
# 与上月对比
"comparison": {
"percentage": round(cost_comparison, 1),
"direction": "up" if cost_comparison > 0 else "down" if cost_comparison < 0 else "stable"
},
# 余额信息
"balance": {
"eu": float(balance.eu_balance),
"cash": 0 # cash_balance 字段已移除,保留字段以兼容前端
},
# EU消费历史(图表数据)
"euHistory": eu_history,
# 费用明细(按类别)
"costBreakdown": {
"categories": cost_categories,
"total": round(month_cost, 2)
},
# 资源使用情况
"resourceUsage": {
"resources": resources
},
# 元数据
"metadata": {
"timestamp": now.isoformat(),
"userId": user_id
}
},
message="成功获取仪表板数据"
)
# ============= 服务网关 =============
@router.post("/gateway/select", response_model=SuccessResponse)
async def select_gateway(
req: GatewaySelectRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
选择网关类型
"""
# 这里只是标记用户选择的网关类型,实际逻辑可以保存到用户配置中
return SuccessResponse(
data={"gatewayType": req.gatewayType},
message=f"已选择 {req.gatewayType} 网关"
)
@router.post("/gateway/api/create", response_model=SuccessResponse)
async def create_gateway_api(
req: CreateAPIRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
创建网关API
"""
user_id = principal.get("user_id")
gateway_api = GatewayAPI(
name=req.name,
method=req.method,
content=req.content,
owner_id=user_id,
)
db.add(gateway_api)
await db.commit()
await db.refresh(gateway_api)
return SuccessResponse(
data={"id": str(gateway_api.id), "name": gateway_api.name},
message="API创建成功"
)
@router.get("/gateway/apis", response_model=SuccessResponse)
async def list_gateway_apis(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取用户的网关API列表
"""
user_id = principal.get("user_id")
result = await db.execute(
select(GatewayAPI).where(GatewayAPI.owner_id == user_id)
)
apis = result.scalars().all()
data = [
{
"id": str(api.id),
"name": api.name,
"method": api.method,
"createdAt": api.created_at.isoformat(),
}
for api in apis
]
return SuccessResponse(data={"apis": data})
@router.get("/gateway/monitoring", response_model=SuccessResponse)
async def get_gateway_monitoring(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取网关监控数据
返回当前租户的网关使用统计:
- uptime: 服务可用率(基于过去24小时的成功率)
- requestsPerMinute: 每分钟请求数(基于过去1小时的平均值)
- averageLatency: 平均延迟(毫秒)
- errorRate: 错误率(百分比)
"""
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 配额 =============
@router.get("/custom-agent-quota", response_model=SuccessResponse)
async def get_my_custom_agent_quota(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取当前租户正在运行的所有平台Agent的资源使用情况
返回数据:
- totalCpu: 所有正在运行的平台Agent使用的总CPU(核心数)
- totalMemory: 所有正在运行的平台Agent使用的总内存(GB)
- agentCount: 正在运行的平台Agent实例数量(包括副本)
- agents: 每个Agent的详细信息
"""
user_id = principal.get("user_id")
# 查询该用户所有正在运行的平台Agent(end_time为NULL表示正在运行)
result = await db.execute(
select(AgentBillingRecord).where(
and_(
AgentBillingRecord.user_id == user_id,
AgentBillingRecord.is_platform_agent == True,
AgentBillingRecord.end_time.is_(None) # 正在运行
)
)
)
running_agents = result.scalars().all()
total_cpu = 0.0
total_memory = 0.0
total_agent_count = 0
agents_detail = []
for agent in running_agents:
# 解析CPU(支持"100m", "0.5", "1"等格式)
cpu_value = _parse_cpu(agent.cpu_used or "0")
# 解析内存(支持"256Mi", "1Gi", "2"等格式)
memory_value = _parse_memory(agent.memory_used or "0")
# 副本数
replicas = agent.replicas or 1
# 计算该Agent的总资源(CPU和内存 × 副本数)
agent_total_cpu = cpu_value * replicas
agent_total_memory = memory_value * replicas
total_cpu += agent_total_cpu
total_memory += agent_total_memory
total_agent_count += replicas
agents_detail.append({
"agentName": agent.agent_name,
"agentType": agent.agent_type,
"templateName": agent.template_name,
"cpuPerPod": cpu_value,
"memoryPerPod": memory_value,
"replicas": replicas,
"totalCpu": agent_total_cpu,
"totalMemory": agent_total_memory,
"startTime": agent.start_time.isoformat() if agent.start_time else None,
})
return SuccessResponse(
data={
"totalCpu": round(total_cpu, 2),
"totalMemory": round(total_memory, 2),
"agentCount": total_agent_count,
"agents": agents_detail,
}
)
# ============= 数据与工具 =============
@router.post("/tools/create", response_model=SuccessResponse)
async def create_tool(
req: dict,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
创建工具(Tools)
支持两种类型的工具:
1. 模板工具(基于 Agent 模板,如 MySQL、PostgreSQL)
2. API 工具(自定义 API 调用配置)
请求体示例(模板工具 - MySQL):
{
"name": "my-mysql-tool",
"description": "我的MySQL数据库连接工具",
"template": "mysql_agent",
"envConfig": {
"MYSQL_HOST": "mysql.example.com",
"MYSQL_USER": "root",
"MYSQL_PASSWORD": "password123",
"MYSQL_DATABASE": "mydb",
"MYSQL_PORT": "3306",
"OPENAI_API_KEY": "sk-xxx"
}
}
请求体示例(API 工具):
{
"name": "weather-query-tool",
"description": "查询天气信息的工具",
"type": "api",
"config": {
"endpoint": "https://api.weather.com/v1/forecast",
"method": "GET",
"apiKey": "your-api-key"
}
}
"""
user_id = principal.get("user_id")
# 获取模板名称(如果有)
template_name = req.get("template")
# 验证模板名称
if template_name:
try:
from app.agent_manager_client import get_agent_manager_client, AgentManagerError
client = get_agent_manager_client()
custom_templates = await client.list_custom_templates()
template_names = [t.template for t in custom_templates]
if template_name not in template_names:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"无效的模板名称: {template_name}。支持的模板: {', '.join(template_names)}"
)
except HTTPException:
raise
except Exception as e:
logger.warning(f"无法获取模板列表进行验证,跳过验证: {str(e)}")
# 检查工具名称是否已存在
result = await db.execute(
select(Tool).where(
and_(
Tool.owner_id == user_id,
Tool.name == req.get("name")
)
)
)
existing_tool = result.scalar_one_or_none()
if existing_tool:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"工具名称 '{req.get('name')}' 已存在"
)
# 从config中提取信息(用于 API 工具)
config = req.get("config", {})
# 根据是否有模板来设置工具类型
if template_name:
# 模板工具:category 设置为 "database",schema 可为空
tool_schema = None
tool_category = "database"
tool_endpoint = None
tool_method = None
tool_auth_type = None
tool_auth_config = {}
else:
# API 工具:保持原有逻辑
tool_schema = {
"type": "object",
"properties": {
"endpoint": {"type": "string"},
"method": {"type": "string"},
}
}
if "schema" in config:
tool_schema = config["schema"]
tool_category = req.get("type", "api")
tool_endpoint = config.get("endpoint")
tool_method = config.get("method", "GET")
tool_auth_type = config.get("authType")
tool_auth_config = {"apiKey": config.get("apiKey")} if config.get("apiKey") else {}
# 创建工具
tool = Tool(
name=req.get("name"),
description=req.get("description"),
category=tool_category,
template=template_name, # 新增:模板名称
env_config=req.get("envConfig", {}), # 新增:环境变量配置
schema=tool_schema,
endpoint=tool_endpoint,
method=tool_method,
auth_type=tool_auth_type,
auth_config=tool_auth_config,
owner_id=user_id,
is_active=True,
is_public=False,
)
db.add(tool)
await db.commit()
await db.refresh(tool)
logger.info(
f"工具创建成功: user_id={user_id}, tool_id={tool.id}, "
f"name={tool.name}, template={template_name}"
)
return SuccessResponse(
data={
"id": str(tool.id),
"name": tool.name,
"template": tool.template,
"type": tool.category,
},
message="工具创建成功"
)
@router.put("/tools/{tool_id}", response_model=SuccessResponse)
async def update_tool(
tool_id: str,
req: dict,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
修改工具功能和配置
请求体示例:
{
"description": "更新后的工具描述",
"config": {
"endpoint": "https://api.weather.com/v2/forecast",
"method": "GET",
"apiKey": "new-api-key"
}
}
"""
user_id = principal.get("user_id")
# 查询工具
try:
tool_uuid = uuid.UUID(tool_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="工具ID格式无效"
)
result = await db.execute(
select(Tool).where(
and_(
Tool.id == tool_uuid,
Tool.owner_id == user_id
)
)
)
tool = result.scalar_one_or_none()
if not tool:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="工具不存在"
)
# 更新工具
if "description" in req:
tool.description = req["description"]
if "config" in req:
config = req["config"]
if "endpoint" in config:
tool.endpoint = config["endpoint"]
if "method" in config:
tool.method = config["method"]
if "apiKey" in config:
tool.auth_config = {"apiKey": config["apiKey"]}
if "schema" in config:
tool.schema = config["schema"]
if "type" in req:
tool.category = req["type"]
if "category" in req:
tool.category = req["category"]
if "is_active" in req:
tool.is_active = req["is_active"]
if "is_public" in req:
tool.is_public = req["is_public"]
tool.updated_at = datetime.utcnow()
await db.commit()
await db.refresh(tool)
return SuccessResponse(
data={
"id": str(tool.id),
"name": tool.name,
"updated_at": tool.updated_at.isoformat(),
},
message="工具更新成功"
)
@router.delete("/tools/{tool_id}", response_model=SuccessResponse)
async def delete_tool(
tool_id: str,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
删除工具
"""
user_id = principal.get("user_id")
# 查询工具
try:
tool_uuid = uuid.UUID(tool_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="工具ID格式无效"
)
result = await db.execute(
select(Tool).where(
and_(
Tool.id == tool_uuid,
Tool.owner_id == user_id
)
)
)
tool = result.scalar_one_or_none()
if not tool:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="工具不存在"
)
# 删除工具
await db.delete(tool)
await db.commit()
return SuccessResponse(
message="工具已删除"
)
# ============= 代理工厂 =============
@router.get("/agents/platform", response_model=SuccessResponse)
async def list_platform_agents(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取平台Agent列表
"""
result = await db.execute(
select(Agent)
.where(Agent.type == "platform")
.where(Agent.status.in_(["available", "active"]))
)
agents = result.scalars().all()
data = [
{
"id": str(agent.id),
"name": agent.name,
"description": agent.description or "",
"category": agent.category or "通用",
"cpu": float(agent.cpu),
"memory": float(agent.memory),
"status": agent.status,
}
for agent in agents
]
return SuccessResponse(data={"data": data})
@router.post("/agents/deploy", response_model=SuccessResponse)
async def deploy_agent(
req: DeployAgentRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
部署平台Agent到Kubernetes
根据配额ID(从 /platform-agents/available 获取)部署平台Agent实例。
流程:
1. 根据 agentId 查询 PlatformAgentQuota 配额记录
2. 验证配额属于当前用户
3. 检查配额是否足够(podRemaining >= instances)
4. 检查用户余额
5. 调用 Agent Manager 创建实例
6. 更新配额使用量
7. 创建 Agent 记录和计费记录
"""
from app.agent_manager_client import get_agent_manager_client, AgentConfig, AgentManagerError
user_id = principal.get("user_id")
channel_id = principal.get("channel_id")
# 查询平台Agent配额记录(使用行锁防止并发部署超限)
quota_result = await db.execute(
select(PlatformAgentQuota)
.where(
and_(
PlatformAgentQuota.id == req.agentId,
PlatformAgentQuota.target_type == "tenant",
PlatformAgentQuota.target_id == user_id
)
)
.with_for_update() # 行锁
)
quota = quota_result.scalar_one_or_none()
if not quota:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Agent不存在或无权访问"
)
# 检查 Pod 配额是否足够
pod_remaining = quota.pod_quota - quota.pod_used
if pod_remaining < req.instances:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Pod配额不足。剩余配额: {pod_remaining},请求实例数: {req.instances}"
)
# 检查用户余额(使用 Balance 表)
balance_result = await db.execute(
select(Balance).where(Balance.user_id == user_id)
)
balance_obj = balance_result.scalar_one_or_none()
if not balance_obj or balance_obj.eu_balance <= 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="余额不足,请先充值后再部署平台 Agent"
)
try:
client = get_agent_manager_client()
# 生成实例名称
safe_template_name = quota.template_name.replace("_", "-")
instance_name = f"{safe_template_name}-{str(user_id)[:8]}-{uuid.uuid4().hex[:6]}"
# 准备环境变量(注入用户模型配置)
env_vars = {}
# 如果用户指定了模型,注入 LiteLLM 相关环境变量
model_name = getattr(req, 'model', None)
if model_name:
tenant_key_result = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == user_id,
TenantModelKey.model_name == model_name,
TenantModelKey.status == "active"
)
)
)
tenant_key = tenant_key_result.scalar_one_or_none()
if tenant_key:
# 解密 LiteLLM Key 并注入环境变量
try:
from app.litellm_client import get_litellm_client
from config import settings
litellm_client = get_litellm_client()
decrypted_key = litellm_client.decrypt_key(tenant_key.litellm_key_hash)
# 注入 LiteLLM 相关环境变量
env_vars["OPENAI_API_BASE"] = settings.litellm_url
env_vars["OPENAI_API_KEY"] = decrypted_key
env_vars["MODEL_NAME"] = model_name
env_vars["LITELLM_MODEL"] = model_name
except Exception as e:
logger.warning(f"注入模型环境变量失败: {str(e)}")
# 创建平台 Agent 配置
agent_config = AgentConfig(
user_id=str(user_id),
cpu_request=quota.cpu_per_pod or "100m",
cpu_limit="500m", # 固定 limit,避免与 request 相等
memory_request=quota.memory_per_pod or "128Mi",
memory_limit="512Mi", # 固定 limit,避免与 request 相等
replicas=req.instances, # 副本数量
)
# 调用 Agent Manager API 创建平台 Agent
# 如果有环境变量,使用 create_agent 而不是 create_platform_agent
if env_vars:
result = await client.create_agent(
name=instance_name,
template=quota.template_name,
config=agent_config,
env=env_vars
)
else:
result = await client.create_platform_agent(
name=instance_name,
template=quota.template_name,
user_id=str(user_id),
config=agent_config
)
# 更新配额使用量
quota.pod_used += req.instances
# 创建 Agent 记录
agent = Agent(
name=instance_name,
type="platform",
template=quota.template_name,
pod_name=result.name,
k8s_namespace=result.namespace,
k8s_status=result.status,
service_port=result.service_port,
owner_id=user_id,
status="active",
cpu_request=quota.cpu_per_pod or "100m",
memory_request=quota.memory_per_pod or "256Mi",
)
db.add(agent)
# 记录计费(包含访问信息)
access_info = result.access_info or {}
billing_record = AgentBillingRecord(
user_id=user_id,
channel_id=channel_id,
agent_type=quota.template_name,
agent_name=instance_name,
is_platform_agent=True,
template_name=quota.template_name,
start_time=datetime.utcnow(),
duration_seconds=0,
cpu_used=quota.cpu_per_pod or "100m",
memory_used=quota.memory_per_pod or "256Mi",
replicas=req.instances,
# ========== 保存访问信息 ==========
external_ip=access_info.get("external_ip"),
domain=access_info.get("domain"),
domain_url=access_info.get("domain_url"),
access_url=access_info.get("recommended") or access_info.get("domain_url"),
service_port=result.service_port,
namespace=result.namespace,
)
db.add(billing_record)
await db.commit()
return SuccessResponse(
data={
"agentId": req.agentId,
"podName": result.name,
"namespace": result.namespace,
"status": result.status,
"servicePort": result.service_port,
"instances": req.instances,
"model": req.model,
"gateway": req.gateway,
"quotaRemaining": quota.pod_quota - quota.pod_used,
},
message=f"Agent {quota.template_name} 部署成功"
)
except AgentManagerError as e:
raise HTTPException(
status_code=e.status_code,
detail={
"error": "deployment_failed",
"message": f"部署失败: {e.message}",
"detail": e.detail
}
)
# ============= 编排中心 =============
@router.post("/workflows/create", response_model=SuccessResponse)
async def create_workflow(
req: CreateWorkflowRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
创建工作流
"""
user_id = principal.get("user_id")
# 验证节点数量
if len(req.nodes) > 3:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="工作流最多支持3个Agent节点"
)
# 转换节点为dict
nodes_data = [node.dict() for node in req.nodes]
workflow = Workflow(
user_id=user_id,
name=req.name,
description=req.description,
gateway=req.gateway,
nodes=nodes_data,
status="active",
)
db.add(workflow)
await db.commit()
await db.refresh(workflow)
return SuccessResponse(
data={"id": str(workflow.id), "name": workflow.name},
message="工作流创建成功"
)
@router.get("/workflows", response_model=SuccessResponse)
async def list_user_workflows(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取用户的工作流列表
"""
user_id = principal.get("user_id")
result = await db.execute(
select(Workflow)
.where(Workflow.user_id == user_id)
.where(Workflow.status != "deleted")
.order_by(desc(Workflow.created_at))
)
workflows = result.scalars().all()
items = [
{
"id": str(wf.id),
"name": wf.name,
"description": wf.description,
"gateway": wf.gateway,
"nodes": wf.nodes or [],
"status": wf.status,
"createdAt": wf.created_at.isoformat() if wf.created_at else None,
}
for wf in workflows
]
return SuccessResponse(data={"items": items, "count": len(items)})
@router.post("/workflows/{workflow_id}/run", response_model=SuccessResponse)
async def run_user_workflow(
workflow_id: str,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
运行工作流
执行指定的工作流,按顺序调用工作流中的各个Agent节点
"""
user_id = principal.get("user_id")
# 查询工作流
try:
wf_uuid = uuid.UUID(workflow_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的工作流ID格式"
)
result = await db.execute(
select(Workflow).where(
and_(
Workflow.id == wf_uuid,
Workflow.user_id == user_id
)
)
)
workflow = result.scalar_one_or_none()
if not workflow:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="工作流不存在或不属于您"
)
# 创建执行记录
execution_id = str(uuid.uuid4())
started_at = datetime.utcnow()
# 模拟执行工作流中的节点
nodes = workflow.nodes or []
node_results = []
for node in nodes:
node_result = {
"nodeId": node.get("agentId"),
"agentName": node.get("agentName"),
"order": node.get("order"),
"status": "completed",
"startedAt": datetime.utcnow().isoformat(),
"completedAt": datetime.utcnow().isoformat(),
"output": {"message": f"Node {node.get('order')} executed successfully"}
}
node_results.append(node_result)
completed_at = datetime.utcnow()
return SuccessResponse(
data={
"executionId": execution_id,
"workflowId": workflow_id,
"workflowName": workflow.name,
"status": "completed",
"startedAt": started_at.isoformat(),
"completedAt": completed_at.isoformat(),
"nodeResults": node_results,
"totalNodes": len(nodes),
"completedNodes": len(nodes),
},
message=f"工作流 {workflow.name} 执行完成"
)
@router.delete("/workflows/{workflow_id}", response_model=SuccessResponse)
async def delete_user_workflow(
workflow_id: str,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
删除工作流
"""
user_id = principal.get("user_id")
# 查询工作流
try:
wf_uuid = uuid.UUID(workflow_id)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的工作流ID格式"
)
result = await db.execute(
select(Workflow).where(
and_(
Workflow.id == wf_uuid,
Workflow.user_id == user_id
)
)
)
workflow = result.scalar_one_or_none()
if not workflow:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="工作流不存在或不属于您"
)
# 软删除
workflow.status = "deleted"
await db.commit()
return SuccessResponse(
data={"deleted": True, "id": workflow_id},
message="工作流已删除"
)
# ============= 模型使用(LiteLLM 集成)=============
# ============= 模型管理 =============
@router.get("/models", response_model=SuccessResponse)
async def get_user_models(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取当前用户可使用的模型列表
返回用户可用的模型基本信息,供创建Agent时选择。
模型由渠道管理员分配给租户,每个模型包含名称、提供商和上下文窗口等信息。
返回数据格式:
{
"models": [
{
"id": "gpt-4o",
"name": "GPT-4o",
"description": "最新的GPT-4优化版本",
"provider": "openai",
"contextWindow": 128000
}
]
}
"""
user_id = principal.get("user_id")
# 查询分配给该租户的模型 Key
result = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == user_id,
TenantModelKey.status == "active"
)
)
)
keys = result.scalars().all()
models = []
for key in keys:
model_id = key.model_name
# 解析提供商(从模型名称推断)
provider = "other"
if "/" in model_id:
provider = model_id.split("/")[0]
elif model_id.startswith("gpt-") or model_id.startswith("o1") or model_id.startswith("dall-e"):
provider = "openai"
elif model_id.startswith("claude"):
provider = "anthropic"
elif model_id.startswith("gemini"):
provider = "google"
elif model_id.startswith("deepseek"):
provider = "deepseek"
# 解析上下文窗口(根据模型名称设置合理的默认值)
context_window = 4096 # 默认值
if "gpt-4" in model_id.lower():
if "turbo" in model_id.lower() or "o" in model_id.lower():
context_window = 128000
else:
context_window = 8192
elif "gpt-3.5" in model_id.lower():
context_window = 16385
elif "claude-3" in model_id.lower():
context_window = 200000
elif "gemini" in model_id.lower():
if "pro" in model_id.lower():
context_window = 1000000
else:
context_window = 32000
# 生成显示名称和描述
display_name = model_id.split("/")[-1] if "/" in model_id else model_id
description = f"{provider.capitalize()} 模型"
if "gpt-4o" in model_id.lower():
display_name = "GPT-4o" if "mini" not in model_id.lower() else "GPT-4o Mini"
description = "最新的GPT-4优化版本" if "mini" not in model_id.lower() else "高性价比的GPT-4轻量版"
elif "gpt-4" in model_id.lower():
display_name = "GPT-4"
description = "OpenAI最先进的语言模型"
elif "gpt-3.5" in model_id.lower():
display_name = "GPT-3.5 Turbo"
description = "快速且经济的模型"
elif "claude-3" in model_id.lower():
if "opus" in model_id.lower():
display_name = "Claude 3 Opus"
description = "Anthropic最强大的模型"
elif "sonnet" in model_id.lower():
display_name = "Claude 3.5 Sonnet"
description = "Anthropic最新的Claude模型"
elif "haiku" in model_id.lower():
display_name = "Claude 3 Haiku"
description = "快速且经济的Claude模型"
elif "gemini" in model_id.lower():
display_name = "Gemini Pro"
description = "Google最新的多模态模型"
models.append({
"id": model_id,
"name": display_name,
"description": description,
"provider": provider,
"contextWindow": context_window,
})
return SuccessResponse(
data={
"models": models,
}
)
@router.get("/models/available", response_model=SuccessResponse)
async def get_available_models(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取当前用户可用的模型列表
返回渠道分配给该租户的模型及其配额信息。
这些模型可以在创建 Agent 时使用。
"""
user_id = principal.get("user_id")
# 查询分配给该租户的模型 Key
result = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == user_id,
TenantModelKey.status == "active"
)
)
)
keys = result.scalars().all()
models = []
for key in keys:
models.append({
"modelName": key.model_name,
"rpmLimit": key.rpm_limit,
"tpmLimit": key.tpm_limit,
"maxBudget": float(key.max_budget) if key.max_budget else None,
"budgetDuration": key.budget_duration,
"status": key.status,
"allocatedAt": key.created_at.isoformat() if key.created_at else None,
})
return SuccessResponse(
data={
"models": models,
"count": len(models),
}
)
@router.get("/models/usage/stats", response_model=SuccessResponse)
async def get_model_usage_stats(
model_name: Optional[str] = Query(None, description="模型名称,不传则返回所有模型"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取模型使用统计
从 LiteLLM 获取用量数据,包括:
- 请求数
- Token 使用量
- 费用
"""
user_id = principal.get("user_id")
# 查询租户的模型 Key
query = select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == user_id,
TenantModelKey.status == "active"
)
)
if model_name:
query = query.where(TenantModelKey.model_name == model_name)
result = await db.execute(query)
keys = result.scalars().all()
if not keys:
return SuccessResponse(
data={
"models": [],
"totalSpend": 0,
}
)
# 尝试从 LiteLLM 获取用量数据
usage_data = []
total_spend = 0
try:
from app.litellm_client import get_litellm_client, LiteLLMClientError
litellm_client = get_litellm_client()
for key in keys:
try:
# 获取该 Key 的用量
spend_logs = await litellm_client.get_spend_logs(
api_key=key.litellm_key_id
)
# 汇总数据
model_spend = sum(log.get("spend", 0) for log in spend_logs)
model_tokens = sum(log.get("total_tokens", 0) for log in spend_logs)
model_requests = len(spend_logs)
total_spend += model_spend
usage_data.append({
"modelName": key.model_name,
"requests": model_requests,
"totalTokens": model_tokens,
"spend": model_spend,
"rpmLimit": key.rpm_limit,
"tpmLimit": key.tpm_limit,
"maxBudget": float(key.max_budget) if key.max_budget else None,
"budgetRemaining": float(key.max_budget) - model_spend if key.max_budget else None,
})
except LiteLLMClientError as e:
# 单个模型查询失败,记录但继续
usage_data.append({
"modelName": key.model_name,
"requests": 0,
"totalTokens": 0,
"spend": 0,
"error": str(e),
})
except Exception as e:
# LiteLLM 不可用,返回基本信息
for key in keys:
usage_data.append({
"modelName": key.model_name,
"requests": 0,
"totalTokens": 0,
"spend": 0,
"rpmLimit": key.rpm_limit,
"tpmLimit": key.tpm_limit,
"maxBudget": float(key.max_budget) if key.max_budget else None,
"note": "LiteLLM 服务暂不可用,无法获取用量数据",
})
return SuccessResponse(
data={
"models": usage_data,
"totalSpend": total_spend,
}
)
# ============= 计费与资源 =============
@router.get("/billing/balance", response_model=SuccessResponse)
async def get_balance(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取余额信息
"""
user_id = principal.get("user_id")
result = await db.execute(
select(User).where(User.id == user_id)
)
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="用户不存在"
)
# 计算本月消费
month_start = datetime.utcnow().replace(day=1, hour=0, minute=0, second=0, microsecond=0)
result = await db.execute(
select(func.sum(BillingRecord.cost))
.where(BillingRecord.tenant_id == user_id)
.where(BillingRecord.timestamp >= month_start)
)
monthly_spent = result.scalar() or 0
# 从 Balance 表获取余额
balance_result = await db.execute(
select(Balance).where(Balance.user_id == user_id)
)
balance_obj = balance_result.scalar_one_or_none()
if balance_obj is None:
# 如果余额记录不存在,创建一个新的(初始余额为0)
balance_obj = Balance(user_id=user_id, eu_balance=0.0)
db.add(balance_obj)
await db.commit()
await db.refresh(balance_obj)
return SuccessResponse(
data={
"balance": float(balance_obj.eu_balance),
"monthlySpent": float(monthly_spent),
"currency": "CNY",
}
)
@router.post("/billing/recharge", response_model=SuccessResponse)
async def recharge_balance(
req: RechargeRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
充值余额
"""
user_id = principal.get("user_id")
# 生成订单
order_id = f"ORD{datetime.utcnow().strftime('%Y%m%d%H%M%S')}{uuid.uuid4().hex[:8]}"
# 创建充值记录
recharge = RechargeRecord(
user_id=user_id,
amount=req.amount,
payment_method=req.paymentMethod,
order_id=order_id,
status="pending",
)
db.add(recharge)
await db.commit()
# 生成支付URL(简化处理)
payment_url = f"https://pay.taiji-ai.com/checkout?order_id={order_id}"
return SuccessResponse(
data={
"orderId": order_id,
"amount": float(req.amount),
"paymentUrl": payment_url,
"status": "pending",
}
)
def _parse_datetime(dt_str: str) -> datetime:
"""
解析日期时间字符串,支持多种格式:
- YYYY-MM-DD
- YYYY-MM-DDTHH:MM:SS
- YYYY-MM-DDTHH:MM:SSZ
- YYYY-MM-DDTHH:MM:SS+00:00
返回的 datetime 对象不包含时区信息(naive datetime),
以便与数据库中的 TIMESTAMP WITHOUT TIME ZONE 兼容。
"""
if not dt_str:
raise ValueError("日期时间字符串不能为空")
dt_str = dt_str.strip()
# 移除末尾的 Z 并替换为 +00:00
if dt_str.endswith("Z"):
dt_str = dt_str[:-1] + "+00:00"
result = None
try:
# 尝试解析完整的 ISO 格式
result = datetime.fromisoformat(dt_str)
except ValueError:
pass
if result is None:
# 尝试解析简单日期格式 YYYY-MM-DD
try:
result = datetime.strptime(dt_str, "%Y-%m-%d")
except ValueError:
pass
if result is None:
# 尝试解析带时间的格式
try:
result = datetime.strptime(dt_str, "%Y-%m-%dT%H:%M:%S")
except ValueError:
pass
if result is None:
raise ValueError(f"无法解析日期时间格式: {dt_str}")
# 移除时区信息,返回 naive datetime
if result.tzinfo is not None:
result = result.replace(tzinfo=None)
return result
@router.get("/billing/history", response_model=SuccessResponse)
async def get_billing_history(
startTime: str = Query(...),
endTime: str = Query(...),
customerName: Optional[str] = Query(None),
minCalls: Optional[int] = Query(None),
maxCalls: Optional[int] = Query(None),
export: Optional[str] = Query(None, pattern="^(excel|csv|pdf)$"),
page: int = Query(1, ge=1),
pageSize: int = Query(20, ge=1, le=100),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取计费历史记录
支持的时间格式:
- YYYY-MM-DD (如: 2026-01-01)
- YYYY-MM-DDTHH:MM:SS (如: 2026-01-01T00:00:00)
- YYYY-MM-DDTHH:MM:SSZ (如: 2026-01-01T00:00:00Z)
"""
user_id = principal.get("user_id")
# 解析时间
try:
start_dt = _parse_datetime(startTime)
end_dt = _parse_datetime(endTime)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"时间格式错误: {str(e)}。支持的格式: YYYY-MM-DD 或 YYYY-MM-DDTHH:MM:SSZ"
)
# 构建查询
query = select(BillingRecord).where(
and_(
BillingRecord.tenant_id == user_id,
BillingRecord.timestamp >= start_dt,
BillingRecord.timestamp <= end_dt,
)
)
# 计算总数
count_result = await db.execute(
select(func.count()).select_from(query.subquery())
)
total = count_result.scalar() or 0
# 分页查询
query = query.order_by(desc(BillingRecord.timestamp))
query = query.offset((page - 1) * pageSize).limit(pageSize)
result = await db.execute(query)
records = result.scalars().all()
# 如果是导出请求
if export:
# 简化处理,实际应该生成文件并上传到S3
file_url = f"https://exports.taiji-ai.com/{user_id}/{export}/billing_{datetime.utcnow().strftime('%Y%m%d%H%M%S')}.{export}"
expires_at = (datetime.utcnow() + timedelta(hours=24)).isoformat()
return SuccessResponse(
data={
"fileUrl": file_url,
"format": export,
"expiresAt": expires_at,
}
)
# 返回查询结果
data = [
{
"id": str(record.id),
"timestamp": record.timestamp.isoformat(),
"agentName": record.agent_name,
"duration": record.duration,
"eu": record.eu,
"cost": float(record.cost),
}
for record in records
]
return SuccessResponse(
data={
"total": total,
"records": data,
}
)
# ============= 平台 Agent 使用 =============
@router.get("/platform-agents/available", response_model=SuccessResponse)
async def get_available_platform_agents(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取当前用户可用的平台 Agent 列表
返回渠道分配给该租户的平台 Agent 配额信息,包括已部署和未部署的
"""
user_id = principal.get("user_id")
# 查询分配给该租户的平台 Agent 配额
result = await db.execute(
select(PlatformAgentQuota).where(
and_(
PlatformAgentQuota.target_type == "tenant",
PlatformAgentQuota.target_id == user_id
)
)
)
quotas = result.scalars().all()
# 查询所有模板配置,用于获取管理员设置的CPU和内存
template_configs_result = await db.execute(select(PlatformAgentTemplateConfig))
template_configs = {config.template_name: config for config in template_configs_result.scalars().all()}
agents = []
for quota in quotas:
# 从模板配置中获取管理员设置的CPU和内存限制
template_config = template_configs.get(quota.template_name)
cpu_limit = template_config.cpu_limit if template_config and template_config.cpu_limit else "100m"
memory_limit = template_config.memory_limit if template_config and template_config.memory_limit else "256Mi"
agents.append({
"id": str(quota.id),
"templateName": quota.template_name,
"displayName": quota.template_name.replace("-", " ").replace("_", " ").title(),
"description": f"Platform Agent: {quota.template_name}",
"podQuota": quota.pod_quota,
"podUsed": quota.pod_used,
"podRemaining": quota.pod_quota - quota.pod_used,
"cpuLimit": cpu_limit,
"memoryLimit": memory_limit,
"category": "platform",
"allocatedAt": quota.allocated_at.isoformat() if quota.allocated_at else None,
})
return SuccessResponse(data={"agents": agents})
@router.post("/platform-agents/deploy", response_model=SuccessResponse)
async def deploy_platform_agent(
req: UsePlatformAgentRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
部署平台 Agent(租户手动部署)
用户在租户端手动部署已分配的平台 Agent。
与自定义Agent一样,用户可以控制何时部署和停止。
请求参数:
- agentType: 平台 Agent 模板名称(如 echo_agent, chat_agent)
- params: 可选的运行参数
检查项:
1. 用户是否有该 Agent 类型的配额
2. 配额是否还有剩余
3. 用户余额是否足够
"""
from app.agent_manager_client import get_agent_manager_client, AgentConfig, AgentManagerError
user_id = principal.get("user_id")
channel_id = principal.get("channel_id")
# 检查用户是否有该 Agent 类型的配额(使用行锁防止并发部署超限)
quota_result = await db.execute(
select(PlatformAgentQuota)
.where(
and_(
PlatformAgentQuota.target_type == "tenant",
PlatformAgentQuota.target_id == user_id,
PlatformAgentQuota.template_name == req.agentType
)
)
.with_for_update() # 行锁
)
quota = quota_result.scalar_one_or_none()
if not quota:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"您没有使用 {req.agentType} 的权限,请联系渠道管理员分配配额"
)
# 检查 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)
)
balance_obj = balance_result.scalar_one_or_none()
if not balance_obj or balance_obj.eu_balance <= 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="余额不足,请先充值后再部署平台 Agent"
)
try:
client = get_agent_manager_client()
# 生成实例名称
safe_agent_type = req.agentType.replace("_", "-")
instance_name = f"{safe_agent_type}-{str(user_id)[:8]}-{uuid.uuid4().hex[:6]}"
# 准备环境变量(从用户的模型配置中注入)
env_vars = req.envOverrides or {}
# 查询用户的活跃模型(如果有)并自动注入
tenant_keys_result = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == user_id,
TenantModelKey.status == "active"
)
).limit(1) # 获取第一个活跃模型
)
tenant_key = tenant_keys_result.scalar_one_or_none()
if tenant_key:
# 解密 LiteLLM Key 并注入环境变量
try:
from app.litellm_client import get_litellm_client
from config import settings
litellm_client = get_litellm_client()
decrypted_key = litellm_client.decrypt_key(tenant_key.litellm_key_hash)
# 注入 LiteLLM 相关环境变量(如果用户没有覆盖)
if "OPENAI_API_BASE" not in env_vars:
env_vars["OPENAI_API_BASE"] = settings.litellm_url
if "OPENAI_API_KEY" not in env_vars:
env_vars["OPENAI_API_KEY"] = decrypted_key
if "MODEL_NAME" not in env_vars:
env_vars["MODEL_NAME"] = tenant_key.model_name
if "LITELLM_MODEL" not in env_vars:
env_vars["LITELLM_MODEL"] = tenant_key.model_name
except Exception as e:
logger.warning(f"注入模型环境变量失败: {str(e)}")
# 创建平台 Agent 实例
agent_config = AgentConfig(
user_id=str(user_id),
cpu_request=quota.cpu_per_pod or "100m",
cpu_limit=quota.cpu_per_pod or "500m",
memory_request=quota.memory_per_pod or "128Mi",
memory_limit=quota.memory_per_pod or "512Mi",
replicas=1, # 平台 Agent 默认单副本
)
# 如果有环境变量,使用 create_agent;否则使用 create_platform_agent
if env_vars:
result = await client.create_agent(
name=instance_name,
template=req.agentType,
config=agent_config,
env=env_vars
)
else:
result = await client.create_platform_agent(
name=instance_name,
template=req.agentType,
user_id=str(user_id),
config=agent_config
)
# 更新配额使用量
quota.pod_used += 1
# 创建 Agent 记录
agent = Agent(
name=instance_name,
type="platform",
template=req.agentType,
pod_name=result.name,
k8s_namespace=result.namespace,
k8s_status=result.status,
service_port=result.service_port,
owner_id=user_id,
status="active"
)
db.add(agent)
# 记录计费(包含访问信息)
access_info = result.access_info or {}
billing_record = AgentBillingRecord(
user_id=user_id,
channel_id=channel_id,
agent_type=req.agentType,
agent_name=instance_name,
is_platform_agent=True,
template_name=req.agentType,
start_time=datetime.utcnow(),
duration_seconds=0,
cpu_used=quota.cpu_per_pod or "100m",
memory_used=quota.memory_per_pod or "256Mi",
replicas=1,
# ========== 保存访问信息 ==========
external_ip=access_info.get("external_ip"),
domain=access_info.get("domain"),
domain_url=access_info.get("domain_url"),
access_url=access_info.get("recommended") or access_info.get("domain_url"),
service_port=result.service_port,
namespace=result.namespace,
)
db.add(billing_record)
await db.commit()
return SuccessResponse(
data={
"instanceName": result.name,
"namespace": result.namespace,
"status": result.status,
"servicePort": result.service_port,
"accessInfo": result.access_info,
"domain": access_info.get("domain"),
"domainUrl": access_info.get("domain_url"),
"quotaRemaining": quota.pod_quota - quota.pod_used,
},
message=f"平台 Agent {req.agentType} 部署成功"
)
except AgentManagerError as e:
raise HTTPException(
status_code=e.status_code,
detail={
"error": "agent_deploy_failed",
"message": f"部署失败: {e.message}",
"detail": e.detail
}
)
@router.post("/platform-agents/use", response_model=SuccessResponse)
async def use_platform_agent(
req: UsePlatformAgentRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
使用平台 Agent(已废弃,请使用 /platform-agents/deploy)
用户只需传入查询参数,平台会自动创建或复用 Agent 实例。
会检查用户的 Pod 配额。
"""
from app.agent_manager_client import get_agent_manager_client, AgentConfig, AgentManagerError
user_id = principal.get("user_id")
channel_id = principal.get("channel_id")
# 检查用户是否有该 Agent 类型的配额(使用行锁防止并发部署超限)
quota_result = await db.execute(
select(PlatformAgentQuota)
.where(
and_(
PlatformAgentQuota.target_type == "tenant",
PlatformAgentQuota.target_id == user_id,
PlatformAgentQuota.template_name == req.agentType
)
)
.with_for_update() # 行锁
)
quota = quota_result.scalar_one_or_none()
if not quota:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"您没有使用 {req.agentType} 的权限,请联系渠道管理员分配配额"
)
# 检查 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()
# 生成实例名称
safe_agent_type = req.agentType.replace("_", "-")
instance_name = f"{safe_agent_type}-{user_id[:8]}-{uuid.uuid4().hex[:6]}"
# 创建平台 Agent 实例
agent_config = AgentConfig(
user_id=str(user_id),
cpu_request=quota.cpu_per_pod or "100m",
cpu_limit=quota.cpu_per_pod or "500m",
memory_request=quota.memory_per_pod or "128Mi",
memory_limit=quota.memory_per_pod or "512Mi",
replicas=1, # 平台 Agent 默认单副本
)
# 如果有环境变量,使用 create_agent;否则使用 create_platform_agent
env_vars = {} # use_platform_agent 不支持环境变量注入
if env_vars:
result = await client.create_agent(
name=instance_name,
template=req.agentType,
config=agent_config,
env=env_vars
)
else:
result = await client.create_platform_agent(
name=instance_name,
template=req.agentType,
user_id=str(user_id),
config=agent_config
)
# 更新配额使用量
quota.pod_used += 1
# 记录计费(包含访问信息)
access_info = result.access_info or {}
billing_record = AgentBillingRecord(
user_id=user_id,
channel_id=channel_id,
agent_type=req.agentType,
agent_name=instance_name,
is_platform_agent=True,
template_name=req.agentType,
start_time=datetime.utcnow(),
duration_seconds=0,
cpu_used=quota.cpu_per_pod or "100m",
memory_used=quota.memory_per_pod or "256Mi",
replicas=1,
# ========== 保存访问信息 ==========
external_ip=access_info.get("external_ip"),
domain=access_info.get("domain"),
domain_url=access_info.get("domain_url"),
access_url=access_info.get("recommended") or access_info.get("domain_url"),
service_port=result.service_port,
namespace=result.namespace,
)
db.add(billing_record)
await db.commit()
return SuccessResponse(
data={
"instanceName": result.name,
"namespace": result.namespace,
"status": result.status,
"servicePort": result.service_port,
"accessInfo": result.access_info,
"domain": access_info.get("domain"),
"domainUrl": access_info.get("domain_url"),
"quotaRemaining": quota.pod_quota - quota.pod_used,
},
message=f"平台 Agent {req.agentType} 启动成功"
)
except AgentManagerError as e:
raise HTTPException(
status_code=e.status_code,
detail={
"error": "agent_start_failed",
"message": f"启动失败: {e.message}",
"detail": e.detail
}
)
@router.get("/platform-agents/instances", response_model=SuccessResponse)
async def list_my_platform_agent_instances(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取当前用户的平台 Agent 实例列表
"""
from app.agent_manager_client import get_agent_manager_client, AgentManagerError
user_id = principal.get("user_id")
# 查询用户的活跃计费记录(未结束的实例)
result = await db.execute(
select(AgentBillingRecord).where(
and_(
AgentBillingRecord.user_id == user_id,
AgentBillingRecord.is_platform_agent == True,
AgentBillingRecord.end_time == None
)
)
)
records = result.scalars().all()
instances = []
client = get_agent_manager_client()
for record in records:
try:
# 获取实例状态
agent_status = await client.get_agent_status(record.agent_name)
instances.append({
"instanceName": record.agent_name,
"agentType": record.agent_type,
"status": agent_status.status,
"startTime": record.start_time.isoformat() if record.start_time else None,
"runningSeconds": int((datetime.utcnow() - record.start_time).total_seconds()) if record.start_time else 0,
})
except AgentManagerError:
# 实例可能已被删除
instances.append({
"instanceName": record.agent_name,
"agentType": record.agent_type,
"status": "unknown",
"startTime": record.start_time.isoformat() if record.start_time else None,
"runningSeconds": 0,
})
return SuccessResponse(data={"instances": instances})
# ============= 自定义 Agent 管理 =============
@router.get("/custom-agents/templates", response_model=SuccessResponse)
async def get_custom_agent_templates(
principal: dict = Depends(require_auth),
):
"""
获取创建自定义 Agent 所需的信息
**重要说明**:
自定义 Agent 现在主要通过【外部数据工具】来定义能力,不再依赖预定义的数据库模板(mysql_agent 等已废弃)。
创建自定义 Agent 的推荐流程:
1. 先创建外部数据工具(POST /api/user/external-tools)
2. 可选:将多个工具组合成工具集(POST /api/user/external-toolkits)
3. 创建自定义 Agent 时,选择外部数据工具或工具集(externalTools 或 toolkit 参数)
返回信息:
1. frameworkTemplates: 框架类型列表(MCP/A2A/langchain)
2. platformTemplates: 平台模板列表(可选使用,如需要特定基础镜像)
3. createModes: 创建模式说明
返回格式:
{
"frameworkTemplates": ["MCP", "A2A", "langchain"],
"platformTemplates": [...], // 可选的平台模板
"createModes": {
"recommended": "external_tools",
"modes": [
{"mode": "external_tools", "description": "使用外部数据工具(推荐)"},
{"mode": "template", "description": "使用平台模板(高级)"}
]
}
}
"""
from app.agent_manager_client import get_agent_manager_client, AgentManagerError
# 框架类型(MCP 为默认推荐)
framework_templates = ["MCP", "A2A", "langchain"]
# 从 Agent Manager 获取可用的平台模板(供高级用户选用)
platform_templates = []
try:
client = get_agent_manager_client()
templates = await client.list_templates()
for t in templates:
platform_templates.append({
"template": t.template,
"port": t.port,
"env_info": t.env_info or {},
"description": _get_template_description(t.template)
})
except AgentManagerError as e:
logger.warning(f"无法从 Agent Manager 获取平台模板列表: {str(e)}")
except Exception as e:
logger.warning(f"获取平台模板失败: {str(e)}")
return SuccessResponse(data={
"frameworkTemplates": framework_templates,
"platformTemplates": platform_templates,
"createModes": {
"recommended": "external_tools",
"modes": [
{
"mode": "external_tools",
"description": "使用外部数据工具创建(推荐)",
"params": ["externalTools", "toolkit"],
"note": "无需指定 template,系统自动生成 Agent"
},
{
"mode": "template",
"description": "使用平台模板创建(高级)",
"params": ["template"],
"note": "需要配置环境变量"
}
]
},
# 兼容旧版前端:dataTemplates 返回空数组(数据库模板已废弃)
"dataTemplates": []
})
def _get_template_description(template_name: str) -> str:
"""
获取模板描述
"""
descriptions = {
# 平台模板描述
"echo_agent": "Echo Agent,用于测试",
"search_agent": "搜索 Agent,支持网页搜索",
"search_agent_mcp": "搜索 Agent (MCP 协议)",
"search_agent_a2a": "搜索 Agent (A2A 协议)",
"jina_search_agent": "Jina 搜索 Agent",
"azure_blob_agent": "Azure Blob 存储 Agent",
"azure_blob_agent_mcp": "Azure Blob 存储 Agent (MCP)",
"azure_blob_agent_a2a": "Azure Blob 存储 Agent (A2A)",
"a2a_litellm_agent": "A2A LiteLLM Agent",
"code_ai_agent": "代码 AI Agent",
"microsoft_learn_agent": "Microsoft Learn 文档 Agent",
# 旧版数据库模板(已废弃,仅保留描述供参考)
"mysql_agent": "MySQL 数据库 Agent(已废弃,请使用外部数据工具)",
"postgresql_agent": "PostgreSQL 数据库 Agent(已废弃,请使用外部数据工具)",
}
return descriptions.get(template_name, f"{template_name} Agent")
@router.post("/custom-agents", response_model=SuccessResponse)
async def create_custom_agent(
req: CreateCustomAgentRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
创建自定义 Agent
用户需要提供自己的终结点、密钥等配置。
会检查用户的 CPU/内存配额。
如果用户指定了模型(通过 req.model),会自动注入 LiteLLM 相关环境变量:
- OPENAI_API_BASE: LiteLLM 网关地址
- OPENAI_API_KEY: 租户的 LiteLLM API Key
- MODEL_NAME: 模型名称
"""
from app.agent_manager_client import get_agent_manager_client, AgentConfig, AgentManagerError
from config import settings
user_id = principal.get("user_id")
channel_id = principal.get("channel_id")
# 检查用户配额
quota_result = await db.execute(
select(TenantCustomAgentQuota).where(
TenantCustomAgentQuota.tenant_id == user_id
)
)
quota = quota_result.scalar_one_or_none()
if not quota:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="您没有自定义 Agent 配额,请联系渠道管理员分配"
)
# 解析请求的资源量
cpu_request = _parse_cpu(req.cpuRequest)
memory_request = _parse_memory(req.memoryRequest)
cpu_quota = float(quota.cpu_quota or 0)
memory_quota = float(quota.memory_quota or 0)
cpu_used = float(quota.cpu_used or 0)
memory_used = float(quota.memory_used or 0)
remaining_cpu = cpu_quota - cpu_used
remaining_memory = memory_quota - memory_used
# 检查 CPU 配额
if cpu_request > remaining_cpu:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"CPU 配额不足。剩余: {remaining_cpu:.2f} 核,请求: {cpu_request:.2f} 核"
)
# 检查内存配额
if memory_request > remaining_memory:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
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"
)
from uuid import UUID as PyUUID
from sqlalchemy import or_
# ================================================
# 自定义 Agent 必须使用外部数据工具或工具集
# ================================================
if not req.externalTools and not req.toolkit:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="创建自定义 Agent 必须指定 externalTools(外部数据工具)或 toolkit(工具集)"
)
# 环境变量(由系统自动注入)
env_vars = {}
import json
# ========== 外部数据工具配置传递给 Agent Manager ==========
from models import ExternalDataTool, ExternalToolkit
external_tool_refs = [] # 存储 tool_ref_id 列表
external_tool_ids_to_process = [] # 需要处理的工具 ID 列表
# 优先处理工具集(如果指定了 toolkit)
if req.toolkit:
logger.info(f"处理工具集: user_id={user_id}, toolkit={req.toolkit}")
try:
toolkit_result = await db.execute(
select(ExternalToolkit).where(
ExternalToolkit.id == PyUUID(req.toolkit),
ExternalToolkit.owner_id == PyUUID(user_id)
)
)
toolkit = toolkit_result.scalar_one_or_none()
if not toolkit:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"工具集不存在或无权限: {req.toolkit}"
)
# 获取工具集中的工具 ID
if toolkit.tool_ids:
external_tool_ids_to_process.extend(toolkit.tool_ids)
logger.info(f"从工具集获取工具: toolkit={toolkit.name}, tools={toolkit.tool_ids}")
# 更新工具集使用次数
toolkit.usage_count = (toolkit.usage_count or 0) + 1
except HTTPException:
raise
except Exception as e:
logger.warning(f"查询工具集失败: {req.toolkit}, error={str(e)}")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"无效的工具集 ID: {req.toolkit}"
)
# 合并直接指定的外部工具(如果同时指定了 externalTools)
if req.externalTools and len(req.externalTools) > 0:
for ext_id in req.externalTools:
if ext_id not in external_tool_ids_to_process:
external_tool_ids_to_process.append(ext_id)
# 处理所有需要使用的外部数据工具
if external_tool_ids_to_process:
logger.info(f"处理外部数据工具: user_id={user_id}, tools={external_tool_ids_to_process}")
for ext_tool_id in external_tool_ids_to_process:
try:
ext_tool_result = await db.execute(
select(ExternalDataTool).where(
ExternalDataTool.id == PyUUID(ext_tool_id),
ExternalDataTool.owner_id == PyUUID(user_id)
)
)
ext_tool = ext_tool_result.scalar_one_or_none()
if not ext_tool:
logger.warning(f"外部数据工具不存在或无权限: {ext_tool_id}")
continue
if ext_tool.status != "active":
logger.warning(f"外部数据工具未就绪: {ext_tool_id}, status={ext_tool.status}")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"外部数据工具 '{ext_tool.name}' 尚未就绪,状态: {ext_tool.status}"
)
if not ext_tool.tool_ref_id:
logger.warning(f"外部数据工具缺少 tool_ref_id: {ext_tool_id}")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"外部数据工具 '{ext_tool.name}' 缺少关联标识,请重新创建"
)
external_tool_refs.append(ext_tool.tool_ref_id)
# 更新工具使用次数
ext_tool.usage_count = (ext_tool.usage_count or 0) + 1
except HTTPException:
raise
except Exception as e:
logger.warning(f"查询外部数据工具失败: {ext_tool_id}, error={str(e)}")
if external_tool_refs:
# 传递外部工具的 tool_ref_id 列表给 Agent Manager
env_vars["EXTERNAL_TOOL_REFS"] = json.dumps(external_tool_refs)
logger.info(
f"外部数据工具配置已准备: user_id={user_id}, "
f"tool_count={len(external_tool_refs)}, "
f"tool_refs={external_tool_refs}"
)
# ================================================
# 如果指定了模型,查询租户的 LiteLLM Key 并注入环境变量
model_name = req.model
if model_name:
tenant_key_result = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == user_id,
TenantModelKey.model_name == model_name,
TenantModelKey.status == "active"
)
)
)
tenant_key = tenant_key_result.scalar_one_or_none()
if not tenant_key:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"您没有使用模型 '{model_name}' 的权限,请联系渠道管理员分配"
)
# 解密 LiteLLM Key 并注入环境变量
try:
from app.litellm_client import get_litellm_client
litellm_client = get_litellm_client()
decrypted_key = litellm_client.decrypt_key(tenant_key.litellm_key_hash)
# 注入 LiteLLM 相关环境变量
env_vars["OPENAI_API_BASE"] = settings.litellm_url
env_vars["OPENAI_API_KEY"] = decrypted_key
env_vars["MODEL_NAME"] = model_name
env_vars["LITELLM_MODEL"] = model_name
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"获取模型密钥失败: {str(e)}"
)
# ========== 自动注入 OPENAI_API_KEY(如果未设置) ==========
# 用户创建工具时无需填写 OPENAI_API_KEY,系统自动注入用户的 LiteLLM 密钥
if "OPENAI_API_KEY" not in env_vars or not env_vars.get("OPENAI_API_KEY"):
try:
# 查找用户任意可用的 LiteLLM Key
any_key_result = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == user_id,
TenantModelKey.status == "active"
)
).limit(1)
)
any_tenant_key = any_key_result.scalar_one_or_none()
if any_tenant_key:
from app.litellm_client import get_litellm_client
litellm_client = get_litellm_client()
decrypted_key = litellm_client.decrypt_key(any_tenant_key.litellm_key_hash)
env_vars["OPENAI_API_KEY"] = decrypted_key
# 同时注入 OPENAI_API_BASE(如果未设置)
if "OPENAI_API_BASE" not in env_vars:
env_vars["OPENAI_API_BASE"] = settings.litellm_url
logger.info(f"已自动注入 OPENAI_API_KEY: user_id={user_id}")
else:
logger.warning(f"用户 {user_id} 没有可用的 LiteLLM 密钥,无法自动注入 OPENAI_API_KEY")
except Exception as e:
logger.warning(f"自动注入 OPENAI_API_KEY 失败: {str(e)}")
# ============================================================
try:
client = get_agent_manager_client()
# ========== 自定义 Agent:使用外部数据工具创建 ==========
# 工具已在 Agent Manager 生成(有 tool_ref_id),直接使用 tool_refs 创建 Agent
# =======================================================
if not external_tool_refs:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="没有可用的外部数据工具(需要 tool_ref_id)"
)
logger.info(
f"创建自定义 Agent: name={req.name}, tool_refs={external_tool_refs}"
)
# 创建 Agent 配置
agent_config = AgentConfig(
user_id=str(user_id),
cpu_request=req.cpuRequest,
cpu_limit=req.cpuRequest, # limit 默认和 request 一致
memory_request=req.memoryRequest,
memory_limit=req.memoryRequest, # limit 默认和 request 一致
replicas=1,
)
# 调用 create_agent_with_tools(使用 /external-tools/agents/create-with-tools 接口)
# template 默认和 name 一致
template_name = req.name
result = await client.create_agent_with_tools(
name=req.name,
template=template_name,
tool_refs=external_tool_refs,
config=agent_config,
env=env_vars
)
logger.info(
f"自定义 Agent 创建成功: name={req.name}, "
f"status={result.status}, tool_count={len(external_tool_refs)}"
)
try:
# 更新配额使用量
quota.cpu_used = cpu_used + cpu_request
quota.memory_used = memory_used + memory_request
quota.agent_count = (quota.agent_count or 0) + 1
# 记录计费(包含访问信息)
access_info = result.access_info or {}
billing_record = AgentBillingRecord(
user_id=user_id,
channel_id=channel_id,
agent_type=template_name, # template 默认和 name 一致
agent_name=req.name,
template_name=template_name,
is_platform_agent=False,
start_time=datetime.utcnow(),
cpu_used=req.cpuRequest,
memory_used=req.memoryRequest,
tools_used=[],
# ========== 保存访问信息 ==========
external_ip=access_info.get("external_ip"),
domain=access_info.get("domain"),
domain_url=access_info.get("domain_url"),
access_url=access_info.get("recommended") or access_info.get("domain_url"),
service_port=result.service_port,
namespace=result.namespace,
)
db.add(billing_record)
await db.commit()
except Exception as db_error:
# 数据库操作失败,回滚 Agent Manager 操作
await db.rollback()
logger.error(f"数据库操作失败,正在回滚 Agent 创建: {req.name}")
try:
await client.delete_agent(result.name)
logger.info(f"已成功回滚 Agent 创建: {result.name}")
except Exception as rollback_error:
logger.error(f"回滚 Agent 创建失败: {result.name}, 错误: {str(rollback_error)}")
# 这里可以考虑将失败记录到一个待清理表中,供后台任务处理
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"保存配额信息失败,Agent 已回滚: {str(db_error)}"
)
return SuccessResponse(
data={
"name": result.name,
"namespace": result.namespace,
"status": result.status,
"servicePort": result.service_port,
"accessInfo": result.access_info,
"domain": access_info.get("domain"),
"domainUrl": access_info.get("domain_url"),
"modelInjected": model_name is not None,
"quotaRemaining": {
"cpu": remaining_cpu - cpu_request,
"memory": remaining_memory - memory_request,
}
},
message=f"自定义 Agent {req.name} 创建成功"
)
except AgentManagerError as e:
raise HTTPException(
status_code=e.status_code,
detail={
"error": "create_agent_failed",
"message": f"创建失败: {e.message}",
"detail": e.detail
}
)
@router.delete("/custom-agents/{name}", response_model=SuccessResponse)
async def delete_custom_agent(
name: str,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
删除自定义 Agent
释放 CPU/内存配额
"""
from app.agent_manager_client import get_agent_manager_client, AgentManagerError
user_id = principal.get("user_id")
# 查找计费记录
billing_result = await db.execute(
select(AgentBillingRecord).where(
and_(
AgentBillingRecord.agent_name == name,
AgentBillingRecord.user_id == user_id,
AgentBillingRecord.is_platform_agent == False,
AgentBillingRecord.end_time == None
)
)
)
billing_record = billing_result.scalar_one_or_none()
if not billing_record:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"未找到 Agent {name} 或该 Agent 不属于您"
)
try:
# 先更新计费记录和释放配额(数据库操作)
billing_record.end_time = datetime.utcnow()
duration = (billing_record.end_time - billing_record.start_time).total_seconds()
billing_record.duration_seconds = int(duration)
billing_record.eu_consumed = _calculate_eu(int(duration))
# 计算成本并扣款(自定义Agent按资源使用量计费)
cpu_cores = _parse_cpu_to_cores(billing_record.cpu_used) if billing_record.cpu_used else 0.1
memory_gb = _parse_memory_to_gb(billing_record.memory_used) if billing_record.memory_used else 0.125
cost = calculate_agent_cost_by_resources(cpu_cores, memory_gb, int(duration))
billing_record.cost = float(cost)
# 扣除用户余额
success, message = await deduct_balance(
user_id, cost, db,
f"自定义Agent使用: {name} (运行{int(duration)}秒)"
)
if not success:
logger.warning(f"扣款失败: {message}, 用户: {user_id}, Agent: {name}")
# 释放配额
try:
cpu_released = _parse_cpu(billing_record.cpu_used or "0")
memory_released = _parse_memory(billing_record.memory_used or "0")
except (ValueError, AttributeError) as e:
logger.error(f"解析资源使用量失败: {e}, 使用默认值 0")
cpu_released = 0
memory_released = 0
quota_result = await db.execute(
select(TenantCustomAgentQuota).where(
TenantCustomAgentQuota.tenant_id == user_id
)
)
quota = quota_result.scalar_one_or_none()
if quota:
quota.cpu_used = max(0, float(quota.cpu_used or 0) - cpu_released)
quota.memory_used = max(0, float(quota.memory_used or 0) - memory_released)
quota.agent_count = max(0, (quota.agent_count or 0) - 1)
# 提交数据库更改
await db.commit()
# 数据库操作成功后,再删除 Agent
client = get_agent_manager_client()
try:
await client.delete_agent(name)
except Exception as agent_delete_error:
logger.error(f"Agent Manager 删除失败: {name}, 错误: {str(agent_delete_error)}")
# 注意:配额已释放,但 Agent 可能未删除
# 可以考虑标记为"待清理"状态,供后台任务处理
return SuccessResponse(
message=f"自定义 Agent {name} 已删除",
data={
"quotaReleased": {
"cpu": cpu_released,
"memory": memory_released,
}
}
)
except AgentManagerError as e:
raise HTTPException(
status_code=e.status_code,
detail={
"error": "delete_agent_failed",
"message": f"删除失败: {e.message}",
"detail": e.detail
}
)
@router.put("/custom-agents/{name}/scale", response_model=SuccessResponse)
async def scale_custom_agent_api(
name: str,
req: ScaleCustomAgentRequest,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
扩缩容自定义 Agent
更新 Agent 的资源配置,会检查配额
"""
from app.agent_manager_client import get_agent_manager_client, AgentManagerError
user_id = principal.get("user_id")
# 查找计费记录
billing_result = await db.execute(
select(AgentBillingRecord).where(
and_(
AgentBillingRecord.agent_name == name,
AgentBillingRecord.user_id == user_id,
AgentBillingRecord.is_platform_agent == False,
AgentBillingRecord.end_time == None
)
)
)
billing_record = billing_result.scalar_one_or_none()
if not billing_record:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"未找到 Agent {name} 或该 Agent 不属于您"
)
# 获取用户配额
quota_result = await db.execute(
select(TenantCustomAgentQuota).where(
TenantCustomAgentQuota.tenant_id == user_id
)
)
quota = quota_result.scalar_one_or_none()
if not quota:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="您没有自定义 Agent 配额"
)
# 计算资源变化
current_cpu = _parse_cpu(billing_record.cpu_used or "0")
current_memory = _parse_memory(billing_record.memory_used or "0")
new_cpu = _parse_cpu(req.cpuRequest) if req.cpuRequest else current_cpu
new_memory = _parse_memory(req.memoryRequest) if req.memoryRequest else current_memory
cpu_delta = new_cpu - current_cpu
memory_delta = new_memory - current_memory
# 检查配额(只检查增加的情况)
if cpu_delta > 0:
remaining_cpu = float(quota.cpu_quota or 0) - float(quota.cpu_used or 0)
if cpu_delta > remaining_cpu:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"CPU 配额不足。剩余: {remaining_cpu:.2f} 核,需要增加: {cpu_delta:.2f} 核"
)
if memory_delta > 0:
remaining_memory = float(quota.memory_quota or 0) - float(quota.memory_used or 0)
if memory_delta > remaining_memory:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
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()
# 调用扩缩容 API(注意:Agent Manager 尚未实现此接口)
try:
await client.scale_agent(
agent_name=name,
cpu_request=req.cpuRequest,
cpu_limit=req.cpuLimit,
memory_request=req.memoryRequest,
memory_limit=req.memoryLimit
)
except NotImplementedError:
# Agent Manager 尚未实现扩缩容接口,暂时只更新本地记录
pass
# 更新配额使用量
quota.cpu_used = float(quota.cpu_used or 0) + cpu_delta
quota.memory_used = float(quota.memory_used or 0) + memory_delta
# 更新计费记录
if req.cpuRequest:
billing_record.cpu_used = req.cpuRequest
if req.memoryRequest:
billing_record.memory_used = req.memoryRequest
await db.commit()
return SuccessResponse(
message=f"Agent {name} 扩缩容成功",
data={
"newCpu": req.cpuRequest or billing_record.cpu_used,
"newMemory": req.memoryRequest or billing_record.memory_used,
"quotaRemaining": {
"cpu": float(quota.cpu_quota or 0) - float(quota.cpu_used or 0),
"memory": float(quota.memory_quota or 0) - float(quota.memory_used or 0),
}
}
)
except AgentManagerError as e:
raise HTTPException(
status_code=e.status_code,
detail={
"error": "scale_agent_failed",
"message": f"扩缩容失败: {e.message}",
"detail": e.detail
}
)
@router.get("/custom-agents", response_model=SuccessResponse)
async def list_my_custom_agents(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取当前用户的自定义 Agent 列表
"""
user_id = principal.get("user_id")
# 查询用户的活跃自定义 Agent
result = await db.execute(
select(AgentBillingRecord).where(
and_(
AgentBillingRecord.user_id == user_id,
AgentBillingRecord.is_platform_agent == False,
AgentBillingRecord.end_time == None
)
)
)
records = result.scalars().all()
agents = []
# 尝试获取 Agent Manager 客户端,如果服务不可用则跳过状态查询
try:
from app.agent_manager_client import get_agent_manager_client, AgentManagerError
client = get_agent_manager_client()
agent_manager_available = True
except Exception:
agent_manager_available = False
for record in records:
agent_info = {
"name": record.agent_name,
"template": record.agent_type,
"status": "unknown",
"cpu": record.cpu_used,
"memory": record.memory_used,
"startTime": record.start_time.isoformat() if record.start_time else None,
"runningSeconds": int((datetime.utcnow() - record.start_time).total_seconds()) if record.start_time else 0,
}
if agent_manager_available:
try:
# 获取 Agent 状态
agent_status = await client.get_agent_status(record.agent_name)
agent_info["status"] = agent_status.status
except Exception:
# Agent Manager 服务不可用或 Agent 不存在
pass
agents.append(agent_info)
return SuccessResponse(data={"agents": agents})
@router.get("/custom-agents/{name}/logs", response_model=SuccessResponse)
async def get_custom_agent_logs(
name: str,
tail_lines: int = Query(100, ge=1, le=1000),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取自定义 Agent 日志
"""
from app.agent_manager_client import get_agent_manager_client, AgentManagerError
user_id = principal.get("user_id")
# 验证 Agent 属于该用户
billing_result = await db.execute(
select(AgentBillingRecord).where(
and_(
AgentBillingRecord.agent_name == name,
AgentBillingRecord.user_id == user_id,
AgentBillingRecord.end_time == None
)
)
)
billing_record = billing_result.scalar_one_or_none()
if not billing_record:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"未找到 Agent {name} 或该 Agent 不属于您"
)
try:
client = get_agent_manager_client()
logs = await client.get_agent_logs(name, tail_lines=tail_lines)
return SuccessResponse(data={"logs": logs})
except AgentManagerError as e:
raise HTTPException(
status_code=e.status_code,
detail={
"error": "get_logs_failed",
"message": f"获取日志失败: {e.message}",
}
)
@router.post("/custom-agents/{name}/restart", response_model=SuccessResponse)
async def restart_custom_agent(
name: str,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
重启自定义 Agent
"""
from app.agent_manager_client import get_agent_manager_client, AgentManagerError
user_id = principal.get("user_id")
# 验证 Agent 属于该用户
billing_result = await db.execute(
select(AgentBillingRecord).where(
and_(
AgentBillingRecord.agent_name == name,
AgentBillingRecord.user_id == user_id,
AgentBillingRecord.end_time == None
)
)
)
billing_record = billing_result.scalar_one_or_none()
if not billing_record:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"未找到 Agent {name} 或该 Agent 不属于您"
)
try:
client = get_agent_manager_client()
await client.restart_agent(name)
return SuccessResponse(message=f"Agent {name} 正在重启")
except AgentManagerError as e:
raise HTTPException(
status_code=e.status_code,
detail={
"error": "restart_failed",
"message": f"重启失败: {e.message}",
}
)
# ============= Agent 计费统计 =============
@router.get("/agent-billing/stats", response_model=SuccessResponse)
async def get_my_agent_billing_stats(
period: str = Query("30d", pattern="^(7d|30d|90d)$"),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取当前用户的 Agent 计费统计
返回数据:
- totalCost: 总成本
- totalDurationSeconds: 总运行时长(秒)
- totalRequests: 总请求数
- byAgentType: 按 Agent 类型统计(platform/custom)
- byTemplate: 按模板统计
"""
user_id = principal.get("user_id")
# 计算时间范围
days_map = {"7d": 7, "30d": 30, "90d": 90}
days = days_map[period]
start_date = datetime.utcnow() - timedelta(days=days)
end_date = datetime.utcnow()
stats = await get_agent_billing_stats(user_id, start_date, end_date, db)
return SuccessResponse(data=stats)
@router.get("/agent-billing/history", response_model=SuccessResponse)
async def get_my_agent_billing_history(
startTime: str = Query(...),
endTime: str = Query(...),
agentType: Optional[str] = Query(None, pattern="^(platform|custom)$"),
page: int = Query(1, ge=1),
pageSize: int = Query(20, ge=1, le=100),
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取当前用户的 Agent 计费历史记录
"""
user_id = principal.get("user_id")
# 解析时间
try:
start_dt = _parse_datetime(startTime)
end_dt = _parse_datetime(endTime)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"时间格式错误: {str(e)}。支持的格式: YYYY-MM-DD 或 YYYY-MM-DDTHH:MM:SSZ"
)
# 构建查询
conditions = [
AgentBillingRecord.user_id == user_id,
AgentBillingRecord.start_time >= start_dt,
AgentBillingRecord.start_time <= end_dt,
]
if agentType:
is_platform = agentType == "platform"
conditions.append(AgentBillingRecord.is_platform_agent == is_platform)
query = select(AgentBillingRecord).where(and_(*conditions))
# 计算总数
count_result = await db.execute(
select(func.count()).select_from(query.subquery())
)
total = count_result.scalar() or 0
# 分页查询
query = query.order_by(desc(AgentBillingRecord.start_time))
query = query.offset((page - 1) * pageSize).limit(pageSize)
result = await db.execute(query)
records = result.scalars().all()
# 格式化结果
data = []
for record in records:
data.append({
"id": str(record.id),
"agentName": record.agent_name,
"agentType": "platform" if record.is_platform_agent else "custom",
"templateName": record.agent_type,
"startTime": record.start_time.isoformat() if record.start_time else None,
"endTime": record.end_time.isoformat() if record.end_time else None,
"durationSeconds": record.duration_seconds or 0,
"euConsumed": record.eu_consumed or 0,
"cpuUsed": record.cpu_used,
"memoryUsed": record.memory_used,
"cost": float(record.cost) if record.cost else 0,
})
return SuccessResponse(
data={
"total": total,
"records": data,
}
)
@router.get("/profile", response_model=SuccessResponse)
async def get_user_profile(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取当前用户信息
"""
user_id = principal.get("user_id")
# 查询用户
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="用户不存在"
)
# 返回用户信息
user_data = {
"id": str(user.id),
"username": user.username or (user.email.split("@")[0] if user.email else ""),
"name": user.name or user.full_name or "",
"full_name": user.full_name or user.name or "",
"email": user.email,
"role": user.role,
"company": getattr(user, "company", None), # 如果模型有company字段则返回,否则返回None
"created_at": user.created_at.isoformat() if user.created_at else None,
"updated_at": user.updated_at.isoformat() if user.updated_at else None,
}
return SuccessResponse(data=user_data)
@router.put("/profile", response_model=SuccessResponse)
async def update_user_profile(
req: dict,
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
更新当前用户信息
请求体示例:
{
"username": "new_username",
"company": "公司名称" // 可选
}
"""
user_id = principal.get("user_id")
# 查询用户
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="用户不存在"
)
# 更新用户名
if "username" in req and req["username"]:
# 检查用户名是否已被其他用户使用
existing_user_result = await db.execute(
select(User).where(
and_(
User.username == req["username"],
User.id != user_id
)
)
)
existing_user = existing_user_result.scalar_one_or_none()
if existing_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="用户名已被使用"
)
user.username = req["username"]
# 更新公司信息(如果模型支持)
if "company" in req:
if hasattr(user, "company"):
user.company = req["company"] if req["company"] else None
# 如果没有company字段,可以存储在permissions或其他JSON字段中
# 这里暂时忽略,如果后端需要支持,可以添加company字段到User模型
user.updated_at = datetime.utcnow()
await db.commit()
await db.refresh(user)
return SuccessResponse(
message="用户信息更新成功",
data={
"id": str(user.id),
"username": user.username,
"company": getattr(user, "company", None),
}
)
# ============= 用户资源信息查询 =============
@router.get("/resources/info", response_model=SuccessResponse)
async def get_user_resources_info(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取用户的 LiteLLM 密钥信息
返回:
- LiteLLM 密钥列表(解密后的完整密钥)
注意:
- 返回的 apiKey 是完整的解密密钥,请妥善保管
- 可直接用于调用 AI 模型(OpenAI 兼容格式)
"""
from app.litellm_client import get_litellm_client
from config import settings
user_id = principal.get("user_id")
# 获取 LiteLLM 密钥
litellm_keys = []
try:
# 查询用户的模型密钥
result = await db.execute(
select(TenantModelKey).where(
and_(
TenantModelKey.tenant_id == user_id,
TenantModelKey.status == "active"
)
)
)
tenant_keys = result.scalars().all()
litellm_client = get_litellm_client()
for key in tenant_keys:
try:
# 解密密钥
decrypted_key = litellm_client.decrypt_key(key.litellm_key_hash)
litellm_keys.append({
"modelName": key.model_name,
"apiKey": decrypted_key,
"apiBase": settings.litellm_url,
"rpmLimit": key.rpm_limit,
"tpmLimit": key.tpm_limit,
"maxBudget": float(key.max_budget) if key.max_budget else None,
"budgetDuration": key.budget_duration,
"status": key.status,
"createdAt": key.created_at.isoformat() if key.created_at else None,
})
except Exception as e:
logger.warning(f"解密 LiteLLM 密钥失败: {e}")
litellm_keys.append({
"modelName": key.model_name,
"apiKey": None,
"apiBase": settings.litellm_url,
"error": "密钥解密失败",
"status": key.status,
})
except Exception as e:
logger.error(f"查询 LiteLLM 密钥失败: {e}")
return SuccessResponse(
data={
"litellmKeys": litellm_keys,
"litellmApiBase": settings.litellm_url,
"summary": {
"totalLitellmKeys": len(litellm_keys),
}
},
message="LiteLLM 密钥信息获取成功"
)
# ============= 用户 Agent 资源查询 =============
@router.get("/resources/agents", response_model=SuccessResponse)
async def get_user_agents_info(
principal: dict = Depends(require_auth),
db: AsyncSession = Depends(get_db)
):
"""
获取用户已部署的 Agent 列表
返回:
- 已部署的平台 Agent 列表(包含 IP 地址和访问信息)
- 已部署的自定义 Agent 列表(包含 IP 地址和访问信息)
注意:
- Agent 的 podIp、status 等信息是实时从 AKS 查询的
- Pod 重启后 IP 地址会改变,建议使用 accessUrl(Service DNS)访问
"""
user_id = principal.get("user_id")
platform_agents = []
custom_agents = []
# 查询用户的活跃 Agent(未停止的)
result = await db.execute(
select(AgentBillingRecord).where(
and_(
AgentBillingRecord.user_id == user_id,
AgentBillingRecord.end_time == None # 正在运行
)
)
)
billing_records = result.scalars().all()
# 尝试获取 Agent Manager 客户端
agent_manager_available = False
client = None
try:
from app.agent_manager_client import get_agent_manager_client, AgentManagerError
client = get_agent_manager_client()
agent_manager_available = True
except Exception as e:
logger.warning(f"Agent Manager 客户端不可用: {e}")
for record in billing_records:
agent_info = {
"name": record.agent_name,
"template": record.agent_type,
"templateName": record.template_name,
"status": "unknown",
"healthStatus": "unknown",
"podIp": None,
# ========== 访问信息(优先使用数据库存储的值) ==========
"externalIp": record.external_ip,
"domain": record.domain,
"domainUrl": record.domain_url,
"accessUrl": record.access_url,
"servicePort": record.service_port,
"namespace": record.namespace or "ai-agents",
# ======================================================
"hostIp": None,
"nodeName": None,
"cpu": record.cpu_used,
"memory": record.memory_used,
"replicas": record.replicas,
"startTime": record.start_time.isoformat() if record.start_time else None,
"runningSeconds": int((datetime.utcnow() - record.start_time).total_seconds()) if record.start_time else 0,
}
# 从 Agent Manager 获取详细状态(实时更新)
if agent_manager_available and client:
try:
agent_status = await client.get_agent_status(record.agent_name)
agent_info["status"] = agent_status.status
agent_info["healthStatus"] = agent_status.health_status
agent_info["podIp"] = agent_status.pod_ip
agent_info["hostIp"] = agent_status.host_ip
agent_info["nodeName"] = agent_status.node_name
# ========== 更新访问信息(如果 Agent Manager 返回了最新值) ==========
if agent_status.external_ip:
agent_info["externalIp"] = agent_status.external_ip
if agent_status.domain:
agent_info["domain"] = agent_status.domain
if agent_status.domain_url:
agent_info["domainUrl"] = agent_status.domain_url
if agent_status.access_url:
agent_info["accessUrl"] = agent_status.access_url
if agent_status.service_port:
agent_info["servicePort"] = agent_status.service_port
if agent_status.namespace:
agent_info["namespace"] = agent_status.namespace
# ==================================================================
# 端点信息(字符串列表,如 ["http://10.244.1.100:8080"])
if agent_status.endpoints:
agent_info["endpoints"] = agent_status.endpoints
except Exception as e:
logger.warning(f"获取 Agent {record.agent_name} 状态失败: {e}")
# 分类存储
if record.is_platform_agent:
platform_agents.append(agent_info)
else:
custom_agents.append(agent_info)
return SuccessResponse(
data={
"platformAgents": platform_agents,
"customAgents": custom_agents,
"summary": {
"totalPlatformAgents": len(platform_agents),
"totalCustomAgents": len(custom_agents),
}
},
message="Agent 列表获取成功"
)