专家列表 + 使用统计查询优化

版本: 1.0.0
更新日期: 2026-02-08
状态: ✅ 已完成

本文档记录专家列表查询与使用统计的优化方案和实施情况。

📚 目录

1. 优化场景
2. 优化方案
3. 实施详情
4. 性能提升

优化场景

问题描述

在专家列表 API 中，如果需要显示每个专家的使用统计信息（如调用次数、今日调用次数）或购买状态，可能会产生 N+1 查询问题：

# ❌ 潜在的 N+1 查询问题
experts = await get_experts()  # 1 次查询
for expert in experts:
    stats = await get_expert_stats(expert.id)  # N 次查询
    purchased = await check_purchase(user_id, expert.id)  # N 次查询

当前实现

文件:

• shared/src/api/expert_pool.py - 专家列表 API
• shared/src/api/admin/market.py - 市场专家管理 API

当前状态:

• ✅ 专家调用统计已使用批量查询（get_expert_call_stats）
• ✅ 市场专家购买检查已优化（批量获取已购买专家ID集合）

优化方案

方案 1: 批量获取专家调用统计

已实现: get_expert_call_stats() 函数

使用 ANY(:expert_ids) 和 GROUP BY 一次性获取所有专家的统计信息。

优化前（潜在问题）:

for expert in experts:
    stats = await get_expert_stats(expert.id)  # N 次查询

优化后:

expert_ids = [e.id for e in experts]
stats = await get_expert_call_stats(expert_ids)  # 1 次批量查询
for expert in experts:
    expert.stats = stats.get(expert.id, {})

方案 2: 批量获取用户已购买的专家

已实现: 在 get_all_experts() 函数中

使用 get_valid_expert_ids() 一次性获取用户所有已购买的专家ID集合，然后在循环中检查。

优化前（N+1 查询）:

for expert in market_experts:
    purchased = await purchase_manager.has_purchased(user_id, expert.id)  # N 次查询

优化后:

# 1 次查询获取所有已购买的专家ID集合
purchased_expert_ids = await purchase_manager.get_valid_expert_ids(user_id)

# 在循环中检查（内存操作，无查询）
for expert in market_experts:
    purchased = expert.id in purchased_expert_ids  # 内存查找

实施详情

1. ✅ 专家调用统计批量查询

文件: shared/src/api/admin/market.py:236-287

实现:

async def get_expert_call_stats(expert_ids: List[str]) -> dict:
    """获取专家调用统计数据（批量查询）"""
    if not expert_ids:
        return {}

    async with AsyncSessionLocal() as session:
        # 查询每个专家的总调用次数（1 次查询）
        call_count_sql = text("""
            SELECT 
                expert_id,
                COUNT(*) as call_count
            FROM behavior_history
            WHERE expert_id = ANY(:expert_ids)
            GROUP BY expert_id
        """)
        result = await session.execute(call_count_sql, {"expert_ids": expert_ids})
        call_counts = {row.expert_id: row.call_count for row in result.fetchall()}

        # 查询今日调用次数（1 次查询）
        today = date.today()
        today_calls_sql = text("""
            SELECT 
                expert_id,
                COUNT(*) as today_count
            FROM behavior_history
            WHERE expert_id = ANY(:expert_ids)
            AND DATE(created_at) = :today
            GROUP BY expert_id
        """)
        today_result = await session.execute(
            today_calls_sql, {"expert_ids": expert_ids, "today": today}
        )
        today_counts = {
            row.expert_id: row.today_count for row in today_result.fetchall()
        }

        # 合并结果
        stats = {}
        for expert_id in expert_ids:
            stats[expert_id] = {
                "call_count": call_counts.get(expert_id, 0),
                "today_calls": today_counts.get(expert_id, 0),
            }

        return stats

使用示例:

# 在 list_market_experts 中使用
expert_ids = [e["id"] for e in all_experts]
call_stats = await get_expert_call_stats(expert_ids)  # 1 次批量查询

