可用版本流程

examples/piaoquan_needs/agent_tools.py

@@ -1,5 +1,5 @@
 from agent.tools import ToolResult, tool
-from examples.piaoquan_needs.topic_build_pattern_tools import _log_tool_output
+from examples.piaoquan_needs.topic_build_pattern_tools import _log_tool_input, _log_tool_output
 
 
 @tool(
@@ -17,6 +17,14 @@ def think_and_plan(thought: str, thought_number: int, action: str, plan: str) ->
     Returns:
         A string describing the thought, plan, and action steps.
     """
+    params = {
+        "thought": thought,
+        "thought_number": thought_number,
+        "action": action,
+        "plan": plan,
+    }
+    _log_tool_input("think_and_plan", params)
+
     result = (
         f"[思考 #{thought_number}]\n"
         f"思考: {thought}\n"

examples/piaoquan_needs/config.py

@@ -10,7 +10,7 @@ from agent.core.runner import RunConfig
 
 RUN_CONFIG = RunConfig(
     # 模型配置
-    model="claude-sonnet-4.5",
+    model="claude-opus-4-6",
     temperature=0.3,
     max_iterations=1000,
 
@@ -29,4 +29,5 @@ LOG_FILE = None  # 设置为文件路径可以同时输出到文件
 ENABLED_TOOLS = ["think_and_plan", "get_category_tree", "get_frequent_itemsets", "get_itemset_detail",
                  "get_post_elements", "search_elements", "get_element_category_chain", "get_category_detail",
                  "search_categories", "get_category_elements", "get_category_co_occurrences",
-                 "get_element_co_occurrences"]
+                 "get_element_co_occurrences",
+                 "get_weight_score_topn", "get_weight_score_by_name"]

examples/piaoquan_needs/data_query_tools.py

@@ -0,0 +1,76 @@
+from odps import ODPS
+from odps.errors import ODPSError
+from datetime import date, timedelta
+
+
+from agent import tool
+
+
+def get_odps_data(sql):
+    # 配置信息
+    access_id = 'LTAI9EBa0bd5PrDa'
+    access_key = 'vAalxds7YxhfOA2yVv8GziCg3Y87v5'
+    project = 'loghubods'
+    endpoint = 'http://service.odps.aliyun.com/api'
+
+    # 1. 初始化 ODPS 入口
+    o = ODPS(access_id, access_key, project, endpoint=endpoint)
+
+    try:
+        # 2. 执行 SQL 并获取结果
+        # execute_sql 会等待任务完成，使用 open_reader 读取数据
+        with o.execute_sql(sql).open_reader() as reader:
+            # reader 类似于 Java 中的 List<Record>
+            # 我们可以直接将其转换为 Python 的 list
+            records = [record for record in reader]
+            return records
+
+    except ODPSError as e:
+        print(f"ODPS 错误: {e}")
+        return None
+
+
+def get_rov_by_merge_leve2_and_video_ids(merge_leve2, video_ids):
+    merge_level_in_clause = f"'{merge_leve2}'"
+    video_ids_in_clause = ", ".join([f"'{video_id}'" for video_id in video_ids])
+    end_date = (date.today() - timedelta(days=1)).strftime("%Y%m%d")
+    start_date = (date.today() - timedelta(days=14)).strftime("%Y%m%d")
+    sql_query = f'''
+SELECT
+    v.videoid,
+    COALESCE(
+        AVG(NULLIF(t3.rov_t0,0)),
+        0
+    ) AS avg_rov_t0
+FROM
+(
+    SELECT
+        t2.videoid,
+        t2.merge_leve2
+    FROM videoods.content_profile t1
+    JOIN loghubods.video_merge_tag t2
+        ON t1.content_id = t2.videoid
+    WHERE
+        t1.status = 3
+        AND t1.is_deleted = 0
+        AND t2.merge_leve2 IN ({merge_level_in_clause})
+) v
+LEFT JOIN loghubods.video_dimension_detail_add_column t3
+    ON v.videoid = t3.视频id
+    AND t3.dt >= '{start_date}'
+    AND t3.dt <= '{end_date}'
+WHERE v.videoid in ({video_ids_in_clause})
+GROUP BY
+    v.videoid
+;
+    '''
+    data = get_odps_data(sql_query)
+    result_dict = {}
+    if data:
+        result_dict = {r[0]: r[1] for r in data}
+    return result_dict
+
+
+if __name__ == '__main__':
+    videos = ["64586310"]
+    print(get_rov_by_merge_leve2_and_video_ids('历史名人', videos))

examples/piaoquan_needs/log_capture.py

@@ -11,7 +11,6 @@ import sys
 import contextvars
 import threading
 from contextlib import contextmanager
-from datetime import datetime
 
 # 当前 Agent 执行绑定的 build_id（通过 contextvars 跨 asyncio.to_thread 传播）
 _current_build_id: contextvars.ContextVar[int | None] = contextvars.ContextVar(
@@ -51,7 +50,7 @@ def build_log(build_id: int):
         with build_log(build_id):
             log("这条会写入 buffer")
             ...
-        # with 结束后自动保存到 DB 并清理
+        # with 结束后仅清理内存缓冲区
     """
     buf = io.StringIO()
     token = _current_build_id.set(build_id)
@@ -62,9 +61,6 @@ def build_log(build_id: int):
     try:
         yield buf
     finally:
-        # 保存到 DB
-        _save_to_db(build_id, buf.getvalue())
-
         # 清理
         with _buffers_lock:
             _buffers.pop(build_id, None)
@@ -89,28 +85,8 @@ def get_log_content(build_id: int) -> str | None:
 
 
 def _save_to_db(build_id: int, content: str) -> bool:
-    """将日志保存到 topic_build_log 表"""
-    if not content:
-        return False
-    try:
-        from db_manager import DatabaseManager
-        from models import TopicBuildLog
-        db = DatabaseManager()
-        session = db.get_session()
-        try:
-            log_entry = TopicBuildLog(
-                build_id=build_id,
-                log_content=content,
-                created_at=datetime.now(),
-            )
-            session.add(log_entry)
-            session.commit()
-            return True
-        finally:
-            session.close()
-    except Exception as e:
-        print(f"[log_capture] 保存日志失败 (build_id={build_id}): {e}", file=_real_stdout)
-        return False
+    """兼容旧接口：已禁用 DB 落库。"""
+    return False
 
 
 # ============================================================================
@@ -142,7 +118,7 @@ class TeeStream(io.TextIOBase):
         return self._buffer.getvalue()
 
     def save_to_db(self, build_id: int) -> bool:
-        return _save_to_db(build_id, self._buffer.getvalue())
+        return False
 
     @property
     def encoding(self):

examples/piaoquan_needs/needs.md

@@ -1,11 +1,24 @@
 ---
-model: anthropic/claude-sonnet-4.5
+model: anthropic/claude-sonnet-4w.5
 temperature: 0.5
+max_iterations: 200
 ---
 
 $system$
-你是一个优秀的agent，能完整的完成任何任务
-
+该文件已改为 Prompt 入口说明，不再承载具体执行流程。
 
 $user$
-使用你可以使用的工具，帮我找到历史名人最佳的需求
+请按任务目标选择以下 prompt：
+
+1. `examples/piaoquan_needs/needs_from_category.md`
+   - 从分类节点出发
+   - 适合先做抽象方向筛选，再做树关系与共现跳转
+
+2. `examples/piaoquan_needs/needs_from_element.md`
+   - 从叶子元素出发
+   - 适合先做具象素材筛选，再向上回溯并做共现扩展
+
+统一要求（两者都遵守）：
+- 结果区只输出树上节点与节点组合
+- 其他解释统一放在 `other_fields`
+- 意图维度不直接参与需求生成，仅在共现关系中辅助使用

examples/piaoquan_needs/needs_from_category.md

@@ -0,0 +1,122 @@
+---
+model: anthropic/claude-sonnet-4.5
+temperature: 0.5
+max_iterations: 200
+---
+
+$system$
+你是一个短视频搜索需求生成专家（分类驱动模式）。
+
+你的任务是：从分类树的"分类节点"出发，基于权重分与共现关系，生成用于搜索视频的候选需求（即搜索词/搜索词组合）。
+
+## 思考输出要求
+
+你在执行过程中，**必须在文本中主动输出你的思考和推理**，而不是只调用工具。具体要求：
+
+1. **行动前先说理由**：每次调用工具之前，先用 1-2 句话说明你为什么要调这个工具、你期望从中得到什么信息、你当前的思路是什么。
+2. **拿到结果后立刻分析**：工具返回数据后，立即输出你对结果的解读——数据说明了什么？有哪些关键发现？是否符合预期？是否需要调整策略？
+3. **阶段性总结**：每个阶段结束时，输出一段简要总结：本阶段做了什么、得到了哪些关键结论、对下一步有什么影响。
+4. **决策透明化**：当你做出筛选/保留/淘汰决策时，必须在文本中明确说明理由（如"X 节点权重仅 0.12，远低于平均水平 0.35，故淘汰"）。
+5. **`think_and_plan` 用于结构化记录**：`think_and_plan` 仍然用于记录计划和关键节点，但它不能替代你在对话中直接输出的思考文本。两者互补，缺一不可。
+
+## 强约束
+
+1. 需求结果区只允许输出：
+   - 树上的单个节点（分类节点）
+   - 树上的节点组合（分类+分类、分类+叶子、叶子+叶子）
+2. 意图维度不直接参与需求生成，仅可在共现关系（跳转）时作为辅助信号。
+3. 每个结论都必须有工具调用证据。
+
+## 效率约束（必须遵守，控制调用开销）
+
+1. **阶段1筛选上限**：从 top_n=20 的结果中，仅对**权重排名前 10** 的分类（实质、形式各最多 10 个）做后续展开（`search_categories`、`get_category_detail`）。排名靠后的仅记录名称和权重备查。
+2. **阶段2下钻收敛**：对高权重父分类做 `get_category_detail` 获取 children 后，仅对 **children 中权重排名前 5** 的子分类查权重。总查询次数不超过 20 次。
+3. **阶段3共现精简**：`get_category_co_occurrences` 的 `top_n` 设为 **15**（不要 30）。只对 **post_count >= 3** 的共现分类做后续下钻（`get_category_elements` 或元素权重查询）。弱共现（post_count < 3）仅记录，不展开。
+4. **元素权重预判**：若目标分类的权重已低于 **0.1**，不再下钻查该分类下的具体元素权重。
+5. **阶段间收敛**：每个阶段结束时，必须输出一个**精简候选列表**（不超过 12 条），只有进入候选列表的项才允许在下一阶段继续展开。
+
+## 质量约束（提升输出可用性）
+
+1. **过滤过于宽泛的节点**：如"标题"、"素材"、"解说"等过于通用、缺乏搜索指向性的词，不可作为单节点需求，但可参与组合。
+2. **组合最低证据门槛**：节点组合必须有共现 post_count >= 3 的证据支撑；若证据仅 1-2 帖，不可进入最终结果。
+3. **desc 必须包含搜索用途说明**：`desc` 除了写证据来源外，**必须用一句话描述这个需求可以用来搜索什么类型的视频**（例如："可用于搜索'伟人鲜为人知的多重身份'相关的盘点类短视频"）。
+4. **主题多样性**：最终结果中，**同一核心人物/主题的需求不超过总数的 40%**。如发现结果过于集中，需主动补充其他人物/事件/维度的需求。
+5. **维度平衡**：单节点需求中，实质维度和形式维度的节点数量差距不超过 5 条。
+6. **输出前自检**：生成最终 JSON 前，必须做一轮去重检查（相同节点不同组合顺序视为重复）和质量过滤（移除证据不足或过于宽泛的条目）。
+
+## 可用工具（按目的）
+
+- 分类权重：`get_weight_score_topn(level="分类", ...)`、`get_weight_score_by_name(level="分类", ...)`
+- 节点定位：`search_categories`、`get_category_detail`
+- 节点下钻：`get_category_elements`
+- 关系跳转：`get_category_co_occurrences`
+- 过程记录：`think_and_plan`
+
+$user$
+## 任务
+
+针对二级品类「%merge_leve2%」，从"分类节点"出发完成搜索视频需求生成。
+
+## 执行流程
+
+### 阶段1：高权重分类筛选（仅实质/形式）
+1. 调用：
+   - `get_weight_score_topn(level="分类", dimension="实质", top_n=20)`
+   - `get_weight_score_topn(level="分类", dimension="形式", top_n=20)`
+2. 各维度按权重排序，**只选排名前 10 的分类**做后续展开；排名 11-20 的仅记录名称和权重备查。
+3. 选择起点时需综合考虑：**权重值**和**覆盖帖数（post_ids_count）**。优先保留"高权重+多帖覆盖"的分类；帖数过少（<= 2）的高权重分类需特别标注其"不稳定"特征。
+4. 使用 `search_categories` 获取 category_id，使用 `get_category_detail` 校验节点在树中的层级位置（是否为可用方向节点）。
+5. **输出你的筛选思考**：哪些分类权重突出？为什么选择这些作为起点？有没有意外发现？
+6. **输出本阶段精简候选列表**（<= 12 条），明确列出进入阶段2的分类及其理由。
+7. `think_and_plan` 记录筛选逻辑。
+
+### 阶段2：树内组合（父子）
+1. 仅对阶段1候选列表中的分类做下钻。对高权重父分类，使用 `get_category_detail` 获取 children。
+2. 子节点用 `get_weight_score_by_name(level="分类", ...)` 查权重并排序，**仅对排名前 5 的子分类**查权重。总查询次数控制在 20 次以内。
+3. 形成"父->子"组合（父分类 + Top子分类）。
+4. **输出你的组合分析**：父子组合中哪些路径价值最高？子节点权重分布是否集中还是分散？你的组合策略是什么？
+5. **输出本阶段精简候选列表**：列出进入阶段3的高权重分类（**<= 6 个**），以及已确认的高价值"父+子"组合。
+6. `think_and_plan` 记录保留/淘汰原因。
+
+### 阶段3：关系组合（分类驱动）
+1. 仅对阶段2候选列表中的 <= 6 个高权重分类做共现查询。调用 `get_category_co_occurrences(category_ids=[A], top_n=15)`（注意 **top_n=15**），跳转到共现分类 B。
+2. 只对 **post_count >= 3** 的共现分类做后续展开。弱共现（post_count < 3）仅记录名称，不做下钻。
+3. 形成"分类+分类"组合（A+B）。
+4. 对 B 用 `get_category_elements` 取元素，并用 `get_weight_score_by_name(level="元素", ...)` 找最高权重叶子，形成"分类+叶子"组合（A+leaf_B）。若分类 B 的权重 < 0.1，跳过元素权重查询，直接使用 `get_category_elements` 返回的高频元素（按 occurrence_count 排序取前 3）。
+5. 对 A 与 B 各取最高权重叶子，形成"叶子+叶子"组合（leaf_A+leaf_B）。
+6. 若共现链路出现意图节点，只作为解释证据，不可直接进入结果区。
+7. **输出你的共现分析**：共现关系揭示了什么模式？哪些跨维度组合最有创意或最有市场潜力？为什么？
+8. `think_and_plan` 记录共现链路价值。
+
+## 输出格式（必须严格遵守）
+
+仅输出 JSON List，格式固定为（`words` 为非空节点列表，可为单节点或多节点组合）：
+
+```json
+[
+  {
+    "words": ["节点A"],
+    "desc": "权重/证据说明 | 搜索方向：可以用来搜索什么类型的短视频"
+  },
+  {
+    "words": ["节点A", "节点B"],
+    "desc": "共现/组合证据说明 | 搜索方向：可以用来搜索什么类型的短视频"
+  },
+  {
+    "words": ["节点A", "节点B", "节点C"],
+    "desc": "多节点组合链路说明 | 搜索方向：可以用来搜索什么类型的短视频"
+  }
+]
+```
+
+## 额外要求
+
+- `words` 只放树上的节点名称（单节点或节点组合），不写完整句子。
+- `words` 节点数量不设上限，只要求为非空列表；节点顺序需能反映你的组合逻辑。
+- 最终结果中，"单节点"条数不少于 20 条，"节点组合"条数不少于 20 条（两类都必须达标）。
+- `desc` 格式为 `"证据说明 | 搜索方向：xxx"`，前半部分放数据证据，**后半部分必须用一句话描述可以用来搜索什么类型的视频**。
+- 过于宽泛的节点（如"标题"、"素材"、"解说"等不具有搜索指向性的词）不可作为单节点需求，但可参与组合。
+- 节点组合必须有共现 post_count >= 3 的证据支撑。
+- 同一核心人物/主题的需求不超过总数的 40%。
+- 生成最终 JSON 前，做一轮去重和质量自检。
+- 若证据不足，宁可减少结果条数，不可臆造节点组合。

examples/piaoquan_needs/needs_from_element.md

@@ -0,0 +1,123 @@
+---
+model: anthropic/claude-sonnet-4.5
+temperature: 0.5
+max_iterations: 200
+---
+
+$system$
+你是一个短视频搜索需求生成专家（元素驱动模式）。
+
+你的任务是：从树的叶子元素出发，基于权重分、向上回溯与共现关系，生成用于搜索视频的候选需求（即搜索词/搜索词组合）。
+
+## 思考输出要求
+
+你在执行过程中，**必须在文本中主动输出你的思考和推理**，而不是只调用工具。具体要求：
+
+1. **行动前先说理由**：每次调用工具之前，先用 1-2 句话说明你为什么要调这个工具、你期望从中得到什么信息、你当前的思路是什么。
+2. **拿到结果后立刻分析**：工具返回数据后，立即输出你对结果的解读——数据说明了什么？有哪些关键发现？是否符合预期？是否需要调整策略？
+3. **阶段性总结**：每个阶段结束时，输出一段简要总结：本阶段做了什么、得到了哪些关键结论、对下一步有什么影响。
+4. **决策透明化**：当你做出筛选/保留/淘汰决策时，必须在文本中明确说明理由（如"X 元素权重 0.45 位列前 3，且所属分类链路完整，故保留为核心起点"）。
+5. **`think_and_plan` 用于结构化记录**：`think_and_plan` 仍然用于记录计划和关键节点，但它不能替代你在对话中直接输出的思考文本。两者互补，缺一不可。
+
+## 强约束
+
+1. 需求结果区只允许输出：
+   - 树上的单个节点（叶子节点/分类节点）
+   - 树上的节点组合（叶子+叶子、叶子+分类、父+子）
+2. 意图维度不直接参与需求生成，仅可在共现关系（跳转）时作为辅助信号。
+3. 每个结论都必须有工具调用证据。
+
+## 效率约束（必须遵守，控制调用开销）
+
+1. **阶段1链路查询上限**：从 top_n=20 的结果中，仅对**权重排名前 10** 的元素（实质、形式各最多 10 个）调用 `get_element_category_chain`。对排名靠后的元素只记录名称和权重，不做链路展开。
+2. **阶段2查询收敛**：只对链路中 **level 3~5 的中间层分类**查权重（太高层如 level 1-2 通用性过强无意义，太低层如 level 6-7 通常和叶子本身重复）。每条链路最多向上查 2 级，总查询次数不超过 20 次。
+3. **阶段3共现精简**：`get_category_co_occurrences` 的 `top_n` 设为 **15**（不要 30）。只对 **post_count >= 3** 的共现分类做后续下钻（`get_category_elements` 或元素权重查询）。弱共现（post_count < 3）仅记录，不展开。
+4. **元素权重预判**：若目标分类的权重已低于 **0.1**，不再下钻查该分类下的具体元素权重。
+5. **阶段间收敛**：每个阶段结束时，必须输出一个**精简候选列表**（不超过 12 条），只有进入候选列表的项才允许在下一阶段继续展开。
+
+## 质量约束（提升输出可用性）
+
+1. **过滤过于宽泛的节点**：如"标题"、"素材"、"解说"等过于通用、缺乏搜索指向性的词，不可作为单节点需求，但可参与组合。
+2. **组合最低证据门槛**：节点组合必须有共现 post_count >= 3 的证据支撑；若证据仅 1-2 帖，不可进入最终结果。
+3. **desc 必须包含搜索用途说明**：`desc` 除了写证据来源外，**必须用一句话描述这个需求可以用来搜索什么类型的视频**（例如："可用于搜索'伟人鲜为人知的多重身份'相关的盘点类短视频"）。
+4. **主题多样性**：最终结果中，**同一核心人物/主题的需求不超过总数的 40%**。如发现结果过于集中，需主动补充其他人物/事件/维度的需求。
+5. **维度平衡**：单节点需求中，实质维度和形式维度的节点数量差距不超过 5 条。
+6. **输出前自检**：生成最终 JSON 前，必须做一轮去重检查（相同节点不同组合顺序视为重复）和质量过滤（移除证据不足或过于宽泛的条目）。
+
+## 可用工具（按目的）
+
+- 元素权重：`get_weight_score_topn(level="元素", ...)`、`get_weight_score_by_name(level="元素", ...)`
+- 向上回溯：`get_element_category_chain`
+- 分类定位：`search_categories`、`get_category_detail`
+- 共现跳转：`get_category_co_occurrences`
+- 元素池：`get_category_elements`
+- 过程记录：`think_and_plan`
+
+$user$
+## 任务
+
+针对二级品类「%merge_leve2%」，从"叶子元素"出发完成搜索视频需求生成。
+
+## 执行流程
+
+### 阶段1：高权重叶子筛选（仅实质/形式）
+1. 调用：
+   - `get_weight_score_topn(level="元素", dimension="实质", top_n=20)`
+   - `get_weight_score_topn(level="元素", dimension="形式", top_n=20)`
+2. 各维度按权重排序，**只选排名前 10 的叶子元素**作为起点；排名 11-20 的仅记录名称和权重备查，不做链路展开。
+3. 选择起点时需综合考虑：**权重值**和**覆盖帖数（post_ids_count）**。优先保留"高权重+多帖覆盖"的元素；仅1帖的高权重元素需特别标注其"不稳定"特征。
+4. 对选出的起点叶子调用 `get_element_category_chain` 获取分类路径。
+5. **输出你的筛选思考**：哪些叶子元素权重突出？它们分别属于什么维度和分类？你选择这些起点的理由是什么？有没有意外发现？
+6. **输出本阶段精简候选列表**（<= 12 条），明确列出进入阶段2的元素及其理由。
+7. `think_and_plan` 记录候选叶子的保留标准。
+
+### 阶段2：树内组合（儿子->父亲）
+1. 仅对阶段1候选列表中的元素做回溯。从分类链提取向上1跳父分类、向上2跳祖父分类（仅查 **level 3~5 的中间层分类**，跳过 level 1-2 和 level 6-7）。
+2. 对父/祖父分类用 `get_weight_score_by_name(level="分类", ...)` 查权重，保留高权重路径。每条链路最多向上查 2 级，**总查询次数控制在 20 次以内**。
+3. 形成"叶子+父分类""叶子+祖父分类"组合。
+4. **输出你的回溯分析**：向上回溯后发现了哪些有价值的路径？父分类/祖父分类的权重是否支持这条链路？哪些组合值得保留，为什么？
+5. **输出本阶段精简候选列表**：列出进入阶段3的高权重分类（**<= 6 个**），以及已确认的高价值"叶子+分类"组合。
+6. `think_and_plan` 记录高价值回溯路径。
+
+### 阶段3：关系组合（元素驱动）
+1. 仅对阶段2候选列表中的 <= 6 个高权重分类做共现查询。调用 `get_category_co_occurrences(category_ids=[A], top_n=15)`（注意 **top_n=15**），跳转到共现分类 B。
+2. 只对 **post_count >= 3** 的共现分类做后续展开。弱共现（post_count < 3）仅记录名称，不调用 `get_category_elements` 或元素权重查询。
+3. 在合格的共现分类 B 下用 `get_category_elements` 获取元素，并用 `get_weight_score_by_name(level="元素", ...)` 取最高权重叶子 leaf_B。若分类 B 的权重 < 0.1，跳过元素权重查询，直接使用 `get_category_elements` 返回的高频元素（按 occurrence_count 排序取前 3）。
+4. 形成"叶子+叶子"组合（leaf_A + leaf_B）。
+5. 必要时补充"分类+叶子"组合（A + leaf_B）用于结构完整性。
+6. 若链路出现意图节点，只作为解释证据，不可直接进入结果区。
+7. **输出你的共现分析**：共现关系揭示了什么模式？跨维度的叶子组合有哪些潜在价值？为什么这些组合比随机配对更有意义？
+8. `think_and_plan` 记录共现跳转有效性。
+
+## 输出格式（必须严格遵守）
+
+仅输出 JSON List，格式固定为（`words` 为非空节点列表，可为单节点或多节点组合）：
+
+```json
+[
+  {
+    "words": ["节点A"],
+    "desc": "权重/证据说明 | 搜索方向：可以用来搜索什么类型的短视频"
+  },
+  {
+    "words": ["节点A", "节点B"],
+    "desc": "共现/组合证据说明 | 搜索方向：可以用来搜索什么类型的短视频"
+  },
+  {
+    "words": ["节点A", "节点B", "节点C"],
+    "desc": "多节点组合链路说明 | 搜索方向：可以用来搜索什么类型的短视频"
+  }
+]
+```
+
+## 额外要求
+
+- `words` 只放树上的节点名称（单节点或节点组合），不写完整句子。
+- `words` 节点数量不设上限，只要求为非空列表；节点顺序需能反映你的组合逻辑。
+- 最终结果中，"单节点"条数不少于 20 条，"节点组合"条数不少于 20 条（两类都必须达标）。
+- `desc` 格式为 `"证据说明 | 搜索方向：xxx"`，前半部分放数据证据，**后半部分必须用一句话描述可以用来搜索什么类型的视频**。
+- 过于宽泛的节点（如"标题"、"素材"、"解说"等不具有搜索指向性的词）不可作为单节点需求，但可参与组合。
+- 节点组合必须有共现 post_count >= 3 的证据支撑。
+- 同一核心人物/主题的需求不超过总数的 40%。
+- 生成最终 JSON 前，做一轮去重和质量自检。
+- 若证据不足，宁可减少结果条数，不可臆造节点组合。

examples/piaoquan_needs/prepare.py

@@ -0,0 +1,161 @@
+import json
+from collections import defaultdict
+from pathlib import Path
+
+from db_manager import DatabaseManager
+from data_query_tools import get_rov_by_merge_leve2_and_video_ids
+from models import TopicPatternElement, TopicPatternExecution
+
+db = DatabaseManager()
+
+
+def _safe_float(value):
+    if value is None:
+        return 0.0
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return 0.0
+
+
+def _build_category_scores(name_scores, name_paths, name_post_ids):
+    """
+    计算分类路径节点权重：
+    - 一个 name 的 score 贡献给其路径上的每个节点
+    - 同一个 name 多条路径时，每条路径都累加
+    """
+    node_scores = defaultdict(float)
+    node_post_ids = defaultdict(set)
+    for name, score in name_scores.items():
+        paths = name_paths.get(name, set())
+        post_ids = name_post_ids.get(name, set())
+        for category_path in paths:
+            if not category_path:
+                continue
+            nodes = [segment.strip() for segment in category_path.split(">") if segment.strip()]
+            for idx in range(len(nodes)):
+                prefix = ">".join(nodes[: idx + 1])
+                node_scores[prefix] += score
+                if post_ids:
+                    node_post_ids[prefix].update(post_ids)
+    return node_scores, node_post_ids
+
+
+def _write_json(path, payload):
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(payload, f, ensure_ascii=False, indent=2)
+
+
+def prepare(execution_id):
+    session = db.get_session()
+    try:
+        execution = session.query(TopicPatternExecution).filter(
+            TopicPatternExecution.id == execution_id
+        ).first()
+        if not execution:
+            raise ValueError(f"execution_id 不存在: {execution_id}")
+
+        merge_leve2 = execution.merge_leve2
+
+        rows = session.query(TopicPatternElement).filter(
+            TopicPatternElement.execution_id == execution_id
+        ).all()
+        if not rows:
+            return {"message": "没有可处理的数据", "execution_id": execution_id}
+
+        # 1) 去重 post_id 拉取 ROV
+        all_post_ids = sorted({r.post_id for r in rows if r.post_id})
+        rov_by_post_id = get_rov_by_merge_leve2_and_video_ids(merge_leve2, all_post_ids) if all_post_ids else {}
+
+        # 2) 按 element_type 分组，计算 name 的平均 ROV 分
+        grouped = {
+            "实质": {
+                "name_post_ids": defaultdict(set),
+                "name_paths": defaultdict(set),
+            },
+            "形式": {
+                "name_post_ids": defaultdict(set),
+                "name_paths": defaultdict(set),
+            },
+            "意图": {
+                "name_post_ids": defaultdict(set),
+                "name_paths": defaultdict(set),
+            },
+        }
+
+        for r in rows:
+            element_type = (r.element_type or "").strip()
+            if element_type not in grouped:
+                continue
+            name = (r.name or "").strip()
+            if not name:
+                continue
+
+            if r.post_id:
+                grouped[element_type]["name_post_ids"][name].add(r.post_id)
+            if r.category_path:
+                grouped[element_type]["name_paths"][name].add(r.category_path.strip())
+
+        output_dir = Path(__file__).parent / "data" / str(execution_id)
+        output_dir.mkdir(parents=True, exist_ok=True)
+
+        summary = {"execution_id": execution_id, "merge_leve2": merge_leve2, "files": {}}
+
+        for element_type, data in grouped.items():
+            name_post_ids = data["name_post_ids"]
+            name_paths = data["name_paths"]
+
+            name_scores = {}
+            for name, post_ids in name_post_ids.items():
+                rovs = [_safe_float(rov_by_post_id.get(pid, 0.0)) for pid in post_ids]
+                score = sum(rovs) / len(rovs) if rovs else 0.0
+                name_scores[name] = score
+
+            element_payload = sorted(
+                [
+                    {
+                        "name": name,
+                        "score": round(score, 6),
+                        "post_ids": sorted(list(name_post_ids.get(name, set()))),
+                        "post_ids_count": len(name_post_ids.get(name, set())),
+                        "category_paths": sorted(list(name_paths.get(name, set()))),
+                    }
+                    for name, score in name_scores.items()
+                ],
+                key=lambda x: x["score"],
+                reverse=True,
+            )
+
+            # 3) 计算分类路径节点权重（节点分 = 覆盖的 name score 求和）
+            category_scores, category_post_ids = _build_category_scores(
+                name_scores, name_paths, name_post_ids
+            )
+            category_payload = sorted(
+                [
+                    {
+                        "category_path": path,
+                        "category": path.split(">")[-1].strip() if path else "",
+                        "score": round(score, 6),
+                        "post_ids_count": len(category_post_ids.get(path, set())),
+                    }
+                    for path, score in category_scores.items()
+                ],
+                key=lambda x: x["score"],
+                reverse=True,
+            )
+
+            element_file = output_dir / f"{element_type}_元素.json"
+            category_file = output_dir / f"{element_type}_分类.json"
+            _write_json(element_file, element_payload)
+            _write_json(category_file, category_payload)
+
+            summary["files"][f"{element_type}_元素"] = str(element_file)
+            summary["files"][f"{element_type}_分类"] = str(category_file)
+
+        return summary
+    finally:
+        session.close()
+
+
+if __name__ == '__main__':
+    prepare(17)

examples/piaoquan_needs/render_log_html.py

@@ -0,0 +1,385 @@
+#!/usr/bin/env python3
+"""将 run_log 文本渲染为可折叠 HTML 页面。
+
+直接在脚本内修改 INPUT_LOG_PATH / OUTPUT_HTML_PATH 后运行:
+    python examples/piaoquan_needs/render_log_html.py
+"""
+
+from __future__ import annotations
+
+import html
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+@dataclass
+class Node:
+    title: str | None = None
+    entries: list[str | "Node"] = field(default_factory=list)
+
+    @property
+    def is_fold(self) -> bool:
+        return self.title is not None
+
+
+def parse_log(content: str) -> Node:
+    root = Node(title=None)
+    stack: list[Node] = [root]
+
+    for raw_line in content.splitlines():
+        line = raw_line.rstrip("\n")
+        if line.startswith("[FOLD:") and line.endswith("]"):
+            title = line[len("[FOLD:") : -1]
+            node = Node(title=title)
+            stack[-1].entries.append(node)
+            stack.append(node)
+            continue
+        if line == "[/FOLD]":
+            if len(stack) > 1:
+                stack.pop()
+            else:
+                root.entries.append(line)
+            continue
+        stack[-1].entries.append(line)
+
+    while len(stack) > 1:
+        unclosed = stack.pop()
+        # 容错: 遇到缺失 [/FOLD] 时，保留原有内容，不丢日志
+        stack[-1].entries.append(unclosed)
+    return root
+
+
+DEFAULT_COLLAPSE_PREFIXES = ["🔧", "📥", "📤"]
+DEFAULT_COLLAPSE_KEYWORDS = ["调用参数", "返回内容"]
+
+# 工具功能摘要（静态映射，用于日志可视化展示）
+TOOL_DESCRIPTION_MAP: dict[str, str] = {
+    "think_and_plan": "系统化记录思考、计划与下一步行动，不查询数据也不修改数据。",
+    "get_weight_score_topn": "按层级和维度查询权重分 TopN，快速定位高权重元素或分类。",
+    "get_weight_score_by_name": "按名称精确查询指定元素或分类的权重分，返回匹配明细。",
+    "get_category_tree": "获取分类树快照，快速查看实质/形式/意图三维结构全貌。",
+    "get_frequent_itemsets": "查询高频共现的分类组合，按维度模式和深度分组返回。",
+    "get_itemset_detail": "获取频繁项集完整详情，包括项集结构和匹配帖子列表。",
+    "get_post_elements": "按帖子查看结构化元素内容，支持点类型及三维元素下钻。",
+    "search_elements": "按关键词搜索元素，返回分类归属、出现频次与帖子覆盖。",
+    "get_element_category_chain": "从元素名称反查所属分类链，查看从根到叶的路径。",
+    "get_category_detail": "查看分类节点上下文，含祖先、子节点、兄弟节点与元素。",
+    "search_categories": "按关键词搜索分类节点，返回分类 ID 与路径等导航信息。",
+    "get_category_elements": "获取指定分类下的元素列表及出现统计，便于落地选题。",
+    "get_category_co_occurrences": "查询分类级共现关系，发现目标分类常同现的其他分类。",
+    "get_element_co_occurrences": "查询元素级共现关系，发现目标元素常同现的其他元素。",
+}
+
+# =========================
+# 运行配置（直接改变量即可）
+# =========================
+INPUT_LOG_PATH = "examples/piaoquan_needs/output/run_log_17_20260324_204533.txt"
+# 设为 None 则默认生成到输入文件同名 .html
+OUTPUT_HTML_PATH: str | None = None
+# 是否默认折叠所有 [FOLD] 块
+COLLAPSE_ALL_FOLDS = False
+# 命中这些前缀/关键词的折叠块默认收起
+COLLAPSE_PREFIXES = DEFAULT_COLLAPSE_PREFIXES
+COLLAPSE_KEYWORDS = DEFAULT_COLLAPSE_KEYWORDS
+
+
+def resolve_config_path(path_str: str) -> Path:
+    """解析配置中的路径，兼容从项目根目录或脚本目录运行。"""
+    raw = Path(path_str).expanduser()
+    if raw.is_absolute():
+        return raw.resolve()
+
+    cwd_candidate = (Path.cwd() / raw).resolve()
+    if cwd_candidate.exists():
+        return cwd_candidate
+
+    script_dir = Path(__file__).resolve().parent
+    script_candidate = (script_dir / raw).resolve()
+    if script_candidate.exists():
+        return script_candidate
+
+    project_root = script_dir.parent.parent
+    project_candidate = (project_root / raw).resolve()
+    if project_candidate.exists():
+        return project_candidate
+
+    # 如果都不存在，返回项目根拼接结果，便于报错信息更稳定
+    return project_candidate
+
+
+def should_collapse(
+    title: str,
+    collapse_prefixes: list[str],
+    collapse_keywords: list[str],
+    collapse_all: bool,
+) -> bool:
+    if collapse_all:
+        return True
+    if any(title.startswith(prefix) for prefix in collapse_prefixes):
+        return True
+    return any(keyword in title for keyword in collapse_keywords)
+
+
+def render_text_block(lines: list[str]) -> str:
+    if not lines:
+        return ""
+
+    normalized = lines[:]
+    while normalized and normalized[0].strip() == "":
+        normalized.pop(0)
+    while normalized and normalized[-1].strip() == "":
+        normalized.pop()
+    if not normalized:
+        return ""
+
+    compact: list[str] = []
+    empty_streak = 0
+    for line in normalized:
+        if line.strip() == "":
+            empty_streak += 1
+            if empty_streak <= 1:
+                compact.append("")
+        else:
+            empty_streak = 0
+            compact.append(line)
+
+    escaped = html.escape("\n".join(compact))
+    return f'<pre class="log-text">{escaped}</pre>'
+
+
+def enrich_fold_title(title: str) -> str:
+    """为工具调用标题附加工具功能描述。"""
+    tool_prefix = "🔧 "
+    if not title.startswith(tool_prefix):
+        return title
+
+    tool_name = title[len(tool_prefix):].strip()
+    description = TOOL_DESCRIPTION_MAP.get(tool_name)
+    if not description:
+        return title
+    return f"{tool_prefix}{tool_name}（{description}）"
+
+
+def render_node(
+    node: Node,
+    collapse_prefixes: list[str],
+    collapse_keywords: list[str],
+    collapse_all: bool,
+) -> str:
+    parts: list[str] = []
+    text_buffer: list[str] = []
+
+    def flush_text_buffer() -> None:
+        if text_buffer:
+            parts.append(render_text_block(text_buffer))
+            text_buffer.clear()
+
+    for entry in node.entries:
+        if isinstance(entry, str):
+            text_buffer.append(entry)
+            continue
+
+        child = entry
+        if child.is_fold:
+            flush_text_buffer()
+            title = child.title or ""
+            is_collapsed = should_collapse(
+                title=title,
+                collapse_prefixes=collapse_prefixes,
+                collapse_keywords=collapse_keywords,
+                collapse_all=collapse_all,
+            )
+            folded_class = "fold tool-fold" if is_collapsed else "fold normal-fold"
+            open_attr = "" if is_collapsed else " open"
+            display_title = enrich_fold_title(title)
+            inner = render_node(
+                child,
+                collapse_prefixes=collapse_prefixes,
+                collapse_keywords=collapse_keywords,
+                collapse_all=collapse_all,
+            )
+            parts.append(
+                f'<details class="{folded_class}"{open_attr}>'
+                f'<summary>{html.escape(display_title)}</summary>'
+                f"{inner}"
+                "</details>"
+            )
+
+    flush_text_buffer()
+
+    return "".join(parts)
+
+
+def build_html(body: str, source_name: str) -> str:
+    return f"""<!doctype html>
+<html lang="zh-CN">
+<head>
+  <meta charset="UTF-8" />
+  <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+  <title>Run Log 可视化 - {html.escape(source_name)}</title>
+  <style>
+    :root {{
+      --bg: #0b1020;
+      --panel: #131a2a;
+      --text: #e8edf7;
+      --muted: #98a2b3;
+      --accent: #6ea8fe;
+      --border: #263146;
+    }}
+    * {{
+      box-sizing: border-box;
+    }}
+    body {{
+      margin: 0;
+      background: var(--bg);
+      color: var(--text);
+      font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, "Helvetica Neue", Arial, "Noto Sans", sans-serif;
+    }}
+    .wrap {{
+      max-width: 1200px;
+      margin: 0 auto;
+      padding: 20px;
+    }}
+    .header {{
+      margin-bottom: 14px;
+      display: flex;
+      align-items: center;
+      gap: 10px;
+      flex-wrap: wrap;
+    }}
+    .title {{
+      font-size: 18px;
+      font-weight: 700;
+    }}
+    .source {{
+      color: var(--muted);
+      font-size: 13px;
+    }}
+    button {{
+      border: 1px solid var(--border);
+      background: var(--panel);
+      color: var(--text);
+      padding: 6px 10px;
+      border-radius: 8px;
+      cursor: pointer;
+    }}
+    button:hover {{
+      border-color: var(--accent);
+      color: var(--accent);
+    }}
+    .content {{
+      background: var(--panel);
+      border: 1px solid var(--border);
+      border-radius: 10px;
+      padding: 10px;
+    }}
+    details {{
+      margin: 6px 0;
+      border: 1px solid var(--border);
+      border-radius: 8px;
+      background: rgba(255, 255, 255, 0.01);
+    }}
+    details > summary {{
+      cursor: pointer;
+      padding: 8px 10px;
+      font-size: 13px;
+      list-style: none;
+      user-select: none;
+      color: #cdd6e5;
+    }}
+    details > summary::-webkit-details-marker {{
+      display: none;
+    }}
+    details > summary::before {{
+      content: "▶";
+      display: inline-block;
+      margin-right: 6px;
+      transform: rotate(0deg);
+      transition: transform 120ms ease;
+      color: var(--muted);
+    }}
+    details[open] > summary::before {{
+      transform: rotate(90deg);
+    }}
+    .tool-fold > summary {{
+      color: #f6cf76;
+    }}
+    .log-text {{
+      margin: 0;
+      padding: 10px;
+      border-top: 1px dashed var(--border);
+      color: var(--text);
+      white-space: pre-wrap;
+      word-break: break-word;
+      line-height: 1.4;
+      font-size: 13px;
+      font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", monospace;
+    }}
+  </style>
+</head>
+<body>
+  <div class="wrap">
+    <div class="header">
+      <div class="title">Run Log 可视化</div>
+      <div class="source">{html.escape(source_name)}</div>
+      <button id="expand-tools">展开全部工具调用</button>
+      <button id="collapse-tools">折叠全部工具调用</button>
+    </div>
+    <div class="content">{body}</div>
+  </div>
+  <script>
+    const toolFolds = Array.from(document.querySelectorAll("details.tool-fold"));
+    document.getElementById("expand-tools").addEventListener("click", () => {{
+      toolFolds.forEach((el) => (el.open = true));
+    }});
+    document.getElementById("collapse-tools").addEventListener("click", () => {{
+      toolFolds.forEach((el) => (el.open = false));
+    }});
+  </script>
+</body>
+</html>
+"""
+
+
+def generate_html(
+    input_path: Path,
+    output_path: Path,
+    collapse_prefixes: list[str],
+    collapse_keywords: list[str],
+    collapse_all: bool = False,
+) -> None:
+    content = input_path.read_text(encoding="utf-8")
+    tree = parse_log(content)
+    body = render_node(
+        tree,
+        collapse_prefixes=collapse_prefixes,
+        collapse_keywords=collapse_keywords,
+        collapse_all=collapse_all,
+    )
+    html_content = build_html(body=body, source_name=input_path.name)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    output_path.write_text(html_content, encoding="utf-8")
+
+
+def main() -> None:
+    input_path = resolve_config_path(INPUT_LOG_PATH)
+    if not input_path.exists():
+        raise FileNotFoundError(f"输入文件不存在: {input_path}")
+
+    if OUTPUT_HTML_PATH:
+        output_path = resolve_config_path(OUTPUT_HTML_PATH)
+    else:
+        output_path = input_path.with_suffix(".html")
+
+    generate_html(
+        input_path=input_path,
+        output_path=output_path,
+        collapse_prefixes=COLLAPSE_PREFIXES,
+        collapse_keywords=COLLAPSE_KEYWORDS,
+        collapse_all=COLLAPSE_ALL_FOLDS,
+    )
+    print(f"HTML 已生成: {output_path}")
+
+
+if __name__ == "__main__":
+    main()

examples/piaoquan_needs/run.py

@@ -5,6 +5,7 @@ import copy
 import importlib
 import os
 import sys
+from datetime import datetime
 from pathlib import Path
 
 from dotenv import load_dotenv
@@ -32,6 +33,7 @@ from agent.llm import create_openrouter_llm_call
 from agent.llm.prompts import SimplePrompt
 from agent.trace import FileSystemTraceStore, Message, Trace
 from agent.utils import setup_logging
+from log_capture import build_log, log
 
 # 导入项目配置
 from examples.piaoquan_needs.config import DEBUG, LOG_FILE, LOG_LEVEL, RUN_CONFIG, TRACE_STORE_PATH
@@ -51,9 +53,10 @@ CUSTOM_TOOL_MODULES = {
     "get_category_elements": "examples.piaoquan_needs.topic_build_pattern_tools",
     "get_category_co_occurrences": "examples.piaoquan_needs.topic_build_pattern_tools",
     "get_element_co_occurrences": "examples.piaoquan_needs.topic_build_pattern_tools",
+    "get_weight_score_topn": "examples.piaoquan_needs.weight_score_query_tools",
+    "get_weight_score_by_name": "examples.piaoquan_needs.weight_score_query_tools",
+
 
-    # 兼容旧示例/占位：当前示例未启用
-    "echo_text": "examples.piaoquan_needs.tool.basic_tools",
 }
 
 
@@ -72,7 +75,8 @@ def extract_assistant_text(message: Message) -> str:
         return content
     if isinstance(content, dict):
         text = content.get("text", "")
-        if text and not content.get("tool_calls"):
+        # 即使本轮包含工具调用，也打印模型给出的文本，便于观察每一步输出
+        if text:
             return text
     return ""
 
@@ -88,6 +92,7 @@ def register_selected_tools(tool_names: list[str]) -> None:
 async def run_once() -> str:
     execution_id = 17
     TopicBuildAgentContext.set_execution_id(execution_id)
+    merge_leve2 = '历史名人'
 
     base_dir = Path(__file__).parent
     output_dir = base_dir / "output"
@@ -96,11 +101,11 @@ async def run_once() -> str:
     setup_logging(level=LOG_LEVEL, file=LOG_FILE)
     register_selected_tools(ENABLED_TOOLS)
 
-    prompt = SimplePrompt(base_dir / "needs.md")
+    prompt = SimplePrompt(base_dir / "needs_from_element.md")
+
     model = resolve_model(prompt)
 
     run_config = copy.deepcopy(RUN_CONFIG)
-    run_config.model = model
     run_config.temperature = float(prompt.config.get("temperature", run_config.temperature))
     run_config.max_iterations = int(prompt.config.get("max_iterations", run_config.max_iterations))
     run_config.tools = ENABLED_TOOLS.copy()
@@ -113,7 +118,7 @@ async def run_once() -> str:
     run_config.knowledge.enable_injection = False
     run_config.trace_id = None
 
-    initial_messages = prompt.build_messages()
+    initial_messages = prompt.build_messages(merge_leve2=merge_leve2)
 
     store = FileSystemTraceStore(base_path=TRACE_STORE_PATH)
     runner = AgentRunner(
@@ -124,29 +129,43 @@ async def run_once() -> str:
     )
 
     final_text = ""
-    async for item in runner.run(messages=initial_messages, config=run_config):
-        if isinstance(item, Trace):
-            if item.status == "failed":
-                print(f"任务失败: {item.error_message}")
-        elif isinstance(item, Message):
-            text = extract_assistant_text(item)
-            if text:
-                final_text = text
-
-    if final_text:
-        output_file = output_dir / "result.txt"
-        with open(output_file, "w", encoding="utf-8") as f:
-            f.write(final_text)
-        print(f"结果已保存: {output_file}")
+    total_tokens = 0
+    total_cost = 0.0
+    has_completed_trace_cost = False
+    log_file_path = output_dir / f"run_log_{execution_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
+    with build_log(execution_id) as log_buffer:
+        async for item in runner.run(messages=initial_messages, config=run_config):
+            if isinstance(item, Trace):
+                if getattr(item, "status", None) == "completed":
+                    total_tokens = int(getattr(item, "total_tokens", 0) or 0)
+                    total_cost = float(getattr(item, "total_cost", 0.0) or 0.0)
+                    has_completed_trace_cost = True
+                continue
+            elif isinstance(item, Message):
+                text = extract_assistant_text(item)
+                if text:
+                    final_text = text
+                    log(f"[assistant] {text}")
+                if not has_completed_trace_cost:
+                    total_tokens += int(getattr(item, "total_tokens", 0) or 0)
+                    total_cost += float(getattr(item, "cost", 0.0) or 0.0)
+
+        if final_text:
+            output_file = output_dir / "result.txt"
+            with open(output_file, "w", encoding="utf-8") as f:
+                f.write(final_text)
+
+        log(f"[cost] total_tokens={total_tokens}, total_cost=${total_cost:.6f}")
+
+        full_log = log_buffer.getvalue()
+        with open(log_file_path, "w", encoding="utf-8") as f:
+            f.write(full_log)
 
     return final_text
 
 
 async def main() -> None:
-    final_text = await run_once()
-    if final_text:
-        print("\n=== Agent 响应 ===\n")
-        print(final_text)
+    await run_once()
 
 
 if __name__ == "__main__":

examples/piaoquan_needs/topic_build_pattern_tools.py

@@ -74,6 +74,9 @@ def get_category_tree(source_type: str = None) -> str:
 
     return _log_tool_output("get_category_tree", result)
 
+if __name__ == '__main__':
+    TopicBuildAgentContext.set_execution_id(17)
+    print(get_category_tree('实质'))
 
 # ============================================================================
 # 项集查询

examples/piaoquan_needs/weight_score_query_tools.py

@@ -0,0 +1,140 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+权重分查询工具
+
+从 examples/piaoquan_needs/data/{execution_id} 目录读取权重分 JSON，
+支持按元素/分类查询 TopN，以及按名称精确查询权重分。
+"""
+import json
+from pathlib import Path
+
+from agent import tool
+from examples.piaoquan_needs.topic_build_agent_context import TopicBuildAgentContext
+from examples.piaoquan_needs.topic_build_pattern_tools import _log_tool_input, _log_tool_output
+
+_VALID_LEVELS = {"元素", "分类"}
+_VALID_DIMENSIONS = {"实质", "形式", "意图"}
+
+
+def _get_weight_file_path(level: str, dimension: str) -> Path:
+    """根据参数构造权重数据文件路径。"""
+    execution_id = TopicBuildAgentContext.get_execution_id()
+    if execution_id is None:
+        raise ValueError("未设置 execution_id，请先在 TopicBuildAgentContext 中设置")
+
+    filename = f"{dimension}_{level}.json"
+    base_dir = Path(__file__).parent / "data" / str(execution_id)
+    return base_dir / filename
+
+
+def _load_weight_data(level: str, dimension: str) -> list[dict]:
+    """读取并返回权重数据列表。"""
+    file_path = _get_weight_file_path(level=level, dimension=dimension)
+    if not file_path.exists():
+        raise FileNotFoundError(f"权重数据文件不存在: {file_path}")
+
+    with file_path.open("r", encoding="utf-8") as f:
+        data = json.load(f)
+
+    if not isinstance(data, list):
+        raise ValueError(f"权重数据格式错误，期望 list，实际为: {type(data).__name__}")
+    return data
+
+
+def _validate_params(level: str, dimension: str):
+    """校验通用参数。"""
+    if level not in _VALID_LEVELS:
+        raise ValueError(f"level 参数非法: {level}，可选值: {sorted(_VALID_LEVELS)}")
+    if dimension not in _VALID_DIMENSIONS:
+        raise ValueError(f"dimension 参数非法: {dimension}，可选值: {sorted(_VALID_DIMENSIONS)}")
+
+
+@tool("查询元素或分类权重分 topN。参数：level(元素/分类)、dimension(实质/形式/意图)、top_n。")
+def get_weight_score_topn(level: str, dimension: str, top_n: int = 10) -> str:
+    """查询元素或分类权重分 topN。
+
+    Args:
+        level: 查询层级，元素 或 分类。
+        dimension: 查询维度，实质 / 形式 / 意图。
+        top_n: 返回数量，默认 10。
+
+    Returns:
+        JSON 字符串，包含查询参数、总量和 topN 数据。
+    """
+    execution_id = TopicBuildAgentContext.get_execution_id()
+    params = {
+        "execution_id": execution_id,
+        "level": level,
+        "dimension": dimension,
+        "top_n": top_n,
+    }
+    _log_tool_input("get_weight_score_topn", params)
+
+    try:
+        _validate_params(level=level, dimension=dimension)
+        if top_n <= 0:
+            return _log_tool_output("get_weight_score_topn", f"错误: top_n 必须大于 0，当前值: {top_n}")
+
+        data = _load_weight_data(level=level, dimension=dimension)
+        sorted_data = sorted(data, key=lambda x: float(x.get("score", 0)), reverse=True)
+        top_items = sorted_data[:top_n]
+
+        result = {
+            "level": level,
+            "dimension": dimension,
+            "top_n": top_n,
+            "total_count": len(data),
+            "items": top_items,
+        }
+        return _log_tool_output("get_weight_score_topn", json.dumps(result, ensure_ascii=False, indent=2))
+    except Exception as e:
+        return _log_tool_output("get_weight_score_topn", f"查询失败: {e}")
+
+
+@tool("查询指定名称的权重分。参数：level(元素/分类)、dimension(实质/形式/意图)、name。")
+def get_weight_score_by_name(level: str, dimension: str, name: str) -> str:
+    """查询指定名称的权重分。
+
+    Args:
+        level: 查询层级，元素 或 分类。
+        dimension: 查询维度，实质 / 形式 / 意图。
+        name: 要查询的名称（元素名或分类名）。
+
+    Returns:
+        JSON 字符串。若命中，返回匹配记录；否则返回未找到提示。
+    """
+    execution_id = TopicBuildAgentContext.get_execution_id()
+    params = {
+        "execution_id": execution_id,
+        "level": level,
+        "dimension": dimension,
+        "name": name,
+    }
+    _log_tool_input("get_weight_score_by_name", params)
+
+    try:
+        _validate_params(level=level, dimension=dimension)
+        if not name or not name.strip():
+            return _log_tool_output("get_weight_score_by_name", "错误: name 不能为空")
+
+        target_name = name.strip()
+        data = _load_weight_data(level=level, dimension=dimension)
+
+        if level == "元素":
+            key = "name"
+        else:
+            key = "category"
+
+        matched = [item for item in data if str(item.get(key, "")).strip() == target_name]
+
+        result = {
+            "level": level,
+            "dimension": dimension,
+            "name": target_name,
+            "matched_count": len(matched),
+            "items": matched,
+        }
+        return _log_tool_output("get_weight_score_by_name", json.dumps(result, ensure_ascii=False, indent=2))
+    except Exception as e:
+        return _log_tool_output("get_weight_score_by_name", f"查询失败: {e}")