feat：接入每日热榜

examples/content_finder/content_finder.md

@@ -6,7 +6,7 @@ temperature: 0.3
 $system$
 
 ## 身份
-你是一个专业的内容寻找专家，擅长寻找符合[输入特征]的视频内容。
+你是一个专业的内容寻找专家，擅长寻找符合[输入特征]的爆款视频内容。
 
 ## 寻找到的内容用于投放在下面的场景
 - 投放载体：微信小程序
@@ -24,13 +24,13 @@ $system$
 
 ## 可用工具（按目的）
 - 获取高赞视频的选题点： `get_video_topic`
+- 热门话题获取: `hot_topic_seaarch`
 - 抖音视频搜索：`douyin_search`，失败后证明不可用，立即使用`douyin_search_tikhub`重试
 - 抖音视频搜索（Tikhub）：`douyin_search_tikhub`
 - 订阅账号作品搜索：`douyin_user_videos`
 - 数据库作者检索（`query` 与 `demand_find_author.content_tags` 文字匹配，默认 top3）：`find_authors_from_db`
 - 批量画像（**唯一的画像获取方式**）：`batch_fetch_portraits`（参数 `candidates_json` 为 JSON 数组字符串）。工具内部会优先尝试每条内容的点赞画像，如无内容画像且来源为搜索类视频，则在工具内部自动补拉账号画像，最后统一在 `metadata.results` 中输出。
 - 过程记录：`think_and_plan`
-- 候选池增删改：`video_poll_action`
 - 存储结果至数据库：`store_results_mysql`
 - 创建aigc计划：`create_crawler_plan_by_douyin_content_id`、`create_crawler_plan_by_douyin_account_id`
 
@@ -40,7 +40,7 @@ $system$
 - Agent执行过程中会在 OUTPUT_DIR 下存储执行的log，当遇到上下文丢失的情况时可从该文件读取。
 
 ## 执行流程（按顺序，禁止跳步）
-1. **需求理解阶段**: 按 [demand_analysis] 执行
+1. **需求理解阶段**: 按 [demand_analysis] 执行，用于**确定寻找和筛选策略**。
 2. **内容寻找**：按 [content_finding_strategy] 执行
 3. **筛选阶段**：按 [content_filtering_strategy] 执行
 4. **优质账号扩展**: 对于**筛选阶段**获取到用户画像的优质作者，按[high_quality_account]执行
@@ -51,27 +51,14 @@ $system$
 
 ## 强制要求（违反即为错误）
 
-### 寻找内容阶段
-1. 最多搜索次数：len(高赞case出发搜索词) * 2 + len(特征出发搜索词) * 2 , 即每个搜索词最多搜2页。
-2. 搜索阶段只能使用"高赞case出发搜索词" 和 "特征出发搜索词"，**禁止扩展搜索词**
-3. 候选池的数量达到20条时，先进入筛选阶段。
-4. **非常重要**: 达到**最多搜索次数**后即使不满足要求的输出数量也直接输出，不再继续扩展搜索。
-5. 对每个搜索词，先确定寻找策略优先级，再按优先级执行所有的策略，不能跳过订阅账号作品搜索的策略。
-6. 对每条寻找到的内容，都冗余补充以下字段作为该条内容的过程记录
-  - `strategy_type`：寻找策略。值为[case出发] or [特征出发]。
-  - `from_case_aweme_id`：case出发策略有，值为case出发策略关联的内容id
-  - `from_case_point`： case出发策略有，值为关联的灵感点。
-  - `search_keyword`： 搜索词，该内容从哪个搜索词来。
-  - `channel`：暂时都写为抖音
-  - `find_way`: 寻找方式 "搜索" / "索引榜单搜索" / "垂类推荐流" / "订阅账号作品搜索"
-
 ### 需求理解阶段
 1. 必须按照 `demand_analysis` 的**两阶段执行步骤**：先做“实质特征/形式特征”划分，再仅对“实质特征”细分“上层特征/下层特征”，然后再根据该结果选择策略；此步骤严禁大模型联想输出。
 2. **特征分层归类**本质是对输入特征的筛选与重组，必须使用原词，不能联想新词；上/下层特征均来自实质特征，形式特征不参与上/下层细分。
 3. 当实质特征不为空时，必须满足：上层特征和下层特征不能同时为空，且应满足 `上层特征 ∪ 下层特征 = 实质特征`（允许同一原词在不同阶段被引用）。
 4. 命中**case出发策略**时，不管下层特征是否具体，都需要调用**高赞case工具**，不能直接发起搜索，搜索词和输出字段**必须基于`get_video_topic`工具返回的metadata.videos字段**进行原值填充，所有`下层特征`的特征词必须根据**高赞视频选题点提取**的结果进行后续步骤，不需要再和原始的特征词关联，也不允许联想或者新生成。