# 填充调用统计数据
for expert in all_experts:
    stats = call_stats.get(expert["id"], {})
    expert["call_count"] = stats.get("call_count", 0)
    expert["chat_count"] = stats.get("call_count", 0)

2. ✅ 市场专家购买检查优化

文件: shared/src/api/expert_pool.py:212-235

优化前:

# ❌ N+1 查询
for expert in market_experts:
    purchase_manager = get_purchase_manager()
    purchased = await purchase_manager.has_purchased(effective_user_id, expert.id)

优化后:

# ✅ 批量获取已购买的专家ID集合（1 次查询）
from users.expert_purchase import get_purchase_manager

purchase_manager = get_purchase_manager()
purchased_expert_ids = await purchase_manager.get_valid_expert_ids(
    effective_user_id
)  # 1 次查询获取所有已购买的专家

# 处理市场专家（内存查找，无查询）
for expert in market_experts:
    # 从已购买的专家ID集合中检查（避免循环查询）
    purchased = expert.id in purchased_expert_ids
    
    info = ExpertInfo(
        # ... 其他字段 ...
        accessible=purchased,
        access_reason="可用" if purchased else "需要购买",
    )

性能提升

查询次数对比

查询分解:

• 获取专家列表: 1 次
• 批量获取调用统计: 2 次（总调用次数 + 今日调用次数）
• 批量获取购买状态: 1 次（已优化）

响应时间对比

• 优化前: 100ms + N × 10ms（假设每次查询 10ms）
• 优化后: 100ms + 50ms（批量查询）
• 提升: 随着专家数量增加，性能提升显著

数据库压力

• 优化前: N+1 次数据库查询，高并发时压力大
• 优化后: 3 次数据库查询，压力显著降低

API 响应格式

优化后的 API 响应包含使用统计信息：

{
  "success": true,
  "experts": [
    {
      "id": "expert_123",
      "name": "专家名称",
      "call_count": 142,
      "chat_count": 142,
      "today_calls": 5,
      "accessible": true,
      // ... 其他字段
    }
  ],
  "stats": {
    "total": 100,
    "published": 80,
    "pending": 10,
    "rejected": 5,
    "draft": 5,
    "today_chats": 150
  }
}

相关优化

1. ✅ 平台专家权限批量检查

文件: shared/src/api/expert_pool.py:184-190

状态: ✅ 已优化

• 使用 batch_check_expert_permissions() 批量检查权限
• 无 N+1 查询问题

2. ✅ 市场专家状态批量获取

文件: shared/src/api/admin/market.py:89-170

状态: ✅ 已优化

• 使用 get_all_market_statuses() 批量获取状态
• 无 N+1 查询问题

最佳实践

1. 批量查询原则

当需要为多个对象获取关联数据时，优先使用批量查询：

# ✅ 好的做法：批量查询
expert_ids = [expert.id for expert in experts]
stats = await batch_get_expert_stats(expert_ids)

# ❌ 不好的做法：循环查询
for expert in experts:
    stats = await get_expert_stats(expert.id)

2. 使用集合查找

对于简单的成员检查，使用集合而不是循环查询：

# ✅ 好的做法：集合查找
purchased_ids = await get_purchased_expert_ids(user_id)  # Set[str]
purchased = expert.id in purchased_ids  # O(1) 查找

# ❌ 不好的做法：循环查询
purchased = await has_purchased(user_id, expert.id)  # 每次查询

3. 合并统计查询

对于多个统计指标，尽量合并到一个查询中：

# ✅ 好的做法：合并查询
SELECT 
    expert_id,
    COUNT(*) as total_calls,
    COUNT(CASE WHEN DATE(created_at) = :today THEN 1 END) as today_calls
FROM behavior_history
WHERE expert_id = ANY(:expert_ids)
GROUP BY expert_id

# ❌ 不好的做法：分开查询
# 查询1: 总调用次数
# 查询2: 今日调用次数

相关资源

• N+1 查询优化指南
• 性能优化文档
• 用户列表 + 对话历史优化

文档版本: 1.0.0
最后更新: 2026-02-08