增加视频话题库检索

examples/create/presets.json

@@ -7,7 +7,8 @@
       "planning",
       "research",
       "browser",
-      "topic_search"
+      "topic_search_image",
+      "topic_search_video"
     ],
     "description": "默认 Agent，拥有全部工具权限"
   }

examples/create/skills/topic_search_image.md

@@ -0,0 +1,9 @@
+---
+name: topic_search_image
+description: 在数据库中检索热门图文帖选题
+---
+
+## 热门图文选题检索
+
+你可以通过热门选题检索工具 `topic_search_image` 获取数据库中已有的热门图文帖子的选题
+调研过程可能需要多次搜索，比如基于搜索结果中获得的启发或信息启动新的搜索，直到得到令人满意的答案。你可以使用 `goal` 工具管理搜索的过程，或者使用文档记录搜索的中间或最终结果。

examples/create/skills/topic_search.md → examples/create/skills/topic_search_video.md

@@ -1,9 +1,9 @@
 ---
-name: topic_search
-description: 在数据库中检索热门选题
+name: topic_search_video
+description: 在数据库中检索热门视频选题
 ---
 
-## 热门选题检索
+## 热门视频选题检索
 
-你可以通过热门选题检索工具 `topic_search` 获取数据库中已有的热门帖子的选题
+你可以通过热门选题检索工具 `topic_search_video` 获取数据库中已有的热门视频帖子的选题
 调研过程可能需要多次搜索，比如基于搜索结果中获得的启发或信息启动新的搜索，直到得到令人满意的答案。你可以使用 `goal` 工具管理搜索的过程，或者使用文档记录搜索的中间或最终结果。

examples/create/tool/__init__.py

@@ -2,6 +2,7 @@
 Create 示例的自定义工具
 """
 
-from examples.create.tool.topic_search import topic_search
+from examples.create.tool.topic_search_image import topic_search_image
+from examples.create.tool.topic_search_video import topic_search_video
 
-__all__ = ["topic_search"]
+__all__ = ["topic_search_image", "topic_search_video"]

examples/create/tool/topic_search.py → examples/create/tool/topic_search_image.py

@@ -49,17 +49,17 @@ def _pick_first(results: List[Dict[str, Any]]) -> Dict[str, Any]:
 
 
 @tool(
-    description="根据关键词在数据库中检索已有帖子的选题，用于创作参考。最多返回5条，取第一条输出。",
+    description="根据关键词在数据库中检索已有图文帖子的选题，用于创作参考。最多返回5条，取第一条输出。",
     display={
         "zh": {
-            "name": "爆款选题检索",
+            "name": "爆款图文选题检索",
             "params": {
                 "keywords": "关键词列表",
             },
         },
     },
 )
-async def topic_search(keywords: List[str]) -> ToolResult:
+async def topic_search_image(keywords: List[str]) -> ToolResult:
     """
     根据关键词检索数据库中已有帖子的选题，取第一条作为参考。
 

examples/create/tool/topic_search_video.py

@@ -0,0 +1,107 @@
+"""
+选题检索工具 - 根据关键词在数据库中匹配已有帖子的选题
+
+用于 Agent 执行时自主调取参考数据。
+"""
+
+import json
+import os
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from agent.tools import tool, ToolResult
+
+# 选题检索 API 配置
+TOPIC_SEARCH_BASE_URL = os.getenv("TOPIC_SEARCH_BASE_URL", "http://supply-content-deconstruction-api.piaoquantv.com")
+DEFAULT_TIMEOUT = 60.0
+
+
+async def _call_search_api(keywords: List[str]) -> Optional[List[Dict[str, Any]]]:
+    """调用选题检索 API，返回结果列表。"""
+    url = f"{TOPIC_SEARCH_BASE_URL.rstrip('/')}/api/v1/content/topics/search"
+    payload = {"keywords": keywords}
+
+    try:
+        async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
+            resp = await client.post(url, json=payload)
+            resp.raise_for_status()
+            data = resp.json()
+    except httpx.HTTPStatusError as e:
+        raise RuntimeError(f"API 请求失败: {e.response.status_code} - {e.response.text[:200]}")
+    except Exception as e:
+        raise RuntimeError(f"请求异常: {str(e)}")
+
+    # 兼容多种响应格式
+    if isinstance(data, list):
+        return data[:5]
+    if isinstance(data, dict):
+        items = data.get("data") or data.get("results") or data.get("items") or []
+        return list(items)[:5] if isinstance(items, (list, tuple)) else []
+    return []
+
+
+def _pick_first(results: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """从结果中取第一条。"""
+    if not results:
+        raise ValueError("无可用结果")
+    return results[0]
+
+
+@tool(
+    description="根据关键词在数据库中检索已有爆款视频的选题，用于创作参考。最多返回5条，取第一条输出。",
+    display={
+        "zh": {
+            "name": "爆款视频选题检索",
+            "params": {
+                "keywords": "关键词列表",
+            },
+        },
+    },
+)
+async def topic_search_video(keywords: List[str]) -> ToolResult:
+    """
+    根据关键词检索数据库中已有帖子的选题，取第一条作为参考。
+
+    Args:
+        keywords: 关键词列表，如 ["中老年健康养生", "爆款", "知识科普"]
+
+    Returns:
+        ToolResult: 选题参考内容
+    """
+    if not keywords:
+        return ToolResult(
+            title="选题检索失败",
+            output="",
+            error="请提供至少一个关键词",
+        )
+
+    try:
+        results = await _call_search_api(keywords)
+    except RuntimeError as e:
+        return ToolResult(
+            title="选题检索失败",
+            output="",
+            error=str(e),
+        )
+
+    if not results:
+        return ToolResult(
+            title="选题检索",
+            output=json.dumps({"message": "未找到匹配的选题", "keywords": keywords}, ensure_ascii=False, indent=2),
+        )
+
+    try:
+        best = _pick_first(results)
+    except ValueError:
+        return ToolResult(
+            title="选题检索",
+            output=json.dumps({"message": "无可用结果", "keywords": keywords}, ensure_ascii=False, indent=2),
+        )
+
+    output = json.dumps(best, ensure_ascii=False, indent=2)
+    return ToolResult(
+        title="选题检索 - 参考数据",
+        output=output,
+        long_term_memory=f"检索到选题参考，关键词: {', '.join(keywords)}",
+    )