先说结论

Agent 如果没有感知，就像闭着眼睛做事 — 不知道外面在流行什么，不知道用户关心什么，只能凭空生成。

感知是 Agent 获取外部信息的唯一通道，设计好坏直接决定 Agent 的"视野"。

在 self-media-agent 项目里，感知模块就是hotspot/— 抓取多平台热点话题，给选题生成提供"弹药"。但真实世界的爬虫随时会挂：API 封了、网络超时、平台要登录。所以感知模块的核心不是"怎么爬"，而是"爬不到怎么办"。

项目的解法是三层降级策略：真实爬取 → 预设数据 → LLM 动态生成。无论哪层挂了，Agent 都不会"失明"。

一、感知的 N 种来源

Agent 的信息来源不止一种，项目里用到了四种：

感知模块的职责是前两种— 从外部世界获取信息。后两种属于配置和规划，在别的模块里。

项目的感知架构：

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(timeout=10.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(title=word, platform="weibo", ...)) ​ 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_prompt=f"请为「{niche}」赛道生成3-5个微博热搜话题。", ) items = json.loads(result) return items except Exception as e: logger.warning(f"LLM 生成热点失败: {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, niche="beauty", platforms=None) -> 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_exceptions=True) ​ # 处理结果：异常转成 HotspotResult.error hotspot_results = [] for i, result in enumerate(results): if isinstance(result, Exception): hotspot_results.append(HotspotResult(platform=platforms[i], error=str(result))) else: hotspot_results.append(result) ​ # 合并去重 items = self._merge_and_dedup(hotspot_results) return items

return_exceptions=True— 某个平台抛异常不会让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(key=lambda x: x.heat_score, reverse=True) # 按热度排序 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

四个维度：

赛道相关度评分— 用关键词匹配：

@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_k：

async def filter_and_rank(self, topics, niche, top_k=20) -> list[HotspotItem]: scored = [] for topic in topics: score = await self.score_topic(topic, niche, topics) scored.append(score) scored.sort(key=lambda x: x.total_score, reverse=True) return [s.topic for s in scored[:top_k]]

感知的完整链路：抓取（crawler）→ 分析评分（analyzer）→ 筛选排序 → 传入选题生成。

踩坑总结

经验总结

1. 感知模块的核心不是"怎么爬"，而是"爬不到怎么办"— 三层降级确保 Agent 永不"失明"

2. 多平台并发 + 容错—asyncio.gather(return_exceptions=True)，一个平台挂不影响其他

3. fallback 永远不能写死成具体值— 否则跨赛道数据污染，用 LLM 动态生成替代

4. 感知要和赛道耦合— 关键词映射 + 相关度评分，确保抓到的热点和赛道匹配

5. 抓到热点只是开始— 还要分析评分、筛选排序，才能变成有用的选题弹药

下篇预告

下一篇讲记忆模块：Agent的短期/中期/长期记忆— 不是所有记忆都要放向量数据库，简单场景用结构化存储就够了。三层记忆架构：ChatSession（短期）→ style_preferences（中期）→ style_profile（长期）。