fix

examples/content_finder/content_finder.md

@@ -38,7 +38,7 @@ $system$
 - 核心指标：分享率、DAU
 
 ## 执行流程（按顺序，禁止跳步）
-1. **需求理解阶段**: 按 `demand_analysis` 执行,输出的内容用于后续的流程
+1. **需求理解阶段**: 按 `demand_analysis` 执行
 2. **内容寻找**：按 `content_finding_strategy` 执行
 3. **筛选阶段**：按 `content_filtering_strategy` 执行
 4. **优质账号扩展**: 对于筛选阶段获取到用户画像的优质作者，按`high_quality_analysis`执行
@@ -49,9 +49,36 @@ $system$
 
 ## 强制要求（违反即为错误）
 
-### 搜索词来源
+### 需求理解阶段
 1. 禁止使用特征作为搜索词。
-2. 搜索词必须来自于 `需求理解阶段`的输出。
+2. 必须按照`demand_analysis`里的**执行步骤**执行，**先做特征分层归类**,再**根据步骤1的归类选择策略**，此步骤严禁大模型联想输出。
+3. **特征分层归类**每个层都是对输入特征的筛选和组合归类，必须使用原词，不能联想新词。层级之间特征可以重复，不是必须只划归到一层。
+4. 当实质特征不为空时，上层特征和下层特征不能都为空，注意检查。
+3. 此阶段必须输出下面的结构
+```json
+{
+  "特征归类": {
+    "实质特征": ["特征词1", "特征词2"],
+    "形式特征": [],
+    "下层特征": [],
+    "上层特征": []
+  },
+  "起点策略": {
+    "case出发搜索词": [],
+    "特征出发搜索词": [],
+    "是否调用高赞case工具": true,
+    "高赞case_灵感点": [],
+    "高赞case_目的点": [],
+    "高赞case_关键点": []
+  },
+  "筛选方案": {
+    "形式规则": [],
+    "目的点对齐规则": [],
+    "关键点打分说明": [],
+    "淘汰规则": []
+  }
+}
+```
 
 ### 画像工具必须调用
 对每条候选内容，**必须**按以下顺序获取画像：

examples/content_finder/skills/demand_analysis.md

@@ -4,77 +4,40 @@ description: 需求分析
 ---
 
 # 需求分析方法论
+实际输入通常是一串**逗号分隔的特征词**（例如：`"养老,防骗,口播,三段式"`）
 
-## 目标
-
-将需求特征拆解为两类可执行路径：
-- **寻找路径**：决定去哪找、用什么词找、走即时还是积累通道
-- **判别路径**：决定候选内容是否符合目标与风格
-
----
-
-## 一、特征分层定义
-
-### 1) 实质特征 vs 形式特征
-
-- **实质特征（主要特征）**：内容“讲什么”，决定寻找方向。  
-  示例：养老、防骗、慢病管理、家庭矛盾。
-- **形式特征（次要特征）**：内容“怎么讲”，决定判别标准。  
-  示例：口播/剧情、强冲突开场、三段式结构、时长节奏。
-
-规则：**实质用于寻找，形式用于判别**。
-
-### 2) 下层特征（具象） vs 上层特征（泛化）
-
-- **下层特征（具象）**：可直接形成词或选题点。  
-  示例：`退休金被骗怎么办`、`高血压晨起注意事项`。
-- **上层特征（泛化）**：抽象类目，覆盖范围大。  
-  示例：`健康`、`养老`、`家庭关系`。
-
-规则：下层偏精准命中，上层偏覆盖扩展。
-
----
-
-## 二、双起点策略（高赞case出发 / 特征出发）
-
-### 0) 先做“可执行的特征分层”（把输入变成决策依据）
-
-实际输入通常是一串**逗号分隔的特征词**（例如：`"养老,防骗,口播,三段式"`）。在选择策略前，必须先完成两级划分：
-
-#### 第一级：实质 vs 形式
-
-- **实质特征**：内容讲什么（决定“去哪找/用什么词找”）
-- **形式特征**：内容怎么讲（决定“怎么判别/怎么排序”）
-
+## 执行步骤
+### 步骤1.先做特征分层归类(**不要产生新词，仅针对输入的特征词进行筛选和归类**)
+步骤1.1 区分实质特征和形式特征
+- **实质特征**：描述具体实质的特征词
+- **形式特征**：描述表现形式的特征词
 输出两组：
