先说结论Agent 如果没有感知就像闭着眼睛做事 — 不知道外面在流行什么不知道用户关心什么只能凭空生成。感知是 Agent 获取外部信息的唯一通道设计好坏直接决定 Agent 的视野。在 self-media-agent 项目里感知模块就是hotspot/— 抓取多平台热点话题给选题生成提供弹药。但真实世界的爬虫随时会挂API 封了、网络超时、平台要登录。所以感知模块的核心不是怎么爬而是爬不到怎么办。项目的解法是三层降级策略真实爬取 → 预设数据 → LLM 动态生成。无论哪层挂了Agent 都不会失明。一、感知的 N 种来源Agent 的信息来源不止一种项目里用到了四种来源项目中的体现特点API 爬取hotspot/crawler.py抓微博热搜实时但不稳定可能被封用户输入CLI--topics 防晒,粉底手动选题最可靠但需要人工文件读取config/personas/*.yaml加载人设稳定但静态LLM 生成topic_gen.generate_topics()AI 选题灵活但有幻觉风险感知模块的职责是前两种— 从外部世界获取信息。后两种属于配置和规划在别的模块里。项目的感知架构hotspot/ ├── crawler.py ← 多平台调度器并发抓取 合并去重 ├── analyzer.py ← 热点分析四维评分 筛选排序 ├── schema.py ← 数据模型HotspotItem / HotspotResult └── platforms/ ├── xiaohongshu.py ← 小红书爬虫预设数据 LLM降级 ├── douyin.py ← 抖音爬虫预设数据 LLM降级 └── weibo.py ← 微博爬虫真实爬取 模拟降级 LLM降级二、三层降级策略爬不到也要有数据这是感知模块最核心的设计 —微博爬虫weibo.py展示了完整的三层降级class WeiboCrawler: async def crawl_niche(self, niche: str) - list[HotspotItem]: # 第1层尝试真实爬取微博热搜 items await self._try_real_crawl(niche) if items: return items # 第2层降级到模拟数据 logger.info([微博] 实际爬取失败使用模拟数据) return await self._mock_data(niche)第1层真实爬取async def _try_real_crawl(self, niche: str) - list[HotspotItem]: try: async with httpx.AsyncClient(timeout10.0) as client: resp await client.get( https://weibo.com/ajax/side/hotSearch, headers{User-Agent: Mozilla/5.0 ...}, ) if resp.status_code ! 200: return [] data resp.json() realtime data.get(data, {}).get(realtime, []) # 按赛道关键词过滤 keywords self._NICHE_KEYWORDS.get(niche, [niche]) items [] for entry in realtime: word entry.get(word, ) if keywords and not any(kw in word for kw in keywords): continue # 不匹配赛道关键词跳过 items.append(HotspotItem(titleword, platformweibo, ...)) return items[:20] except ImportError: return [] # httpx 未安装 except Exception as e: return [] # 网络异常真实爬取的三个失败点httpx 没装、请求被拦截、返回格式变了。每个都return []不崩交给上层降级。第2层预设模拟数据async def _mock_data(self, niche: str) - list[HotspotItem]: mock_items { beauty: [ {title: 夏季护肤误区盘点, heat_score: 88}, {title: 国货美妆品牌崛起, heat_score: 82}, ], 北漂: [ {title: 北漂租房中介避坑, heat_score: 89}, {title: 北京生活成本真实分享, heat_score: 82}, ], 劳务派遣: [ {title: 劳务派遣合同陷阱曝光, heat_score: 91}, {title: 派遣工维权成功案例, heat_score: 84}, ], } presets mock_items.get(niche) # 如果连预设数据都没有 → 第3层 if presets is None: presets await self._generate_niche_hotspots(niche) return [HotspotItem(...) for p in presets]第3层LLM 动态生成当赛道是自定义的预设数据里没有用 LLM 生成async def _generate_niche_hotspots(self, niche: str) - list[dict]: try: llm LLMClient(...) result await llm.generate( system_prompt你是一位自媒体热点分析师。根据赛道名称生成该赛道在微博上的热搜话题。\n 输出格式JSON数组..., user_promptf请为「{niche}」赛道生成3-5个微博热搜话题。, ) items json.loads(result) return items except Exception as e: logger.warning(fLLM 生成热点失败: {e}) # 最终兜底硬编码通用模板 return [ {title: f{niche}热门话题, heat_score: 70}, {title: f{niche}避坑经验, heat_score: 65}, ]三层降级的完整链路crawl_niche(niche) ↓ [第1层] _try_real_crawl() ← 真实爬微博API ↓ 失败被封/超时/没装httpx [第2层] _mock_data() ← 查预设数据字典 ↓ 没有自定义赛道 [第3层] _generate_niche_hotspots() ← LLM 动态生成 ↓ 失败API Key 没配 [兜底] 硬编码通用模板 ← {niche}热门话题无论怎么失败crawl_niche永远返回一个非空列表。Agent 永远不会失明。三、多平台并发抓取单个平台的数据量有限项目同时抓三个平台 —crawler.py用asyncio.gather并发class HotspotCrawler: def _ensure_platforms(self): from .platforms.xiaohongshu import XiaohongshuCrawler from .platforms.douyin import DouyinCrawler from .platforms.weibo import WeiboCrawler self._platforms { xiaohongshu: XiaohongshuCrawler(), douyin: DouyinCrawler(), weibo: WeiboCrawler(), } async def crawl(self, nichebeauty, platformsNone) - list[HotspotItem]: if platforms is None: platforms list(self._platforms.keys()) # 并发抓取所有平台 tasks [self.crawl_platform(p, niche) for p in platforms] results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果异常转成 HotspotResult.error hotspot_results [] for i, result in enumerate(results): if isinstance(result, Exception): hotspot_results.append(HotspotResult(platformplatforms[i], errorstr(result))) else: hotspot_results.append(result) # 合并去重 items self._merge_and_dedup(hotspot_results) return itemsreturn_exceptionsTrue— 某个平台抛异常不会让gather整体崩溃异常会作为结果返回。三个平台挂了两个剩一个的数据照样能用。合并去重staticmethod def _merge_and_dedup(results: list[HotspotResult]) - list[HotspotItem]: seen_titles: set[str] set() merged: list[HotspotItem] [] for result in results: if result.error: continue # 跳过失败的平台 for item in result.items: normalized item.title.strip().lower() if normalized in seen_titles: continue # 重复标题跳过 seen_titles.add(normalized) merged.append(item) merged.sort(keylambda x: x.heat_score, reverseTrue) # 按热度排序 return merged多平台抓取 → 合并 → 去重 → 按热度排序输出一个统一的热点列表下游不需要关心数据来自哪个平台。四、感知与赛道的耦合不同赛道需要不同的热点 — 美妆赛道要防晒测评职场赛道要面试技巧。项目用关键词映射把赛道和热点耦合起来。关键词映射analyzer.py和weibo.py都有关键词映射_NICHE_KEYWORDS: dict[str, list[str]] { beauty: [护肤, 美妆, 化妆, 口红, 防晒, 粉底, 底妆, 卸妆, 眼线, 成分], career: [职场, 面试, 工作, 跳槽, 副业, 简历, 薪资, 升职], 北漂: [北漂, 租房, 北京, 通勤, 医保, 合租, 中介], 劳务派遣: [劳务派遣, 派遣, 合同, 维权, 转正, 外包, 试用期], }微博真实爬取时用关键词过滤热搜keywords self._NICHE_KEYWORDS.get(niche, [niche]) for entry in realtime: word entry.get(word, ) if keywords and not any(kw in word for kw in keywords): continue # 不匹配赛道关键词跳过踩坑赛道劳务派遣抓到美妆热点早期版本预设数据的 fallback 写死成beauty赛道。当用户配了劳务派遣赛道但预设数据里没有这个赛道时代码 fallback 到美妆数据 — 结果劳务派遣的 Agent 抓到夏季防晒霜测评TOP10。根因# ❌ 早期fallback 写死 presets _PRESET_HOTSPOTS.get(niche) if presets is None: presets _PRESET_HOTSPOTS[beauty] # ← 写死 fallback 到美妆修复改为 LLM 动态生成用[niche]关键词降级# ✅ 修复自定义赛道用 LLM 生成 presets _PRESET_HOTSPOTS.get(niche) if presets is None: presets await self._generate_niche_hotspots(niche) # ← LLM 为该赛道生成教训fallback 永远不能写死成一个具体赛道否则跨赛道数据污染。五、热点分析与筛选抓到热点只是第一步还要分析和筛选—analyzer.py对每个热点做四维评分dataclass class TopicScore: topic: HotspotItem timeliness: float 0.5 # 时效性 relevance: float 0.5 # 赛道相关度 competition: float 0.5 # 竞争度 novelty: float 0.5 # 新颖度 total_score: float 0.0 def compute_total(self) - float: self.total_score ( self.timeliness * 0.3 self.relevance * 0.3 (1.0 - self.competition) * 0.2 # 竞争度越低越好 self.novelty * 0.2 ) return self.total_score四个维度维度怎么算权重含义时效性heat_score / 1000.3热度越高越时效赛道相关度标题/关键词与赛道匹配0.3越匹配越好竞争度heat_score / 100 * 0.80.2热度越高竞争越激烈取反新颖度与已有选题的差异度0.2越不同越好赛道相关度评分— 用关键词匹配staticmethod def _score_relevance(topic: HotspotItem, niche: str) - float: keywords _NICHE_KEYWORDS.get(niche, [niche]) title topic.title.lower() title_matches sum(1 for kw in keywords if kw in title) kw_matches sum(1 for kw in keywords if any(kw in tk for tk in topic_keywords)) total_matches title_matches kw_matches if total_matches 0: return 0.2 # 无匹配但保留不直接淘汰 return min(1.0, total_matches / max(len(keywords) * 0.3, 1))筛选排序— 按总分取 top_kasync def filter_and_rank(self, topics, niche, top_k20) - list[HotspotItem]: scored [] for topic in topics: score await self.score_topic(topic, niche, topics) scored.append(score) scored.sort(keylambda x: x.total_score, reverseTrue) return [s.topic for s in scored[:top_k]]感知的完整链路抓取crawler→ 分析评分analyzer→ 筛选排序 → 传入选题生成。踩坑总结坑根因修复爬虫被封/超时外部 API 不稳定三层降级真实爬取 → 预设数据 → LLM 生成自定义赛道无数据预设只覆盖常见赛道LLM 动态生成该赛道热点fallback 写死成 beauty早期偷懒改为[niche]关键词降级 LLM 生成劳务派遣抓到美妆热点fallback 数据污染修复 fallback不写死具体赛道httpx 未安装依赖缺失except ImportError: return []降级单平台挂掉影响全局串行抓取asyncio.gather(return_exceptionsTrue)并发容错热点与赛道不相关无过滤关键词映射 相关度评分经验总结感知模块的核心不是怎么爬而是爬不到怎么办— 三层降级确保 Agent 永不失明多平台并发 容错—asyncio.gather(return_exceptionsTrue)一个平台挂不影响其他fallback 永远不能写死成具体值— 否则跨赛道数据污染用 LLM 动态生成替代感知要和赛道耦合— 关键词映射 相关度评分确保抓到的热点和赛道匹配抓到热点只是开始— 还要分析评分、筛选排序才能变成有用的选题弹药下篇预告下一篇讲记忆模块Agent的短期/中期/长期记忆— 不是所有记忆都要放向量数据库简单场景用结构化存储就够了。三层记忆架构ChatSession短期→ style_preferences中期→ style_profile长期。
先说结论Agent 如果没有感知就像闭着眼睛做事 — 不知道外面在流行什么不知道用户关心什么只能凭空生成。感知是 Agent 获取外部信息的唯一通道设计好坏直接决定 Agent 的视野。在 self-media-agent 项目里感知模块就是hotspot/— 抓取多平台热点话题给选题生成提供弹药。但真实世界的爬虫随时会挂API 封了、网络超时、平台要登录。所以感知模块的核心不是怎么爬而是爬不到怎么办。项目的解法是三层降级策略真实爬取 → 预设数据 → LLM 动态生成。无论哪层挂了Agent 都不会失明。一、感知的 N 种来源Agent 的信息来源不止一种项目里用到了四种来源项目中的体现特点API 爬取hotspot/crawler.py抓微博热搜实时但不稳定可能被封用户输入CLI--topics 防晒,粉底手动选题最可靠但需要人工文件读取config/personas/*.yaml加载人设稳定但静态LLM 生成topic_gen.generate_topics()AI 选题灵活但有幻觉风险感知模块的职责是前两种— 从外部世界获取信息。后两种属于配置和规划在别的模块里。项目的感知架构hotspot/ ├── crawler.py ← 多平台调度器并发抓取 合并去重 ├── analyzer.py ← 热点分析四维评分 筛选排序 ├── schema.py ← 数据模型HotspotItem / HotspotResult └── platforms/ ├── xiaohongshu.py ← 小红书爬虫预设数据 LLM降级 ├── douyin.py ← 抖音爬虫预设数据 LLM降级 └── weibo.py ← 微博爬虫真实爬取 模拟降级 LLM降级二、三层降级策略爬不到也要有数据这是感知模块最核心的设计 —微博爬虫weibo.py展示了完整的三层降级class WeiboCrawler: async def crawl_niche(self, niche: str) - list[HotspotItem]: # 第1层尝试真实爬取微博热搜 items await self._try_real_crawl(niche) if items: return items # 第2层降级到模拟数据 logger.info([微博] 实际爬取失败使用模拟数据) return await self._mock_data(niche)第1层真实爬取async def _try_real_crawl(self, niche: str) - list[HotspotItem]: try: async with httpx.AsyncClient(timeout10.0) as client: resp await client.get( https://weibo.com/ajax/side/hotSearch, headers{User-Agent: Mozilla/5.0 ...}, ) if resp.status_code ! 200: return [] data resp.json() realtime data.get(data, {}).get(realtime, []) # 按赛道关键词过滤 keywords self._NICHE_KEYWORDS.get(niche, [niche]) items [] for entry in realtime: word entry.get(word, ) if keywords and not any(kw in word for kw in keywords): continue # 不匹配赛道关键词跳过 items.append(HotspotItem(titleword, platformweibo, ...)) return items[:20] except ImportError: return [] # httpx 未安装 except Exception as e: return [] # 网络异常真实爬取的三个失败点httpx 没装、请求被拦截、返回格式变了。每个都return []不崩交给上层降级。第2层预设模拟数据async def _mock_data(self, niche: str) - list[HotspotItem]: mock_items { beauty: [ {title: 夏季护肤误区盘点, heat_score: 88}, {title: 国货美妆品牌崛起, heat_score: 82}, ], 北漂: [ {title: 北漂租房中介避坑, heat_score: 89}, {title: 北京生活成本真实分享, heat_score: 82}, ], 劳务派遣: [ {title: 劳务派遣合同陷阱曝光, heat_score: 91}, {title: 派遣工维权成功案例, heat_score: 84}, ], } presets mock_items.get(niche) # 如果连预设数据都没有 → 第3层 if presets is None: presets await self._generate_niche_hotspots(niche) return [HotspotItem(...) for p in presets]第3层LLM 动态生成当赛道是自定义的预设数据里没有用 LLM 生成async def _generate_niche_hotspots(self, niche: str) - list[dict]: try: llm LLMClient(...) result await llm.generate( system_prompt你是一位自媒体热点分析师。根据赛道名称生成该赛道在微博上的热搜话题。\n 输出格式JSON数组..., user_promptf请为「{niche}」赛道生成3-5个微博热搜话题。, ) items json.loads(result) return items except Exception as e: logger.warning(fLLM 生成热点失败: {e}) # 最终兜底硬编码通用模板 return [ {title: f{niche}热门话题, heat_score: 70}, {title: f{niche}避坑经验, heat_score: 65}, ]三层降级的完整链路crawl_niche(niche) ↓ [第1层] _try_real_crawl() ← 真实爬微博API ↓ 失败被封/超时/没装httpx [第2层] _mock_data() ← 查预设数据字典 ↓ 没有自定义赛道 [第3层] _generate_niche_hotspots() ← LLM 动态生成 ↓ 失败API Key 没配 [兜底] 硬编码通用模板 ← {niche}热门话题无论怎么失败crawl_niche永远返回一个非空列表。Agent 永远不会失明。三、多平台并发抓取单个平台的数据量有限项目同时抓三个平台 —crawler.py用asyncio.gather并发class HotspotCrawler: def _ensure_platforms(self): from .platforms.xiaohongshu import XiaohongshuCrawler from .platforms.douyin import DouyinCrawler from .platforms.weibo import WeiboCrawler self._platforms { xiaohongshu: XiaohongshuCrawler(), douyin: DouyinCrawler(), weibo: WeiboCrawler(), } async def crawl(self, nichebeauty, platformsNone) - list[HotspotItem]: if platforms is None: platforms list(self._platforms.keys()) # 并发抓取所有平台 tasks [self.crawl_platform(p, niche) for p in platforms] results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果异常转成 HotspotResult.error hotspot_results [] for i, result in enumerate(results): if isinstance(result, Exception): hotspot_results.append(HotspotResult(platformplatforms[i], errorstr(result))) else: hotspot_results.append(result) # 合并去重 items self._merge_and_dedup(hotspot_results) return itemsreturn_exceptionsTrue— 某个平台抛异常不会让gather整体崩溃异常会作为结果返回。三个平台挂了两个剩一个的数据照样能用。合并去重staticmethod def _merge_and_dedup(results: list[HotspotResult]) - list[HotspotItem]: seen_titles: set[str] set() merged: list[HotspotItem] [] for result in results: if result.error: continue # 跳过失败的平台 for item in result.items: normalized item.title.strip().lower() if normalized in seen_titles: continue # 重复标题跳过 seen_titles.add(normalized) merged.append(item) merged.sort(keylambda x: x.heat_score, reverseTrue) # 按热度排序 return merged多平台抓取 → 合并 → 去重 → 按热度排序输出一个统一的热点列表下游不需要关心数据来自哪个平台。四、感知与赛道的耦合不同赛道需要不同的热点 — 美妆赛道要防晒测评职场赛道要面试技巧。项目用关键词映射把赛道和热点耦合起来。关键词映射analyzer.py和weibo.py都有关键词映射_NICHE_KEYWORDS: dict[str, list[str]] { beauty: [护肤, 美妆, 化妆, 口红, 防晒, 粉底, 底妆, 卸妆, 眼线, 成分], career: [职场, 面试, 工作, 跳槽, 副业, 简历, 薪资, 升职], 北漂: [北漂, 租房, 北京, 通勤, 医保, 合租, 中介], 劳务派遣: [劳务派遣, 派遣, 合同, 维权, 转正, 外包, 试用期], }微博真实爬取时用关键词过滤热搜keywords self._NICHE_KEYWORDS.get(niche, [niche]) for entry in realtime: word entry.get(word, ) if keywords and not any(kw in word for kw in keywords): continue # 不匹配赛道关键词跳过踩坑赛道劳务派遣抓到美妆热点早期版本预设数据的 fallback 写死成beauty赛道。当用户配了劳务派遣赛道但预设数据里没有这个赛道时代码 fallback 到美妆数据 — 结果劳务派遣的 Agent 抓到夏季防晒霜测评TOP10。根因# ❌ 早期fallback 写死 presets _PRESET_HOTSPOTS.get(niche) if presets is None: presets _PRESET_HOTSPOTS[beauty] # ← 写死 fallback 到美妆修复改为 LLM 动态生成用[niche]关键词降级# ✅ 修复自定义赛道用 LLM 生成 presets _PRESET_HOTSPOTS.get(niche) if presets is None: presets await self._generate_niche_hotspots(niche) # ← LLM 为该赛道生成教训fallback 永远不能写死成一个具体赛道否则跨赛道数据污染。五、热点分析与筛选抓到热点只是第一步还要分析和筛选—analyzer.py对每个热点做四维评分dataclass class TopicScore: topic: HotspotItem timeliness: float 0.5 # 时效性 relevance: float 0.5 # 赛道相关度 competition: float 0.5 # 竞争度 novelty: float 0.5 # 新颖度 total_score: float 0.0 def compute_total(self) - float: self.total_score ( self.timeliness * 0.3 self.relevance * 0.3 (1.0 - self.competition) * 0.2 # 竞争度越低越好 self.novelty * 0.2 ) return self.total_score四个维度维度怎么算权重含义时效性heat_score / 1000.3热度越高越时效赛道相关度标题/关键词与赛道匹配0.3越匹配越好竞争度heat_score / 100 * 0.80.2热度越高竞争越激烈取反新颖度与已有选题的差异度0.2越不同越好赛道相关度评分— 用关键词匹配staticmethod def _score_relevance(topic: HotspotItem, niche: str) - float: keywords _NICHE_KEYWORDS.get(niche, [niche]) title topic.title.lower() title_matches sum(1 for kw in keywords if kw in title) kw_matches sum(1 for kw in keywords if any(kw in tk for tk in topic_keywords)) total_matches title_matches kw_matches if total_matches 0: return 0.2 # 无匹配但保留不直接淘汰 return min(1.0, total_matches / max(len(keywords) * 0.3, 1))筛选排序— 按总分取 top_kasync def filter_and_rank(self, topics, niche, top_k20) - list[HotspotItem]: scored [] for topic in topics: score await self.score_topic(topic, niche, topics) scored.append(score) scored.sort(keylambda x: x.total_score, reverseTrue) return [s.topic for s in scored[:top_k]]感知的完整链路抓取crawler→ 分析评分analyzer→ 筛选排序 → 传入选题生成。踩坑总结坑根因修复爬虫被封/超时外部 API 不稳定三层降级真实爬取 → 预设数据 → LLM 生成自定义赛道无数据预设只覆盖常见赛道LLM 动态生成该赛道热点fallback 写死成 beauty早期偷懒改为[niche]关键词降级 LLM 生成劳务派遣抓到美妆热点fallback 数据污染修复 fallback不写死具体赛道httpx 未安装依赖缺失except ImportError: return []降级单平台挂掉影响全局串行抓取asyncio.gather(return_exceptionsTrue)并发容错热点与赛道不相关无过滤关键词映射 相关度评分经验总结感知模块的核心不是怎么爬而是爬不到怎么办— 三层降级确保 Agent 永不失明多平台并发 容错—asyncio.gather(return_exceptionsTrue)一个平台挂不影响其他fallback 永远不能写死成具体值— 否则跨赛道数据污染用 LLM 动态生成替代感知要和赛道耦合— 关键词映射 相关度评分确保抓到的热点和赛道匹配抓到热点只是开始— 还要分析评分、筛选排序才能变成有用的选题弹药下篇预告下一篇讲记忆模块Agent的短期/中期/长期记忆— 不是所有记忆都要放向量数据库简单场景用结构化存储就够了。三层记忆架构ChatSession短期→ style_preferences中期→ style_profile长期。
操作全解析与运维实践)