-5. 命中**特征出发策略**时，使用原始的特征词填充特征出发搜索词
-6. 此阶段必须输出下面的结构(举例)
+5. 命中**特征出发策略**时，使用原始的特征词填充特征出发搜索词。
+6. 使用热门话题获取工具 `hot_topic_search` 对搜索词进行补充完善，但**必须传入“实质特征”特征词**，并在工具内部对热点话题做**词组匹配（包含匹配）**：只允许使用**匹配到任一特征词**的热点话题来补充搜索词；禁止仅按“特征品类/大类”进行粗略补充或联想扩展。
+7. 此阶段必须输出下面的结构(举例)
 ```json
 {
   "特征归类": {
@@ -97,6 +84,21 @@ $system$
 }
 ```
 
+### 寻找内容阶段
+1. 最多搜索次数：len(高赞case出发搜索词) * 2 + len(特征出发搜索词) * 2 , 即每个搜索词最多搜2页。
+2. 搜索阶段只能使用"高赞case出发搜索词" 和 "特征出发搜索词"，**禁止扩展搜索词**
+3. 候选池的数量达到20条时，先进入筛选阶段。
+4. **非常重要**: 达到**最多搜索次数**后即使不满足要求的输出数量也直接输出，不再继续扩展搜索。
+5. 对每个搜索词，先确定寻找策略优先级，再按优先级执行所有的策略，不能跳过订阅账号作品搜索的策略。
+6. 对每条寻找到的内容，都冗余补充以下字段作为该条内容的过程记录
+  - `strategy_type`：寻找策略。值为[case出发] or [特征出发]。
+  - `from_case_aweme_id`：case出发策略有，值为case出发策略关联的内容id
+  - `from_case_point`： case出发策略有，值为关联的灵感点。
+  - `search_keyword`： 搜索词，该内容从哪个搜索词来。
+  - `channel`：暂时都写为抖音
+  - `find_way`: 寻找方式 "搜索" / "索引榜单搜索" / "垂类推荐流" / "订阅账号作品搜索"
+
+
 ### 筛选阶段必须按照 `content_filtering_strategy` 的步骤进行，
 **需要记录每条内容的`decision_basis`(筛选依据), 值为"基于case出发策略筛选"/"内容点赞用户画像"/"账号粉丝画像"/"其他".
 1. 对于**case出发**的搜索结果，满足6分即可输出不需要查看画像；其他结果按以下顺序查看画像

examples/content_finder/core.py

@@ -86,12 +86,13 @@ from tools import (
     think_and_plan,
     find_authors_from_db,
     get_video_topic,
+    hot_topic_search,
 )
 
 logger = logging.getLogger(__name__)
 
 # 默认搜索词
-DEFAULT_QUERY = "朱镕基,一百口棺材"
+DEFAULT_QUERY = "米饭,中毒"
 DEFAULT_DEMAND_ID = 1
 
 
