From a26f0e936e01519dc5038f3b8e77af661cdc9561 Mon Sep 17 00:00:00 2001 From: Yuzhiran Dev Date: Thu, 21 May 2026 08:06:24 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E9=87=87=E9=9B=86=E5=99=A8?= =?UTF-8?q?=E6=B5=81=E7=A8=8B=E9=A1=BA=E5=BA=8F:=20=E9=87=87=E9=9B=86?= =?UTF-8?q?=E2=86=92=E5=88=86=E6=9E=90=E2=86=92LLM=E6=80=BB=E7=BB=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _generate_topic_with_llm 改名为 _generate_topics_with_llm - 支持同时接收已分析的案例+搜索结果作为LLM上下文 - run()顺序改为: 采集→RSS提炼案例→LLM基于全部数据生成→降级回退 - LLM现在在流程末尾做总结生成,而非开头替代搜索 --- scripts/collector.py | 64 +++++++++++++++++++++----------------------- 1 file changed, 31 insertions(+), 33 deletions(-) diff --git a/scripts/collector.py b/scripts/collector.py index d18c392..7a0d5ac 100644 --- a/scripts/collector.py +++ b/scripts/collector.py @@ -377,13 +377,13 @@ class SustainabilityCollector: logger.warning(f"web_search失败 {source.name}: {e}") return [] - def _generate_topic_with_llm(self, search_results: Optional[List[Dict]] = None) -> Optional[SustainabilityTopic]: - """用LLM生成选题（有搜索结果时参考，无结果时直接生成）""" + def _generate_topics_with_llm(self, cases: List[SustainabilityCase] = None, search_results: List[Dict] = None) -> List[SustainabilityTopic]: + """用LLM基于采集数据生成选题（数据充分时精确生成，无数据时凭知识生成）""" try: from app.core.nvidia_client import call_llm except ImportError: logger.warning("LLM不可用，跳过AI选题生成") - return None + return [] existing = self._get_existing_titles() existing_hint = "" @@ -394,30 +394,28 @@ class SustainabilityCollector: day_idx = datetime.datetime.now().timetuple().tm_yday % len(categories) target_category = categories[day_idx] - search_section = "" + data_section = "" if search_results: - summaries = [f"- {r.get('title','')}: {r.get('content','')[:120]}" for r in search_results[:4]] - search_section = "搜索结果参考：\n" + "\n".join(summaries) + "\n" + summaries = [f"- {r.get('title','')}: {r.get('content','')[:100]}" for r in search_results[:5]] + data_section += "搜索结果：\n" + "\n".join(summaries) + "\n" + if cases: + case_lines = [f"- {c.title[:40]}({c.category})" for c in cases[:5]] + data_section += "\n采集案例：\n" + "\n".join(case_lines) + "\n" - prompt = f"""你是一个内容策略师。生成一个面向中国年轻读者、有价值、适合传播的选题。 + prompt = f"""你是一个内容策略师。基于以下信息，为「{target_category}」类别生成一个高质量选题。 -目标类别：{target_category} -{search_section} +{data_section if data_section else "（当前无实时采集数据，请基于你对中文互联网趋势的了解直接生成）"} {existing_hint} -生成一个选题，直接输出JSON（不要其他文字）： +输出一个选题，格式JSON： {{{{ - "title": "标题（20字内，含核心关键词，避免「新趋势」「指南」这类烂尾词）", - "core_concept": "核心观点（一句话说清独特价值）", - "audience_pain": "受众痛点（真实用户的困惑）", + "title": "标题（20字内，含核心关键词）", + "core_concept": "核心观点（一句话）", + "audience_pain": "受众痛点", "unique_angle": "差异化切入点", "format": "内容形式（趋势洞察/实操指南/对比分析/案例解读）" }}}} - -要求： -- 标题像普通人会搜索的 -- 切入点具体，不泛泛而谈 -- 优先考虑中国读者能实操的内容""" +只输出JSON。""" try: resp = call_llm(prompt, temperature=0.7) @@ -431,7 +429,7 @@ class SustainabilityCollector: topic = SustainabilityTopic( id=topic_id, title=data.get("title", f"{target_category}新观察"), - cases=[], + cases=[c.id for c in (cases or [])[:3]], audience="城市焦虑青年(26-35岁)", china_pain_points=data.get("audience_pain", ""), localization_solution="文章中将提供具体可执行的建议", @@ -456,10 +454,10 @@ class SustainabilityCollector: platform_urls={} ) logger.info(f"LLM生成选题: {topic.title}") - return topic + return [topic] except Exception as e: logger.warning(f"LLM选题生成失败: {e}") - return None + return [] def analyze_article(self, article: Dict) -> Optional[SustainabilityCase]: """分析文章内容,提炼案例""" @@ -786,24 +784,24 @@ class SustainabilityCollector: logger.info(f"RSS采集 {sum(1 for a in all_articles if a.get('source_name','') not in [s.name for s in self.sources if s.type=='web_search'])} 篇, " f"搜索采集 {len(web_search_results)} 篇") - # ---------------------- 第二阶段：LLM选题生成 ---------------------- - llm_topic = self._generate_topic_with_llm(web_search_results if web_search_results else None) - - if llm_topic and not self._is_duplicate_topic(llm_topic.title, existing_titles): - llm_topic.created_at = datetime.datetime.now().isoformat() - llm_topic.lock_by = None - llm_topic.lock_at = None - llm_topic.status = "待处理" - self.new_topics.append(llm_topic) - logger.info(f"✅ LLM生成选题: {llm_topic.title}") - - # ---------------------- 第三阶段：RSS文章提炼案例 ---------------------- + # ---------------------- 第二阶段：RSS文章提炼案例 ---------------------- rss_articles = [a for a in all_articles if a not in web_search_results] for article in rss_articles[:15]: case = self.analyze_article(article) if case: self.new_cases.append(case) + # ---------------------- 第三阶段：LLM基于采集数据生成选题 ---------------------- + llm_topics = self._generate_topics_with_llm(cases=self.new_cases, search_results=web_search_results) + for topic in llm_topics: + if not self._is_duplicate_topic(topic.title, existing_titles): + topic.created_at = datetime.datetime.now().isoformat() + topic.lock_by = None + topic.lock_at = None + topic.status = "待处理" + self.new_topics.append(topic) + logger.info(f"✅ LLM生成选题: {topic.title}") + # ---------------------- 第四阶段：降级策略 ---------------------- if not self.new_topics and len(self.new_cases) < 2: logger.warning(f"LLM选题和RSS案例均不足,启动本地案例降级")