|
|
@@ -1,7 +1,7 @@
|
|
|
"""
|
|
|
选题检索工具 - 根据关键词在数据库中匹配已有帖子的选题
|
|
|
|
|
|
-用于 Agent 执行时自主调取参考数据,并选择与当前人设最匹配的内容输出。
|
|
|
+用于 Agent 执行时自主调取参考数据。
|
|
|
"""
|
|
|
|
|
|
import json
|
|
|
@@ -41,57 +41,15 @@ async def _call_search_api(keywords: List[str]) -> Optional[List[Dict[str, Any]]
|
|
|
return []
|
|
|
|
|
|
|
|
|
-def _extract_text(obj: Any) -> str:
|
|
|
- """从结果对象中提取可比较的文本。"""
|
|
|
- if obj is None:
|
|
|
- return ""
|
|
|
- if isinstance(obj, str):
|
|
|
- return obj
|
|
|
- if isinstance(obj, dict):
|
|
|
- text_parts = []
|
|
|
- for k in ("title", "content", "主题", "选题", "描述", "description", "摘要"):
|
|
|
- v = obj.get(k)
|
|
|
- if v and isinstance(v, str):
|
|
|
- text_parts.append(v)
|
|
|
- if not text_parts:
|
|
|
- text_parts = [str(v) for v in obj.values() if isinstance(v, str)]
|
|
|
- return " ".join(text_parts)
|
|
|
- return str(obj)
|
|
|
-
|
|
|
-
|
|
|
-def _score_match(result: Dict[str, Any], persona_summary: str) -> float:
|
|
|
- """
|
|
|
- 计算单条结果与人设摘要的匹配度(简单关键词重叠)。
|
|
|
- 返回 0~1 之间的分数,越高表示越匹配。
|
|
|
- """
|
|
|
- if not persona_summary or not persona_summary.strip():
|
|
|
- return 1.0
|
|
|
-
|
|
|
- result_text = _extract_text(result).lower()
|
|
|
- persona_words = set(
|
|
|
- w for w in persona_summary.lower().replace(",", " ").replace(",", " ").split()
|
|
|
- if len(w) > 1
|
|
|
- )
|
|
|
- if not persona_words:
|
|
|
- return 1.0
|
|
|
-
|
|
|
- hits = sum(1 for w in persona_words if w in result_text)
|
|
|
- return hits / len(persona_words)
|
|
|
-
|
|
|
-
|
|
|
-def _pick_best_match(results: List[Dict[str, Any]], persona_summary: Optional[str]) -> Dict[str, Any]:
|
|
|
- """从结果中选出与人设最匹配的一条。"""
|
|
|
+def _pick_first(results: List[Dict[str, Any]]) -> Dict[str, Any]:
|
|
|
+ """从结果中取第一条。"""
|
|
|
if not results:
|
|
|
raise ValueError("无可用结果")
|
|
|
- if not persona_summary or len(results) == 1:
|
|
|
- return results[0]
|
|
|
-
|
|
|
- best = max(results, key=lambda r: _score_match(r, persona_summary))
|
|
|
- return best
|
|
|
+ return results[0]
|
|
|
|
|
|
|
|
|
@tool(
|
|
|
- description="根据关键词在数据库中检索已有帖子的选题,用于创作参考。最多返回5条,自动选择与当前人设最匹配的一条输出。",
|
|
|
+ description="根据关键词在数据库中检索已有帖子的选题,用于创作参考。最多返回5条,取第一条输出。",
|
|
|
display={
|
|
|
"zh": {
|
|
|
"name": "爆款选题检索",
|
|
|
@@ -101,19 +59,15 @@ def _pick_best_match(results: List[Dict[str, Any]], persona_summary: Optional[st
|
|
|
},
|
|
|
},
|
|
|
)
|
|
|
-async def topic_search(
|
|
|
- keywords: List[str],
|
|
|
- persona_summary: Optional[str] = None,
|
|
|
-) -> ToolResult:
|
|
|
+async def topic_search(keywords: List[str]) -> ToolResult:
|
|
|
"""
|
|
|
- 根据关键词检索数据库中已有帖子的选题,选择与人设最匹配的一条作为参考。
|
|
|
+ 根据关键词检索数据库中已有帖子的选题,取第一条作为参考。
|
|
|
|
|
|
Args:
|
|
|
keywords: 关键词列表,如 ["中老年健康养生", "爆款", "知识科普"]
|
|
|
- persona_summary: 当前人设摘要,用于从多条结果中筛选最匹配的(可选)
|
|
|
|
|
|
Returns:
|
|
|
- ToolResult: 最匹配的选题参考内容
|
|
|
+ ToolResult: 选题参考内容
|
|
|
"""
|
|
|
if not keywords:
|
|
|
return ToolResult(
|
|
|
@@ -138,7 +92,7 @@ async def topic_search(
|
|
|
)
|
|
|
|
|
|
try:
|
|
|
- best = _pick_best_match(results, persona_summary)
|
|
|
+ best = _pick_first(results)
|
|
|
except ValueError:
|
|
|
return ToolResult(
|
|
|
title="选题检索",
|
|
|
@@ -149,5 +103,5 @@ async def topic_search(
|
|
|
return ToolResult(
|
|
|
title="选题检索 - 参考数据",
|
|
|
output=output,
|
|
|
- long_term_memory=f"检索到与人设匹配的选题参考,关键词: {', '.join(keywords)}",
|
|
|
+ long_term_memory=f"检索到选题参考,关键词: {', '.join(keywords)}",
|
|
|
)
|