diff --git a/apps/api/app/api/routes/ai.py b/apps/api/app/api/routes/ai.py index f315d3126..6d0340147 100755 --- a/apps/api/app/api/routes/ai.py +++ b/apps/api/app/api/routes/ai.py @@ -7,7 +7,7 @@ from __future__ import annotations from typing import List, Literal, Optional -from app.services.ai_service import TITLE_STYLES, generate_smart_titles +from app.services.ai_service import TITLE_STYLES, generate_smart_titles, semantic_match_assets from fastapi import APIRouter from pydantic import BaseModel, Field @@ -45,6 +45,42 @@ class TitleStyleInfo(BaseModel): description: str +# ── 素材语义匹配 ──────────────────────────────────────────────────────────── + + +class AssetMatchItem(BaseModel): + """待匹配素材项.""" + + id: str = Field(..., description="素材ID") + name: str = Field(default="", description="素材名称") + tags: List[str] = Field(default_factory=list, description="标签列表") + description: str = Field(default="", description="素材描述") + + +class SemanticMatchRequest(BaseModel): + """语义匹配请求.""" + + description: str = Field(..., min_length=1, max_length=500, description="目标视频内容描述") + assets: List[AssetMatchItem] = Field(..., min_length=1, max_length=100, description="待匹配素材列表") + top_k: int = Field(default=0, ge=0, le=100, description="返回前K个，0返回全部") + + +class SemanticMatchResultItem(AssetMatchItem): + """匹配结果项.""" + + match_score: float = Field(..., description="匹配度评分 0-1") + match_reason: str = Field(..., description="匹配方式：doubao_semantic / fallback_keyword / fallback_default") + + +class SemanticMatchResponse(BaseModel): + """语义匹配响应.""" + + matches: List[SemanticMatchResultItem] = Field(..., description="按匹配度降序排列的素材列表") + source: str = Field(..., description="来源：doubao / fallback") + description: str = Field(..., description="原始描述") + total: int = Field(..., description="输入素材总数") + + # ── 路由 ──────────────────────────────────────────────────────────────────── @@ -70,3 +106,31 @@ def list_title_styles(): TitleStyleInfo(key=key, name=info["name"], description=info["description"]) for key, info in TITLE_STYLES.items() ] + + +@router.post("/assets/match", response_model=SemanticMatchResponse) +def match_assets(request: SemanticMatchRequest): + """智能素材语义匹配. + + 根据用户描述，对素材列表做语义匹配并按匹配度排序。 + 未配置豆包 API Key 时自动降级为关键词匹配。 + + - 支持最多 100 个素材同时匹配 + - 返回 match_score (0-1)，按降序排列 + - top_k 可限制返回数量 + """ + # 转为 dict 传给服务层 + assets_dict = [asset.model_dump() for asset in request.assets] + + result = semantic_match_assets( + description=request.description, + assets=assets_dict, + top_k=request.top_k, + ) + + return SemanticMatchResponse( + matches=[SemanticMatchResultItem(**m) for m in result["matches"]], + source=result["source"], + description=result["description"], + total=result["total"], + ) diff --git a/apps/api/app/services/ai_service.py b/apps/api/app/services/ai_service.py index 4f82b972b..4af575745 100755 --- a/apps/api/app/services/ai_service.py +++ b/apps/api/app/services/ai_service.py @@ -14,6 +14,7 @@ from __future__ import annotations import json import logging +import math import random import time from typing import Any, Dict, List, Optional @@ -314,6 +315,266 @@ def generate_smart_titles( } +# ── 智能素材语义匹配 ─────────────────────────────────────────────────────────── + + +def _semantic_match_fallback( + description: str, + assets: List[Dict[str, Any]], +) -> List[Dict[str, Any]]: + """本地降级：基于关键词的简单匹配. + + 计算描述中的关键词与素材名称/标签/描述的重叠度， + 作为匹配度评分。0-1分。 + """ + import re + + # 提取关键词（中文按2字以上片段，英文按单词） + desc = description.lower() + # 简单分词：提取2字以上的中文字符串和英文单词 + keywords = set() + # 英文单词 + for word in re.findall(r"[a-zA-Z]{3,}", desc): + keywords.add(word) + # 中文2-4字片段 + for i in range(len(desc)): + for j in range(i + 2, min(i + 5, len(desc) + 1)): + fragment = desc[i:j] + if all("\u4e00" <= c <= "\u9fff" for c in fragment): + keywords.add(fragment) + + if not keywords: + # 没有关键词时给所有素材中等分数 + for asset in assets: + asset["match_score"] = 0.5 + asset["match_reason"] = "fallback_default" + return assets + + results = [] + for asset in assets: + # 组合素材的文本信息：名称 + 标签 + 描述 + asset_text_parts = [ + str(asset.get("name", "")).lower(), + " ".join(str(t) for t in asset.get("tags", [])).lower(), + str(asset.get("description", "")).lower(), + ] + asset_text = " | ".join(asset_text_parts) + + # 计算匹配度：命中关键词占比 + 稀有关键词加权 + hit_count = 0 + hit_keywords = [] + for kw in keywords: + if kw in asset_text: + hit_count += 1 + hit_keywords.append(kw) + + # 基础匹配度 = 命中关键词数 / 总关键词数（开根号平滑） + base_score = math.sqrt(hit_count / len(keywords)) if keywords else 0.5 + + # 名称命中加分（名称匹配更重要） + name = str(asset.get("name", "")).lower() + name_hits = sum(1 for kw in hit_keywords if kw in name) + name_bonus = min(0.2, name_hits * 0.05) + + score = min(1.0, base_score * 0.8 + name_bonus) + score = round(score, 3) + + results.append({ + **asset, + "match_score": score, + "match_reason": "fallback_keyword", + }) + + # 按匹配度降序 + results.sort(key=lambda x: x["match_score"], reverse=True) + return results + + +def _parse_semantic_match_response( + content: str, + asset_ids: List[str], +) -> Optional[Dict[str, float]]: + """从模型返回中解析素材匹配度. + + 期望格式：JSON 对象 {asset_id: score} 或 {"matches": [{asset_id, score}]} + score 范围 0-1。 + """ + if not content: + return None + + # 尝试解析 JSON + try: + cleaned = content.strip() + if cleaned.startswith("```"): + cleaned = cleaned.strip("`") + if cleaned.lower().startswith("json"): + cleaned = cleaned[4:] + cleaned = cleaned.strip() + + data = json.loads(cleaned) + + result: Dict[str, float] = {} + + # 格式1: {"asset_id1": 0.8, "asset_id2": 0.6} + if isinstance(data, dict): + if "matches" in data and isinstance(data["matches"], list): + # 格式2: {"matches": [{"asset_id": "...", "score": 0.8}]} + for item in data["matches"]: + if isinstance(item, dict): + aid = item.get("asset_id") or item.get("id") + score = item.get("score", 0) + if aid and isinstance(score, (int, float)): + result[str(aid)] = max(0.0, min(1.0, float(score))) + else: + for key, value in data.items(): + if isinstance(value, (int, float)): + result[str(key)] = max(0.0, min(1.0, float(value))) + + # 格式3: [{"asset_id": "...", "score": 0.8}] + elif isinstance(data, list): + for item in data: + if isinstance(item, dict): + aid = item.get("asset_id") or item.get("id") + score = item.get("score", 0) + if aid and isinstance(score, (int, float)): + result[str(aid)] = max(0.0, min(1.0, float(score))) + + if len(result) >= max(1, len(asset_ids) // 2): # 至少一半素材有评分才算成功 + return result + + except (json.JSONDecodeError, ValueError): + pass + + return None + + +def semantic_match_assets( + description: str, + assets: List[Dict[str, Any]], + top_k: int = 0, +) -> Dict[str, Any]: + """智能素材语义匹配. + + 根据用户描述，评估每个素材的语义匹配度并排序。 + + Args: + description: 用户描述的目标视频内容 + assets: 素材列表，每个素材需含 id/name/tags/description 等字段 + top_k: 返回前K个，0表示返回全部 + + Returns: + { + "matches": [{"asset_id": ..., "match_score": ..., ...}], + "source": "doubao" | "fallback", + "description": "...", + "total": 总数, + } + """ + description = (description or "").strip() + if not assets: + return {"matches": [], "source": "fallback", "description": description, "total": 0} + + client = DoubaoAIClient() + if not client.is_available: + logger.info("豆包API未配置，使用本地降级做素材语义匹配") + matched = _semantic_match_fallback(description, assets) + if top_k > 0: + matched = matched[:top_k] + return { + "matches": matched, + "source": "fallback", + "description": description, + "total": len(assets), + } + + # 构建素材信息（控制 token 数量） + asset_summaries = [] + for asset in assets[:50]: # 最多传50个素材给模型 + aid = asset.get("id", "") + name = asset.get("name", "")[:50] + tags = asset.get("tags", []) + tags_str = ",".join(str(t) for t in tags[:5]) + desc = str(asset.get("description", ""))[:80] + asset_summaries.append( + f"ID:{aid} | 名称:{name} | 标签:[{tags_str}] | 描述:{desc}" + ) + + asset_ids = [str(a.get("id", "")) for a in assets[:50]] + + system_prompt = ( + "你是一个专业的视频素材匹配助手。" + "根据用户的视频目标描述，评估每个素材的匹配程度。\n" + "评分规则：\n" + "- 0.0-0.3: 完全不相关\n" + "- 0.3-0.6: 有一定关联但不够匹配\n" + "- 0.6-0.8: 比较匹配，适合使用\n" + "- 0.8-1.0: 高度匹配，非常适合\n" + "只返回JSON对象，key为素材ID，value为匹配分数（0-1之间的小数）。" + "不要其他文字说明。" + ) + + user_prompt = ( + f"目标视频描述：{description}\n\n" + f"素材列表：\n" + "\n".join(asset_summaries) + + f"\n\n请返回每个素材的匹配分数JSON：" + ) + + messages = [ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt}, + ] + + result = client._chat_completion( + messages=messages, + temperature=0.3, + max_tokens=1024, + ) + + if result: + scores = _parse_semantic_match_response(result, asset_ids) + if scores: + # 把评分填回素材 + matched = [] + for asset in assets: + aid = str(asset.get("id", "")) + score = scores.get(aid, 0.3) # 没评分的给默认偏低分 + matched.append({ + **asset, + "match_score": round(score, 3), + "match_reason": "doubao_semantic", + }) + matched.sort(key=lambda x: x["match_score"], reverse=True) + + logger.info( + "豆包语义匹配完成: assets=%d top_score=%.2f description=%s...", + len(matched), + matched[0]["match_score"] if matched else 0, + description[:20], + ) + + if top_k > 0: + matched = matched[:top_k] + + return { + "matches": matched, + "source": "doubao", + "description": description, + "total": len(assets), + } + logger.warning("豆包语义匹配返回解析失败，降级到本地: %s", result[:100]) + + # 降级 + matched = _semantic_match_fallback(description, assets) + if top_k > 0: + matched = matched[:top_k] + return { + "matches": matched, + "source": "fallback", + "description": description, + "total": len(assets), + } + + # ── 单例入口 ───────────────────────────────────────────────────────────────── @@ -345,3 +606,11 @@ class AIService: count: int = 5, ) -> Dict[str, Any]: return generate_smart_titles(description, style, count) + + def semantic_match( + self, + description: str, + assets: List[Dict[str, Any]], + top_k: int = 0, + ) -> Dict[str, Any]: + return semantic_match_assets(description, assets, top_k) diff --git a/tests/unit/test_ai_service.py b/tests/unit/test_ai_service.py index 738737659..5050aedf2 100755 --- a/tests/unit/test_ai_service.py +++ b/tests/unit/test_ai_service.py @@ -22,7 +22,10 @@ from app.services.ai_service import ( # noqa: E402 TITLE_STYLES, _generate_titles_fallback, _parse_titles_from_response, + _parse_semantic_match_response, + _semantic_match_fallback, generate_smart_titles, + semantic_match_assets, ) @@ -295,5 +298,230 @@ class TestTitleStyles(unittest.TestCase): self.assertIn("informative", TITLE_STYLES) +# ── 语义匹配测试 ────────────────────────────────────────────────────────────── + + +class TestSemanticMatchFallback(unittest.TestCase): + """降级关键词匹配测试.""" + + def _make_assets(self): + return [ + {"id": "a1", "name": "海边日落风景", "tags": ["风景", "海边", "日落"], "description": "美丽的海边日落"}, + {"id": "a2", "name": "城市夜景航拍", "tags": ["城市", "夜景", "航拍"], "description": "城市夜景航拍素材"}, + {"id": "a3", "name": "美食制作过程", "tags": ["美食", "烹饪", "教程"], "description": "美食制作教程"}, + ] + + def test_fallback_returns_sorted_scores(self): + """返回按匹配度降序排列.""" + assets = self._make_assets() + result = _semantic_match_fallback("海边日落风景视频", assets) + self.assertEqual(len(result), 3) + # 第一个应该是海边日落 + self.assertEqual(result[0]["id"], "a1") + self.assertGreater(result[0]["match_score"], result[2]["match_score"]) + + def test_fallback_each_has_match_score(self): + """每个素材都有 match_score.""" + assets = self._make_assets() + result = _semantic_match_fallback("测试", assets) + for item in result: + self.assertIn("match_score", item) + self.assertGreaterEqual(item["match_score"], 0.0) + self.assertLessEqual(item["match_score"], 1.0) + self.assertIn("match_reason", item) + + def test_fallback_unrelated_desc_low_scores(self): + """完全不相关的描述得分低.""" + assets = self._make_assets() + result = _semantic_match_fallback("篮球比赛运动", assets) + # 所有素材得分都应该较低 + for item in result: + self.assertLess(item["match_score"], 0.8) + + def test_fallback_empty_keywords_default_score(self): + """无有效关键词时给默认分.""" + assets = self._make_assets() + result = _semantic_match_fallback("a", assets) # 单字符无有效关键词 + for item in result: + self.assertEqual(item["match_score"], 0.5) + self.assertEqual(item["match_reason"], "fallback_default") + + def test_fallback_name_match_higher(self): + """名称命中得分更高.""" + assets = [ + {"id": "a1", "name": "美食探店vlog", "tags": [], "description": ""}, + {"id": "a2", "name": "风景视频", "tags": ["美食"], "description": ""}, + ] + result = _semantic_match_fallback("美食", assets) + # a1名称含美食，a2标签含美食，名称命中应有额外加分 + self.assertEqual(result[0]["id"], "a1") + self.assertGreater(result[0]["match_score"], result[1]["match_score"]) + + +class TestSemanticMatchParsing(unittest.TestCase): + """语义匹配返回解析测试.""" + + def test_parse_dict_format(self): + """解析 {id: score} 格式.""" + content = json.dumps({"asset1": 0.85, "asset2": 0.62, "asset3": 0.3}) + result = _parse_semantic_match_response(content, ["asset1", "asset2", "asset3"]) + self.assertIsNotNone(result) + self.assertEqual(len(result), 3) + self.assertAlmostEqual(result["asset1"], 0.85) + + def test_parse_matches_list_format(self): + """解析 {matches: [...]} 格式.""" + content = json.dumps({ + "matches": [ + {"asset_id": "a1", "score": 0.9}, + {"asset_id": "a2", "score": 0.7}, + ] + }) + result = _parse_semantic_match_response(content, ["a1", "a2"]) + self.assertIsNotNone(result) + self.assertAlmostEqual(result["a1"], 0.9) + self.assertAlmostEqual(result["a2"], 0.7) + + def test_parse_array_format(self): + """解析数组格式.""" + content = json.dumps([ + {"id": "x1", "score": 0.5}, + {"id": "x2", "score": 0.88}, + ]) + result = _parse_semantic_match_response(content, ["x1", "x2"]) + self.assertIsNotNone(result) + self.assertAlmostEqual(result["x1"], 0.5) + + def test_parse_score_clamped(self): + """分数被限制在0-1.""" + content = json.dumps({"a1": 1.5, "a2": -0.2}) + result = _parse_semantic_match_response(content, ["a1", "a2"]) + self.assertIsNotNone(result) + self.assertAlmostEqual(result["a1"], 1.0) + self.assertAlmostEqual(result["a2"], 0.0) + + def test_parse_markdown_code_block(self): + """解析markdown代码块.""" + content = "```json\n{\"a1\": 0.7}\n```" + result = _parse_semantic_match_response(content, ["a1", "a2"]) + # 只有1个素材评分，少于一半（需要至少1个，max(1, 2//2)=1） + self.assertIsNotNone(result) + self.assertAlmostEqual(result["a1"], 0.7) + + def test_parse_empty_returns_none(self): + """空内容返回None.""" + result = _parse_semantic_match_response("", ["a1"]) + self.assertIsNone(result) + + def test_parse_invalid_json_returns_none(self): + """无效JSON返回None.""" + result = _parse_semantic_match_response("不是json", ["a1", "a2", "a3"]) + self.assertIsNone(result) + + +class TestSemanticMatchAssets(unittest.TestCase): + """semantic_match_assets 集成测试.""" + + def _make_assets(self): + return [ + {"id": "a1", "name": "海边日落", "tags": ["风景"], "description": ""}, + {"id": "a2", "name": "城市夜景", "tags": ["城市"], "description": ""}, + {"id": "a3", "name": "美食制作", "tags": ["美食"], "description": ""}, + ] + + def test_fallback_mode_without_api_key(self): + """无API Key时走降级.""" + with patch("app.services.ai_service.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + DOUBAO_API_KEY="", + DOUBAO_MODEL="test", + DOUBAO_BASE_URL="https://test.com", + DOUBAO_TIMEOUT=30, + DOUBAO_MAX_RETRIES=0, + ) + result = semantic_match_assets("海边", self._make_assets()) + self.assertEqual(result["source"], "fallback") + self.assertEqual(result["total"], 3) + self.assertEqual(len(result["matches"]), 3) + + def test_empty_assets(self): + """空素材列表.""" + result = semantic_match_assets("test", []) + self.assertEqual(result["total"], 0) + self.assertEqual(len(result["matches"]), 0) + + def test_top_k_limit(self): + """top_k 限制返回数量.""" + with patch("app.services.ai_service.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + DOUBAO_API_KEY="", + DOUBAO_MODEL="test", + DOUBAO_BASE_URL="https://test.com", + DOUBAO_TIMEOUT=30, + DOUBAO_MAX_RETRIES=0, + ) + result = semantic_match_assets("测试", self._make_assets(), top_k=2) + self.assertEqual(len(result["matches"]), 2) + + def test_with_doubao_success(self): + """豆包调用成功路径.""" + with patch("app.services.ai_service.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + DOUBAO_API_KEY="sk-test-123", + DOUBAO_MODEL="test-model", + DOUBAO_BASE_URL="https://test.com", + DOUBAO_TIMEOUT=30, + DOUBAO_MAX_RETRIES=0, + ) + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "choices": [{ + "message": { + "content": json.dumps({"a1": 0.9, "a2": 0.5, "a3": 0.2}) + } + }] + } + mock_response.raise_for_status = MagicMock() + + with patch("httpx.post", return_value=mock_response): + result = semantic_match_assets("风景视频", self._make_assets()) + self.assertEqual(result["source"], "doubao") + self.assertEqual(len(result["matches"]), 3) + # 按分数降序，a1最高 + self.assertEqual(result["matches"][0]["id"], "a1") + self.assertAlmostEqual(result["matches"][0]["match_score"], 0.9) + + def test_with_doubao_failure_fallback(self): + """豆包调用失败降级.""" + with patch("app.services.ai_service.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + DOUBAO_API_KEY="sk-test-123", + DOUBAO_MODEL="test-model", + DOUBAO_BASE_URL="https://test.com", + DOUBAO_TIMEOUT=1, + DOUBAO_MAX_RETRIES=0, + ) + with patch("httpx.post", side_effect=Exception("API Error")): + result = semantic_match_assets("测试", self._make_assets()) + self.assertEqual(result["source"], "fallback") + + def test_each_match_has_required_fields(self): + """每个匹配结果都有必要字段.""" + with patch("app.services.ai_service.get_settings") as mock_settings: + mock_settings.return_value = MagicMock( + DOUBAO_API_KEY="", + DOUBAO_MODEL="test", + DOUBAO_BASE_URL="https://test.com", + DOUBAO_TIMEOUT=30, + DOUBAO_MAX_RETRIES=0, + ) + result = semantic_match_assets("测试", self._make_assets()) + for item in result["matches"]: + self.assertIn("id", item) + self.assertIn("match_score", item) + self.assertIn("match_reason", item) + + if __name__ == "__main__": unittest.main()