diff --git a/PR_DESCRIPTION.md b/PR_DESCRIPTION.md deleted file mode 100644 index ad3d16d..0000000 --- a/PR_DESCRIPTION.md +++ /dev/null @@ -1,106 +0,0 @@ -# 功能实现总结：智能 Token 路由（视频生成） - -## 🎯 功能概述 - -实现了视频生成的智能 Token 路由功能，根据视频需求自动选择最合适的 Token 池（ssoSuper vs ssoBasic）。 - -## 🔧 实现内容 - -### 1. Token 管理器增强 (`app/services/token/manager.py`) - -新增两个方法： - -#### `get_token_info(pool_name)` -- 返回完整的 TokenInfo 对象（而非仅 token 字符串） -- 便于后续获取 Token 元数据 - -#### `get_token_for_video(resolution, video_length)` ⭐ 核心功能 -智能路由逻辑： -- **720p 或 >6s 视频** → 优先使用 `ssoSuper` Token -- **480p 且 ≤6s 视频** → 使用 `ssoBasic` Token -- **Super Token 不可用** → 自动回退到 Basic Token（带警告日志） - -### 2. 视频服务修改 (`app/services/grok/services/media.py`) - -- 使用新的 `get_token_for_video()` 方法替代原有的循环查找 -- 自动根据视频配置选择最佳 Token -- 保持向后兼容 - -### 3. 配置选项 (`config.defaults.toml`) - -新增 `[video]` 配置段： -```toml -[video] -# 是否启用智能 Token 路由 -smart_token_routing = true - -# 当需要 Super Token 但不可用时是否允许回退 -allow_fallback_to_basic = true - -# 回退时的警告日志 -log_fallback_warning = true -``` - -## 📊 修改统计 - -- **新增代码**: ~90 行 -- **修改文件**: 3 个 -- **测试文件**: 1 个 (`test_video_token_routing.py`) - -## ✅ 解决的问题 - -1. **非 SSO Super Token 被静默降级**: 现在系统会主动选择正确的 Token 类型 -2. **720p 视频生成失败**: 自动使用 Super Token 避免降级 -3. **Token 选择不透明**: 详细的日志记录便于调试 - -## 🧪 测试场景 - -测试覆盖以下场景： -1. ✅ 480p, 6s → 使用 Basic Token -2. ✅ 720p, 6s → 使用 Super Token -3. ✅ 480p, 10s → 使用 Super Token（>6s） -4. ✅ 720p, Super 池空 → 回退到 Basic Token - -## 📝 使用说明 - -无需修改现有代码，系统会自动根据视频请求参数选择 Token： - -```python -# 请求 720p 视频 -{ - "model": "grok-video", - "video_config": { - "resolution_name": "720p", - "video_length": 6 - } -} -# → 自动使用 ssoSuper Token - -# 请求 480p 视频 -{ - "model": "grok-video", - "video_config": { - "resolution_name": "480p", - "video_length": 6 - } -} -# → 使用 ssoBasic Token -``` - -## 🔍 日志示例 - -``` -# 正常路由 -Video token routing: resolution=720p, length=6s -> pool=ssoSuper (token=eyJ0eXAiOi...) - -# 回退情况 -Video token routing: ssoSuper pool has no available token for resolution=720p, length=6s. -Falling back to ssoBasic pool. -Video token routing: fallback from ssoSuper -> ssoBasic (token=eyJ0eXAiOi...) -``` - -## 🎉 预期效果 - -- 720p 视频请求将正确使用 SSO Super Token -- 避免因 Token 类型不匹配导致的静默降级 -- 当 Super Token 不足时自动回退，保证服务可用性 diff --git a/app/services/grok/services/media.py b/app/services/grok/services/media.py index a9a2961..79d9b67 100644 --- a/app/services/grok/services/media.py +++ b/app/services/grok/services/media.py @@ -17,7 +17,7 @@ from app.core.exceptions import ( ErrorType, ) from app.services.grok.models.model import ModelService -from app.services.token import get_token_manager, EffortType, TokenInfo +from app.services.token import get_token_manager, EffortType from app.services.grok.processors import VideoStreamProcessor, VideoCollectProcessor from app.services.grok.utils.headers import apply_statsig, build_sso_cookie from app.services.grok.utils.stream import wrap_stream_with_usage @@ -305,9 +305,12 @@ class VideoService: token_mgr = await get_token_manager() await token_mgr.reload_if_stale() - # 使用智能路由选择 token（根据视频需求） + # 使用智能路由选择 token（根据视频需求与候选池） + pool_candidates = ModelService.pool_candidates_for_model(model) token_info = token_mgr.get_token_for_video( - resolution=resolution, video_length=video_length + resolution=resolution, + video_length=video_length, + pool_candidates=pool_candidates, ) if not token_info: diff --git a/app/services/token/manager.py b/app/services/token/manager.py index 8f30afc..b5e28f1 100644 --- a/app/services/token/manager.py +++ b/app/services/token/manager.py @@ -233,19 +233,23 @@ class TokenManager: return token_info def get_token_for_video( - self, resolution: str = "480p", video_length: int = 6 + self, + resolution: str = "480p", + video_length: int = 6, + pool_candidates: Optional[List[str]] = None, ) -> Optional["TokenInfo"]: """ 根据视频需求智能选择 Token 池 路由策略: - 如果 resolution 是 "720p" 或 video_length > 6: 优先使用 "ssoSuper" 池 - - 如果 "ssoSuper" 池没有可用 token: 记录警告并回退到 "ssoBasic" 池 - - 如果 resolution 是 "480p" 且 video_length <= 6: 使用 "ssoBasic" 池 + - 否则优先使用 "ssoBasic" 池 + - 当提供 pool_candidates 时，按候选池顺序回退 Args: resolution: 视频分辨率 ("480p" 或 "720p") video_length: 视频时长(秒) + pool_candidates: 候选 Token 池（按优先级） Returns: TokenInfo 对象或 None（无可用 token） @@ -253,32 +257,39 @@ class TokenManager: # 确定首选池 requires_super = resolution == "720p" or video_length > 6 primary_pool = SUPER_POOL_NAME if requires_super else BASIC_POOL_NAME - fallback_pool = BASIC_POOL_NAME if requires_super else SUPER_POOL_NAME - # 优先从主池获取 token - primary_token = self.get_token_info(primary_pool) - if primary_token: - logger.info( - f"Video token routing: resolution={resolution}, length={video_length}s -> " - f"pool={primary_pool} (token={primary_token.token[:10]}...)" - ) - return primary_token + if pool_candidates: + ordered_pools = list(pool_candidates) + if primary_pool in ordered_pools: + ordered_pools.remove(primary_pool) + ordered_pools.insert(0, primary_pool) + else: + fallback_pool = BASIC_POOL_NAME if requires_super else SUPER_POOL_NAME + ordered_pools = [primary_pool, fallback_pool] - # 主池无可用 token，尝试回退池 - if requires_super: - logger.warning( - f"Video token routing: ssoSuper pool has no available token for " - f"resolution={resolution}, length={video_length}s. " - f"Falling back to ssoBasic pool." - ) + for idx, pool_name in enumerate(ordered_pools): + token_info = self.get_token_info(pool_name) + if token_info: + if idx == 0: + logger.info( + f"Video token routing: resolution={resolution}, length={video_length}s -> " + f"pool={pool_name} (token={token_info.token[:10]}...)" + ) + else: + logger.info( + f"Video token routing: fallback from {ordered_pools[0]} -> {pool_name} " + f"(token={token_info.token[:10]}...)" + ) + return token_info - fallback_token = self.get_token_info(fallback_pool) - if fallback_token: - logger.info( - f"Video token routing: fallback from {primary_pool} -> {fallback_pool} " - f"(token={fallback_token.token[:10]}...)" - ) - return fallback_token + if idx == 0 and requires_super and pool_name == primary_pool: + next_pool = ordered_pools[1] if len(ordered_pools) > 1 else None + if next_pool: + logger.warning( + f"Video token routing: {primary_pool} pool has no available token for " + f"resolution={resolution}, length={video_length}s. " + f"Falling back to {next_pool} pool." + ) # 两个池都没有可用 token logger.warning( diff --git a/config.defaults.toml b/config.defaults.toml index 8d92937..8694424 100644 --- a/config.defaults.toml +++ b/config.defaults.toml @@ -138,16 +138,3 @@ nsfw_batch_size = 50 # NSFW 单次最大数量 nsfw_max_tokens = 1000 - -# ==================== 视频 Token 路由 ==================== -[video] -# 是否启用智能 Token 路由（根据视频需求自动选择 Token 池） -# true: 720p 或 >6s 视频自动使用 ssoSuper Token -# false: 所有视频使用默认 Token 池 -smart_token_routing = true - -# 当需要 Super Token 但不可用时是否允许回退到 Basic Token -allow_fallback_to_basic = true - -# 回退时的警告日志 -log_fallback_warning = true diff --git a/test_video_token_routing.py b/test_video_token_routing.py deleted file mode 100644 index 12f882c..0000000 --- a/test_video_token_routing.py +++ /dev/null @@ -1,75 +0,0 @@ -""" -测试视频 Token 智能路由功能 -""" -import sys -sys.path.insert(0, '.') - -from app.services.token.models import TokenInfo, TokenStatus, BASIC__DEFAULT_QUOTA, SUPER_DEFAULT_QUOTA -from app.services.token.pool import TokenPool -from app.services.token.manager import TokenManager - - -def test_token_routing(): - """测试 Token 路由逻辑""" - print("🧪 测试视频 Token 智能路由功能\n") - - # 创建模拟 Token 池 - basic_pool = TokenPool("ssoBasic") - super_pool = TokenPool("ssoSuper") - - # 添加测试 Token - basic_token = TokenInfo( - token="basic_test_token_123", - status=TokenStatus.ACTIVE, - quota=BASIC__DEFAULT_QUOTA - ) - super_token = TokenInfo( - token="super_test_token_456", - status=TokenStatus.ACTIVE, - quota=SUPER_DEFAULT_QUOTA - ) - - basic_pool.add(basic_token) - super_pool.add(super_token) - - # 创建 TokenManager 并设置池 - manager = TokenManager.__new__(TokenManager) - manager.pools = { - "ssoBasic": basic_pool, - "ssoSuper": super_pool - } - - # 测试场景 1: 480p, 6s - 应该使用 Basic - print("场景 1: 480p, 6s") - result = manager.get_token_for_video("480p", 6) - assert result is not None, "应该获取到 Token" - assert result.token == "basic_test_token_123", f"应该使用 Basic Token, 但得到: {result.token}" - print(" ✅ 正确使用 Basic Token\n") - - # 测试场景 2: 720p, 6s - 应该使用 Super - print("场景 2: 720p, 6s") - result = manager.get_token_for_video("720p", 6) - assert result is not None, "应该获取到 Token" - assert result.token == "super_test_token_456", f"应该使用 Super Token, 但得到: {result.token}" - print(" ✅ 正确使用 Super Token\n") - - # 测试场景 3: 480p, 10s (>6s) - 应该使用 Super - print("场景 3: 480p, 10s (>6s)") - result = manager.get_token_for_video("480p", 10) - assert result is not None, "应该获取到 Token" - assert result.token == "super_test_token_456", f"应该使用 Super Token, 但得到: {result.token}" - print(" ✅ 正确使用 Super Token\n") - - # 测试场景 4: 720p, 但 Super 池为空 - 应该回退到 Basic - print("场景 4: 720p, 但 Super 池为空（回退测试）") - super_pool.remove(super_token.token) # 清空 Super 池 - result = manager.get_token_for_video("720p", 6) - assert result is not None, "应该获取到 Token（回退）" - assert result.token == "basic_test_token_123", f"应该回退到 Basic Token, 但得到: {result.token}" - print(" ✅ 正确回退到 Basic Token\n") - - print("🎉 所有测试通过！") - - -if __name__ == "__main__": - test_token_routing()