-- `substance_terms`: `[...]`
-- `form_terms`: `[...]`
-
-#### 第二级：上层 vs 下层（仅对“实质特征”再分层）
-
-- **上层（泛化）**：主题类目（覆盖扩展用），如：健康、养老、家庭关系
-- **下层（具象）**：可直接形成搜索词/选题点（精准命中用），如：退休金被骗怎么办、高血压晨起注意事项
+- `实质特征`: `[...]`
+- `形式特征`: `[...]`
 
+步骤1.2 对**步骤1.1的所有实质特征**继续细分为
+- **上层特征**：宽泛的，不能用于搜索的实质特征
+- **下层特征**：具体的，可直接搜索的实质特征，如：退休金被骗套路、高血压晨起注意事项
 输出两组：
-- `upper_substance_terms`: `[...]`
-- `lower_substance_terms`: `[...]`
+- `上层特征`: `[...]`
+- `下层特征`: `[...]`
 
-> 重要：**形式特征不参与“上层/下层”分层**，它们只进入后续的判别规则（如表达结构、节奏、可分享程度）。
+**注意：上层特征和下层特征都取自实质特征列表。且上层特征和下层特征的并集 = 实质特征**。
 
-### 0.1) 策略选择规则（可并行执行）
+> 重要：**形式特征不参与“上层/下层”分层**，它们只进入后续的判别规则（如表达结构、节奏、可分享程度）。
 
+### 步骤2. 根据步骤1的归类选择策略
 根据上述分层结果决定要执行哪些起点策略：
-
-- **当 `lower_substance_terms` 非空**：执行 **A. 高赞case出发**（用 `get_video_topic` 拉“灵感点/目的点/关键点”）
-- **当 `upper_substance_terms` 非空**：执行 **B. 特征出发**（构建主题树 → 下钻出可执行词）
+- **当 `下层特征` 非空**：必须执行 **A. 高赞视频选题点提取**（用 `get_video_topic` 拉“灵感点/目的点/关键点”）
+- **当 `上层特征` 非空**：执行 **B. 特征出发**（构建主题树 → 下钻出可执行词）
 - **两者都非空**：A + B **都执行**，最后合并去重
 - **只有形式词/实质词为空**：承认信息不足，只能先按用户原话/补充提问（或用最少假设）生成候选搜索词包；不要编造“高赞case选题点”
 
-### A. 高赞case出发（优先用于下层特征）
+### A. 高赞视频选题点提取
 
-适用：需求里已有具象表达，或需要从案例中补全/完善搜索词。  
+适用：`下层特征` 非空必须执行此步骤。  
 动作：
-1. 输入：使用 `lower_substance_terms`（或用户给的具象词）拼成 `features`（逗号分隔字符串）调用 **`get_video_topic`**
+1. 输入：使用 `下层特征`拼成 `features`（逗号分隔字符串）调用 **`get_video_topic`**
 2. 将工具返回 `metadata.videos` 内的选题点按用途拆分：
    - `灵感点` -> 用于构建**搜索词包**（写入寻找清单的候选词）
    - `目的点` -> 用于构建**判别目标**（写入判别清单的“该对齐什么”）
@@ -89,7 +52,7 @@ description: 需求分析
 
 适用：需求偏抽象，先建立主题覆盖框架。  
 动作：
-1. 输入：使用 `upper_substance_terms` 作为主题根
+1. 输入：使用 `上层特征` 作为主题根
 2. 以上层特征构建主题树（主题 -> 子主题 -> 关键词）
 3. 将树上的子主题/关键词**下钻成可执行搜索词**（落到能直接丢给 `douyin_search` 的词）
 4. 结合库内优质作者特征做扩展（可选：`find_authors_from_db` → `douyin_user_videos`）
@@ -97,17 +60,6 @@ description: 需求分析
 
 > 两条起点可并行，不互斥；最后合并去重。
 
----
-
-## 三、完整执行流程
-
-1. **拆需求**：标注每个特征为`实质/形式` + `上层/下层`
-2. **选起点**：
-   - 下层优先走`case出发`
-   - 上层优先走`特征出发`
-3. **生成两份清单**：
-   - `寻找清单`：词包、来源、通道、优先级
-   - `判别清单`：形式规则、打分点、淘汰条件
 
 ## 四、限制
 输出**必须**基于`get_video_topic`返回的选题信息生成，**严谨模型自己联想**生成