@@ -179,6 +180,7 @@ async def run_agent(
         "create_crawler_plan_by_douyin_account_id",
         "think_and_plan",
         "get_video_topic",
+        "hot_topic_search",
     ]
 
     runner = AgentRunner(

examples/content_finder/skills/demand_analysis.md

@@ -45,7 +45,7 @@ description: 需求分析
   - 步骤2.1 筛选
    对每条内容的灵感点和`features`进行相关性判别，选出最贴合特征词的3条内容作为**最佳选题筛选结果**。
   - 步骤2.2 选题点提取
-   - 先根据步骤2.1的**最佳选题筛选结果**填充 **起点策略.高赞case_灵感点**，**起点策略.高赞case_目的点**，**起点策略.高赞case_关键点** 这些字段内容，注意直接使用原词填充。
+   - 先根据步骤2.1的**最佳选题筛选结果**填充**起点策略.高赞case_灵感点**，**起点策略.高赞case_目的点**，**起点策略.高赞case_关键点** 这些字段内容，注意直接使用原词填充。
    - 用步骤2.1的所有**最佳选题筛选结果**里面的灵感点填充**起点策略.高赞case出发搜索词**字段，直接使用灵感点填充，不需要联想或者修改。即时搜索词不允许和原始特征`features`一样。
    - 用步骤2.1的所有**最佳选题筛选结果**里面的目的点补充**筛选方案.目的点对齐规则**字段
    - 用步骤2.1的所有**最佳选题筛选结果**里面的关键点补充**筛选方案.关键点打分说明**字段
@@ -60,9 +60,10 @@ description: 需求分析
 
 
 ## 四、限制
-输出**必须**基于`get_video_topic`返回的选题信息生成，**严谨模型自己联想**生成
+输出**必须**基于`get_video_topic`返回的选题信息生成，**严谨模型自己联想**生成。
 
----
+## 五、热门话题参考
+借助`hot_topic_search`工具返回的热门话题，对**高赞case出发搜索词**和**特征出发搜索词**进行补充完善，此步骤匹配到的搜索词**优先级最高**。
 
 ## 六、输出模板
 
@@ -91,8 +92,6 @@ description: 需求分析
 }
 ```
 
----
-
 ## 七、质量自检
 - 是否先完成了`实质/形式`与`上层/下层`双重标注
 - 下层特征是否调用了 `get_video_topic`选题工具做补全

examples/content_finder/tools/__init__.py

@@ -15,6 +15,7 @@ from .aigc_platform_api import create_crawler_plan_by_douyin_content_id, create_
 from .think_and_plan import think_and_plan
 from .find_authors_from_db import find_authors_from_db
 from .get_video_topic import get_video_topic
+from .hot_topic_search import hot_topic_search
 
 __all__ = [
     "douyin_search",
@@ -29,4 +30,5 @@ __all__ = [
     "think_and_plan",
     "find_authors_from_db",
     "get_video_topic",
+    "hot_topic_search",
 ]

examples/content_finder/tools/hot_topic_search.py

@@ -0,0 +1,396 @@
+"""
+每日热点话题检索工具（示例）
+
+调用内部爬虫服务获取“今日热榜”类榜单数据，并按业务规则筛选需要的平台来源。
+"""
+
+import asyncio
+import json
+import logging
+import sys
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional, TypedDict
+
+import requests
+
+def _ensure_import_paths() -> None:
+    """
+    允许直接执行本文件时，也能导入仓库根目录下的 `agent`，
+    以及 content_finder 目录下的 `utils` 等模块。
+    """
+    content_finder_root = Path(__file__).resolve().parents[1]  # .../examples/content_finder
+    repo_root = Path(__file__).resolve().parents[3]  # .../Agent
+    for p in (repo_root, content_finder_root):
+        p_str = str(p)
+        if p_str not in sys.path:
+            sys.path.insert(0, p_str)
+
+
+_ensure_import_paths()
+
+from agent.tools import ToolResult, tool
+from utils.tool_logging import format_tool_result_for_log, log_tool_call
+
+logger = logging.getLogger(__name__)
+
+_LOG_LABEL = "工具调用：hot_topic_search -> 每日热点话题检索（今日热榜）"
+
+HOT_TOPIC_API = "http://crawapi.piaoquantv.com/crawler/jin_ri_re_bang/content_rank"
+DEFAULT_TIMEOUT = 60.0
+MAX_MATCHED_TOPICS = 3
+
+
+class HotTopicItem(TypedDict):
+    title: str
+    heat: str
+
+
+class HotTopicSourceBlock(TypedDict, total=False):
+    source: str
+    jump_url: str
+    type: str
+    topics: List[HotTopicItem]
+
+
+class MatchedHotTopic(TypedDict, total=False):
+    title: str
+    heat: str
+    source: str
+    jump_url: str
+    type: str
+    score: int
+    match_mode: str  # "phrase" | "char" | "none"
+    matched_keywords: List[str]
+    matched_chars: List[str]
+
+
+def _normalize_text(text: str) -> str:
+    return text.strip().lower()
+
+
+def _prepare_feature_keywords(feature_keywords: Optional[List[str]]) -> List[str]:
+    if not feature_keywords:
+        return []
+    cleaned: List[str] = []
+    for kw in feature_keywords:
+        if not isinstance(kw, str):
+            continue
+        kw_norm = kw.strip()
+        if not kw_norm:
+            continue
+        cleaned.append(kw_norm)
+    # 保持顺序去重
+    seen: set[str] = set()
+    deduped: List[str] = []
+    for kw in cleaned:
+        key = _normalize_text(kw)
+        if key in seen:
+            continue
+        seen.add(key)
+        deduped.append(kw)
+    return deduped
+
+
+def _extract_chars_from_keywords(feature_keywords: List[str]) -> List[str]:
+    chars: List[str] = []
+    seen: set[str] = set()
+    for kw in feature_keywords:
+        for ch in kw.strip():
+            if ch.isspace():
+                continue
+            if ch in seen:
+                continue
+            seen.add(ch)
+            chars.append(ch)
+    return chars
+
+
+def _score_title_match(title: str, feature_keywords: List[str]) -> MatchedHotTopic:
+    """
+    匹配策略（业务规则）：
+    - 优先词组（feature_keywords）包含匹配；只要命中任一词组，即进入 phrase 模式
+    - 若一个词组都没命中，再进行“单字/字符”匹配，按命中字符数计分
+    - 返回 score 与命中依据，供后续 prompt 再做相关性判断
+    """
+    title_norm = _normalize_text(title)
+    if not feature_keywords:
+        return MatchedHotTopic(title=title, score=0, match_mode="none", matched_keywords=[], matched_chars=[])
+
+    matched_keywords: List[str] = []
+    for kw in feature_keywords:
+        kw_norm = _normalize_text(kw)
+        if kw_norm and kw_norm in title_norm:
+            matched_keywords.append(kw)
+
+    if matched_keywords:
+        # phrase 模式：命中词组数优先，其次命中词组总长度作为细粒度排序
+        length_bonus = sum(len(k.strip()) for k in matched_keywords)
+        score = 1000 * len(matched_keywords) + length_bonus
+        return MatchedHotTopic(
+            title=title,
+            score=int(score),
+            match_mode="phrase",
+            matched_keywords=matched_keywords,
+            matched_chars=[],
+        )
+
+    # char 模式：仅在“无任何词组命中”时启用
+    keyword_chars = _extract_chars_from_keywords(feature_keywords)
+    title_chars = set(title.strip())
+    matched_chars = [ch for ch in keyword_chars if ch in title_chars]
+    score = len(matched_chars)
+    return MatchedHotTopic(
+        title=title,
+        score=int(score),
+        match_mode="char" if score > 0 else "none",
+        matched_keywords=[],
+        matched_chars=matched_chars,
+    )
+
+
+def _build_summary(
+    *,
+    blocks: List[HotTopicSourceBlock],
+    has_more: bool,
+    next_cursor: Any,
+    feature_keywords: List[str],
+) -> str:
+    lines: List[str] = []
+    total = sum(len(b.get("topics", [])) for b in blocks)
+    if feature_keywords:
+        lines.append(f"标题匹配特征词：{', '.join(feature_keywords)}")
+    else:
+        lines.append("标题匹配特征词：未提供（不过滤，返回全部话题）")
+    lines.append(f"共筛出 {len(blocks)} 个来源块，话题 {total} 条")
+    if has_more:
+        lines.append(f"还有更多，可用 cursor={next_cursor} 继续拉取")
+    lines.append("")
+    for b in blocks:
+        source = b.get("source") or "未知来源"
+        jump_url = b.get("jump_url") or ""
+        b_type = b.get("type") or ""
+        topics = b.get("topics", [])
+        header = f"【{source}】{b_type}".strip()
+        lines.append(header)
+        if jump_url:
+            lines.append(f"榜单页: {jump_url}")
+        for i, t in enumerate(topics[:20], 1):
+            title = t.get("title", "").strip() or "无标题"
+            heat = t.get("heat", "").strip() or "-"
+            lines.append(f"{i}. {title}（{heat}）")
+        if len(topics) > 20:
+            lines.append(f"... 其余 {len(topics) - 20} 条已省略（完整见 metadata）")
+        lines.append("")
+    return "\n".join(lines).rstrip()
+
+
+def _parse_filtered_topics(raw: Dict[str, Any], *, feature_keywords: List[str]) -> Dict[str, Any]:
+    data_block = raw.get("data", {}) if isinstance(raw.get("data"), dict) else {}
+    has_more = bool(data_block.get("has_more", False))
+    next_cursor = data_block.get("next_cursor")
+    items = data_block.get("data", []) if isinstance(data_block.get("data"), list) else []
+
+    candidates: List[MatchedHotTopic] = []
+
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        source = (item.get("source") or "").strip()
+        rank_list = item.get("rankList", []) if isinstance(item.get("rankList"), list) else []
+        for r in rank_list:
+            if not isinstance(r, dict):
+                continue
+            title = (r.get("title") or "").strip()
+            heat = (r.get("heat") or "").strip()
+            if not title:
+                continue
+            scored = _score_title_match(title, feature_keywords)
+            score = int(scored.get("score") or 0)
+            match_mode = str(scored.get("match_mode") or "none")
+            # 要求“词组优先；无词组再按单字”，所以仅保留有得分/有命中的候选
+            if feature_keywords and match_mode == "none":
+                continue
+            candidates.append(
+                MatchedHotTopic(
+                    title=title,
+                    heat=heat,
+                    source=source,
+                    jump_url=item.get("jump_url") or "",
+                    type=item.get("type") or "",
+                    score=score,
+                    match_mode=match_mode,
+                    matched_keywords=list(scored.get("matched_keywords") or []),
+                    matched_chars=list(scored.get("matched_chars") or []),
+                )
+            )
+
+    # 全局排序取 Top3
+    top_topics = sorted(candidates, key=lambda x: int(x.get("score") or 0), reverse=True)[:MAX_MATCHED_TOPICS]
+
+    blocks_by_source: Dict[str, HotTopicSourceBlock] = {}
+    topics_by_source: Dict[str, List[HotTopicItem]] = {}
+    for t in top_topics:
+        source = (t.get("source") or "").strip()
+        if source not in blocks_by_source:
+            blocks_by_source[source] = HotTopicSourceBlock(
+                source=source,
+                jump_url=t.get("jump_url") or "",
+                type=t.get("type") or "",
+                topics=[],
+            )
+        topic_item: HotTopicItem = {"title": t.get("title") or "", "heat": t.get("heat") or ""}
+        blocks_by_source[source].setdefault("topics", []).append(topic_item)
+        topics_by_source.setdefault(source, []).append(topic_item)
+
+    blocks: List[HotTopicSourceBlock] = list(blocks_by_source.values())
+    matched_total = len(top_topics)
+
+    return {
+        "has_more": has_more,
+        "next_cursor": next_cursor,
+        "blocks": blocks,
+        "topics_by_source": topics_by_source,
+        "top_topics": top_topics,
+        "matched_total": matched_total,
+    }
+
+
+@tool(
+    description='检索“今日热榜”热点话题；可传入 feature_keywords 做标题包含匹配，仅保留命中话题（title/heat）。若不传则不做过滤，返回全部话题'
+)
+async def hot_topic_search(
+    sort_type: str = "最热",
+    cursor: int = 1,
+    feature_keywords: Optional[List[str]] = None,
+    timeout: Optional[float] = None,
+) -> ToolResult:
+    """
+    检索每日热点话题（今日热榜）
+
+    Args:
+        sort_type: 榜单排序方式（如 "最热"），默认 "最热"
+        cursor: 分页游标（从 1 开始），默认 1
+        feature_keywords: 实质特征词列表。若传入，则仅保留“标题包含任一特征词”的话题用于补充搜索词；不传入则不做标题匹配过滤（返回全部话题）。
+        timeout: 超时时间（秒），默认 60
+
+    Returns:
+        ToolResult:
+            - output: 人类可读摘要（为节省 token：最多返回命中的前 3 条话题）
+            - metadata.has_more: 是否还有下一页
+            - metadata.next_cursor: 下一页 cursor
+            - metadata.blocks: 按来源块输出的结构化结果（每块 topics 仅含 title/heat）
+            - metadata.topics_by_source: 按来源聚合的话题列表（仅含 title/heat）
+            - metadata.top_topics: Top3 话题明细（含 score/match_mode/命中依据），用于 prompt 再做相关性判断
+            - metadata.matched_total: 实际返回的命中话题总数（<=3）
+            - metadata.feature_keywords: 本次用于标题匹配的特征词（清洗/去重后）
+            - metadata.raw_data: 原始 API 返回
+    """
+    start_time = time.time()
+    request_timeout = timeout if timeout is not None else DEFAULT_TIMEOUT
+    cleaned_keywords = _prepare_feature_keywords(feature_keywords)
+    call_params: Dict[str, Any] = {
+        "sort_type": sort_type,
+        "cursor": cursor,
+        "feature_keywords": cleaned_keywords,
+        "timeout": request_timeout,
+    }
+
+    if not isinstance(sort_type, str) or not sort_type.strip():
+        err = ToolResult(title="热点话题检索失败", output="", error="sort_type 参数无效：必须是非空字符串")
+        log_tool_call(_LOG_LABEL, call_params, format_tool_result_for_log(err))
+        return err
+    if not isinstance(cursor, int) or cursor <= 0:
+        err = ToolResult(title="热点话题检索失败", output="", error="cursor 参数无效：必须是正整数")
+        log_tool_call(_LOG_LABEL, call_params, format_tool_result_for_log(err))
+        return err
+    if feature_keywords is not None and not isinstance(feature_keywords, list):
+        err = ToolResult(title="热点话题检索失败", output="", error="feature_keywords 参数无效：必须是字符串列表或不传")
+        log_tool_call(_LOG_LABEL, call_params, format_tool_result_for_log(err))
+        return err
+
+    try:
+        payload = {"sort_type": sort_type.strip(), "cursor": cursor}
+        response = requests.post(
+            HOT_TOPIC_API,
+            json=payload,
+            headers={"Content-Type": "application/json"},
+            timeout=request_timeout,
+        )
+        response.raise_for_status()
+        raw = response.json()
+    except requests.exceptions.HTTPError as e:
+        logger.error(
+            "hot_topic_search HTTP error",
+            extra={"sort_type": sort_type, "cursor": cursor, "status_code": e.response.status_code},
+        )
+        err = ToolResult(title="热点话题检索失败", output="", error=f"HTTP {e.response.status_code}: {e.response.text}")
+        log_tool_call(_LOG_LABEL, call_params, format_tool_result_for_log(err))
+        return err
+    except requests.exceptions.Timeout:
+        logger.error("hot_topic_search timeout", extra={"sort_type": sort_type, "cursor": cursor, "timeout": request_timeout})
+        err = ToolResult(title="热点话题检索失败", output="", error=f"请求超时（{request_timeout}秒）")
+        log_tool_call(_LOG_LABEL, call_params, format_tool_result_for_log(err))
+        return err
+    except requests.exceptions.RequestException as e:
+        logger.error("hot_topic_search network error", extra={"sort_type": sort_type, "cursor": cursor, "error": str(e)})
+        err = ToolResult(title="热点话题检索失败", output="", error=f"网络错误: {str(e)}")
+        log_tool_call(_LOG_LABEL, call_params, format_tool_result_for_log(err))
+        return err
+    except Exception as e:
+        logger.error(
+            "hot_topic_search unexpected error",
+            extra={"sort_type": sort_type, "cursor": cursor, "error": str(e)},
+            exc_info=True,
+        )
+        err = ToolResult(title="热点话题检索失败", output="", error=f"未知错误: {str(e)}")
+        log_tool_call(_LOG_LABEL, call_params, format_tool_result_for_log(err))
+        return err
+
+    parsed = _parse_filtered_topics(raw if isinstance(raw, dict) else {}, feature_keywords=cleaned_keywords)
+    has_more = bool(parsed.get("has_more", False))
+    next_cursor = parsed.get("next_cursor")
+    blocks = parsed.get("blocks", [])
+    matched_total = int(parsed.get("matched_total") or 0)
+    summary = _build_summary(blocks=blocks, has_more=has_more, next_cursor=next_cursor, feature_keywords=cleaned_keywords)
+
+    duration_ms = int((time.time() - start_time) * 1000)
+    logger.info(
+        "hot_topic_search completed",
+        extra={
+            "sort_type": sort_type,
+            "cursor": cursor,
+            "blocks_count": len(blocks),
+            "has_more": has_more,
+            "next_cursor": next_cursor,
+            "duration_ms": duration_ms,
+        },
+    )
+
+    out = ToolResult(
+        title=f"今日热榜热点话题（{sort_type}，cursor={cursor}）",
+        output=summary,
+        long_term_memory=f"Fetched hot topics sort_type='{sort_type}' cursor={cursor}",
+        metadata={
+            "raw_data": raw,
+            "has_more": has_more,
+            "next_cursor": next_cursor,
+            "blocks": blocks,
+            "topics_by_source": parsed.get("topics_by_source", {}),
+            "top_topics": parsed.get("top_topics", []),
+            "matched_total": matched_total,
+            "feature_keywords": cleaned_keywords,
+        },
+        include_metadata_in_llm=True,
+    )
+    log_tool_call(_LOG_LABEL, call_params, json.dumps(out.metadata.get("topics_by_source", {}), ensure_ascii=False))
+    return out
+
+
+async def main() -> None:
+    result = await hot_topic_search(sort_type="最热", cursor=1)
+    print(result.output)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())