@@ -118,25 +70,25 @@ description: 需求分析
 
 ```json
 {
-  "feature_parse": {
-    "substance_main_features": [],
-    "form_secondary_features": [],
-    "lower_specific_features": [],
-    "upper_generic_features": []
+  "特征归类": {
+    "实质特征": [],
+    "形式特征": [],
+    "下层特征": [],
+    "上层特征": []
   },
-  "entry_strategy": {
-    "case_entry_terms": [],
-    "feature_entry_terms": [],
-    "goodcase_tool_used": true,
-    "goodcase_inspiration_points": [],
-    "goodcase_goal_points": [],
-    "goodcase_key_points": []
+  "起点策略": {
+    "case出发搜索词": [],
+    "特征出发搜索词": [],
+    "是否调用高赞case工具": true,
+    "高赞case_灵感点": [],
+    "高赞case_目的点": [],
+    "高赞case_关键点": []
   },
-  "filter_plan": {
-    "form_rules": [],
-    "goal_alignment_rules": [],
-    "key_point_scoring_notes": [],
-    "reject_rules": []
+  "筛选方案": {
+    "形式规则": [],
+    "目的点对齐规则": [],
+    "关键点打分说明": [],
+    "淘汰规则": []
   }
 }
 ```

examples/content_finder/tools/get_video_topic.py

@@ -13,6 +13,8 @@ from typing import Any, Dict, Iterable, List, Optional, Sequence, Set
 
 from agent.tools import ToolResult, tool
 
+import pymysql
+
 from db import get_open_aigc_pattern_connection, get_connection
 
 JsonDict = Dict[str, Any]
@@ -85,7 +87,7 @@ def _query_post_ids_by_feature(conn, feature: str) -> Set[str]:
 
     sql = f"""
     SELECT DISTINCT post_id
-    FROM element_classification_mapping
+    FROM topic_pattern_element
     WHERE name = %s
     """
     with conn.cursor() as cur:
@@ -110,6 +112,27 @@ def _chunked(items: Sequence[str], size: int) -> Iterable[Sequence[str]]:
         yield items[i : i + size]
 
 
+def _coerce_limit(value: Any, default: int = 20) -> int:
+    if value is None:
+        return default
+    if isinstance(value, bool):
+        return default
+    if isinstance(value, int):
+        return value
+    if isinstance(value, str):
+        s = value.strip()
+        if not s:
+            return default
+        try:
+            return int(s)
+        except Exception:
+            return default
+    try:
+        return int(value)
+    except Exception:
+        return default
+
+
 def _query_points_by_post_ids(conn, post_ids: List[str]) -> Dict[str, JsonDict]:
     """
     从 workflow_decode_task_result 取三类 points，按 post_id 映射。
@@ -172,11 +195,12 @@ async def get_video_topic(
     """
 
     feature_list = _split_features(features)
+    limit_int = _coerce_limit(limit, default=20)
     if not feature_list:
         return ToolResult(
             title="选题解构",
             output="features 为空：返回空视频列表（videos=0）。",
-            metadata={"videos": [], "features": [], "limit": limit, "post_ids": []},
+            metadata={"videos": [], "features": [], "limit": limit_int, "post_ids": []},
         )
 
     open_aigc_conn = get_open_aigc_pattern_connection()
@@ -187,8 +211,8 @@ async def get_video_topic(
             feature_sets.append(_query_post_ids_by_feature(open_aigc_conn, f))
 
         post_ids = _intersect_post_ids(feature_sets)
-        if limit and limit > 0:
-            post_ids = post_ids[: int(limit)]
+        if limit_int > 0:
+            post_ids = post_ids[:limit_int]
 
         points_map = _query_points_by_post_ids(supply_conn, post_ids)
 
@@ -214,11 +238,12 @@ async def get_video_topic(
             metadata={
                 "videos": videos,
                 "features": feature_list,
-                "limit": limit,
+                "limit": limit_int,
                 # 调试/可追溯：不放在 videos 条目里，避免污染“每条视频字段约定”
                 "post_ids": post_ids,
             },
             long_term_memory=f"Get video topic points by features: {','.join(feature_list)} (videos={len(videos)})",
         )
     finally:
-        conn.close()
+        supply_conn.close()
+        open_aigc_conn.close()