Hermes + TrendRadar 集成方案设计

方案概述

将 TrendRadar 作为 Hermes 的智能信息感知层，为 Hermes Agent 提供实时热点数据、舆情监控和趋势分析能力。

┌─────────────────────────────────────────────────────────────────┐

│                    集成架构概览                                  │ ├─────────────────────────────────────────────────────────────────┤ │                                                                 │ │   ┌──────────────────┐         ┌──────────────────┐            │ │   │   TrendRadar     │         │     Hermes       │            │ │   │                  │         │                  │            │ │   │ • 热点聚合       │◄───────►│ • Agent 执行     │            │ │   │ • AI 筛选        │   API   │ • Skill 系统     │            │ │   │ • 舆情分析       │         │ • 工作流编排     │            │ │   │ • 多渠道推送     │         │ • 自主改进       │            │ │   └──────────────────┘         └──────────────────┘            │ │            │                            │                      │ │            │                            │                      │ │            └────────────┬───────────────┘                      │ │                         ↓                                      │ │              ┌────────────────────┐                           │ │              │   统一数据层        │                           │ │              │   - 热点数据库      │                           │ │              │   - 分析结果缓存    │                           │ │              │   - 推送队列        │                           │ │              └────────────────────┘                           │ │                                                                 │ └─────────────────────────────────────────────────────────────────┘

集成模式设计

模式 1: TrendRadar 作为 Hermes 数据源 (推荐)

TrendRadar 监控热点 ──► 数据存储 ──► Hermes Skill 查询 ──► Agent 决策

• Agent 需要了解当前热点上下文
• 基于趋势做出决策
• 舆情监控任务

模式 2: Hermes 触发 TrendRadar 分析

用户请求 ──► Hermes Agent ──► 调用 TrendRadar API ──► 深度分析 ──► 结果返回

• 特定主题的深度研究
• 竞品分析
• 趋势预测

模式 3: 双向协同 (高级)

TrendRadar 热点 ──► Hermes 分析 ──► 生成洞察 ──► TrendRadar 推送

     ▲                                               │      └──────────────── 反馈优化 ──────────────────────┘

• 全自动舆情监控
• 智能简报生成
• 趋势预警系统

详细实现方案

方案 1: TrendRadar 作为数据源

1.1 数据同步架构

<h1>trendradar_ingestor.py</h1>

""" TrendRadar 数据摄入服务 负责将 TrendRadar 数据同步到 Hermes 可用的数据存储 """

import asyncio import json from datetime import datetime, timedelta from pathlib import Path from typing import List, Dict, Optional import httpx

class TrendRadarIngestor:     """TrendRadar 数据摄入器"""

    def __init__(self, config_path: str = "config/trendradar.yaml"):         self.config = self._load_config(config_path)         self.data_dir = Path("~/.hermes/trendradar").expanduser()         self.data_dir.mkdir(parents=True, exist_ok=True)         self.cache_file = self.data_dir / "hot_topics_cache.json"         self.last_update = None

    def _load_config(self, path: str) -> dict:         """加载配置"""         import yaml         with open(path, 'r') as f:             return yaml.safe_load(f)

    async def fetch_from_trendradar(self) -> List[Dict]:         """         从 TrendRadar 获取最新热点数据

        支持多种数据获取方式:         1. 直接读取 TrendRadar 输出文件         2. 调用 TrendRadar API         3. 读取 TrendRadar GitHub Pages JSON         """         topics = []

        # 方式 1: 读取本地输出文件         output_dir = Path(self.config.get('output_dir', 'output'))         if output_dir.exists():             for json_file in output_dir.glob('*.json'):                 with open(json_file, 'r') as f:                     data = json.load(f)                     topics.extend(self._normalize_data(data))

        # 方式 2: 调用 TrendRadar API (如果部署了服务)         if self.config.get('api_endpoint'):             async with httpx.AsyncClient() as client:                 resp = await client.get(                     f"{self.config['api_endpoint']}/api/hot_topics",                     timeout=30                 )                 if resp.status_code == 200:                     topics.extend(self._normalize_data(resp.json()))

        # 方式 3: 从 GitHub Pages 获取 (公开数据)         if self.config.get('github_pages_url'):             async with httpx.AsyncClient() as client:                 resp = await client.get(                     f"{self.config['github_pages_url']}/data/hot_topics.json",                     timeout=30                 )                 if resp.status_code == 200:                     topics.extend(self._normalize_data(resp.json()))

        return topics

    def _normalize_data(self, raw_data: dict) -> List[Dict]:         """标准化数据格式"""         normalized = []

        for item in raw_data.get('topics', []):             normalized.append({                 'id': item.get('id', ''),                 'title': item.get('title', ''),                 'source': item.get('source', ''),                 'url': item.get('url', ''),                 'hot_score': item.get('score', 0),                 'ai_relevance': item.get('ai_relevance', 0),                 'category': item.get('category', 'general'),                 'keywords': item.get('keywords', []),                 'summary': item.get('summary', ''),                 'sentiment': item.get('sentiment', 'neutral'),                 'timestamp': item.get('timestamp', datetime.now().isoformat()),                 'raw_content': item.get('content', '')             })

        return normalized

    async def update_cache(self, topics: List[Dict]):         """更新本地缓存"""         cache_data = {             'last_update': datetime.now().isoformat(),             'topics': topics,             'stats': {                 'total': len(topics),                 'sources': list(set(t['source'] for t in topics)),                 'categories': list(set(t['category'] for t in topics))             }         }

        with open(self.cache_file, 'w', encoding='utf-8') as f:             json.dump(cache_data, f, ensure_ascii=False, indent=2)

        self.last_update = datetime.now()         print(f"✅ 缓存已更新: {len(topics)} 条热点")

    async def run_scheduled_sync(self):         """定时同步任务"""         while True:             try:                 print("🔄 同步 TrendRadar 数据...")                 topics = await self.fetch_from_trendradar()                 await self.update_cache(topics)

                # 等待下次同步 (默认 10 分钟)                 await asyncio.sleep(self.config.get('sync_interval', 600))             except Exception as e:                 print(f"❌ 同步失败: {e}")                 await asyncio.sleep(60)  # 出错后 1 分钟重试

<h1>启动摄入服务</h1> if __name__ == "__main__":     ingestor = TrendRadarIngestor()     asyncio.run(ingestor.run_scheduled_sync())

1.2 Hermes Skill 实现

<h1>skills/trendradar_skill.py</h1>

""" TrendRadar Skill for Hermes 为 Hermes Agent 提供热点数据查询和分析能力 """

import json from datetime import datetime, timedelta from pathlib import Path from typing import List, Dict, Optional

from hermes.tools.registry import registry

class TrendRadarSkill:     """TrendRadar 数据查询 Skill"""

    def __init__(self):         self.cache_dir = Path("~/.hermes/trendradar").expanduser()         self.cache_file = self.cache_dir / "hot_topics_cache.json"

    def _load_cache(self) -> Optional[Dict]:         """加载缓存数据"""         if not self.cache_file.exists():             return None

        with open(self.cache_file, 'r', encoding='utf-8') as f:             return json.load(f)

    def query_hot_topics(         self,         category: Optional[str] = None,         source: Optional[str] = None,         keyword: Optional[str] = None,         min_score: float = 0.5,         limit: int = 10,         time_range: str = "24h"     ) -> List[Dict]:         """         查询热点话题

        Args:             category: 分类筛选 (tech/finance/social/...)             source: 来源筛选 (zhihu/weibo/github/...)             keyword: 关键词筛选             min_score: 最低相关度分数             limit: 返回数量限制             time_range: 时间范围 (1h/6h/24h/7d)         """         cache = self._load_cache()         if not cache:             return []

        topics = cache.get('topics', [])

        # 时间范围过滤         cutoff_time = self._parse_time_range(time_range)         topics = [             t for t in topics              if datetime.fromisoformat(t['timestamp']) > cutoff_time         ]

        # 分类过滤         if category:             topics = [t for t in topics if t['category'] == category]

        # 来源过滤         if source:             topics = [t for t in topics if t['source'] == source]

        # 关键词过滤         if keyword:             keyword_lower = keyword.lower()             topics = [                 t for t in topics                  if keyword_lower in t['title'].lower()                  or keyword_lower in ' '.join(t['keywords']).lower()             ]

        # 分数过滤         topics = [t for t in topics if t['ai_relevance'] >= min_score]

        # 按热度排序         topics.sort(key=lambda x: x['hot_score'], reverse=True)

        return topics[:limit]

    def get_trend_analysis(self, topic_id: str) -> Optional[Dict]:         """获取单个话题的详细分析"""         cache = self._load_cache()         if not cache:             return None

        for topic in cache.get('topics', []):             if topic['id'] == topic_id:                 return {                     'topic': topic,                     'analysis': {                         'sentiment': topic.get('sentiment', 'neutral'),                         'keywords': topic.get('keywords', []),                         'related_topics': self._find_related(topic, cache['topics'])                     }                 }

        return None

    def get_summary_report(self, time_range: str = "24h") -> str:         """生成热点摘要报告"""         topics = self.query_hot_topics(time_range=time_range, limit=20)

        if not topics:             return "暂无热点数据"

        report_lines = [             f"## 📈 热点趋势报告 ({time_range})",             f"",             f"<strong>统计时间</strong>: {datetime.now().strftime('%Y-%m-%d %H:%M')}",             f"<strong>热点数量</strong>: {len(topics)}",             f"",             f"### 🔥 热门话题 TOP 10",             f""         ]

        for i, topic in enumerate(topics[:10], 1):             emoji = "🥇" if i == 1 else "🥈" if i == 2 else "🥉" if i == 3 else "•"             report_lines.append(                 f"{emoji} <strong>{topic['title']}</strong>\n"                 f"   📊 热度: {topic['hot_score']:.1f} | "                 f"相关度: {topic['ai_relevance']:.1f} | "                 f"来源: {topic['source']}\n"                 f"   📝 {topic.get('summary', '暂无摘要')[:100]}..."             )

        # 分类统计         categories = {}         for t in topics:             cat = t['category']             categories[cat] = categories.get(cat, 0) + 1

        report_lines.extend([             f"",             f"### 📊 分类分布",             f""         ])         for cat, count in sorted(categories.items(), key=lambda x: -x[1]):             report_lines.append(f"- {cat}: {count} 条")

        return '\n'.join(report_lines)

    def _parse_time_range(self, time_range: str) -> datetime:         """解析时间范围"""         now = datetime.now()

        if time_range == "1h":             return now - timedelta(hours=1)         elif time_range == "6h":             return now - timedelta(hours=6)         elif time_range == "24h":             return now - timedelta(days=1)         elif time_range == "7d":             return now - timedelta(days=7)         else:             return now - timedelta(days=1)

    def _find_related(self, topic: Dict, all_topics: List[Dict], limit: int = 5) -> List[Dict]:         """查找相关话题"""         related = []         topic_keywords = set(topic.get('keywords', []))

        for t in all_topics:             if t['id'] == topic['id']:                 continue

            other_keywords = set(t.get('keywords', []))             overlap = topic_keywords & other_keywords

            if len(overlap) >= 2:  # 至少 2 个共同关键词                 related.append({                     'id': t['id'],                     'title': t['title'],                     'common_keywords': list(overlap),                     'score': len(overlap)                 })

        related.sort(key=lambda x: -x['score'])         return related[:limit]

<h1>注册到 Hermes</h1> @registry.register(     name="trendradar_query",     toolset="trendradar",     schema={         "name": "trendradar_query",         "description": "查询 TrendRadar 热点数据，获取实时舆情和趋势信息",         "parameters": {             "type": "object",             "properties": {                 "query_type": {                     "type": "string",                     "enum": ["hot_topics", "trend_analysis", "summary_report"],                     "description": "查询类型"                 },                 "category": {                     "type": "string",                     "description": "分类筛选 (tech/finance/social/...)"                 },                 "keyword": {                     "type": "string",                     "description": "关键词筛选"                 },                 "time_range": {                     "type": "string",                     "enum": ["1h", "6h", "24h", "7d"],                     "default": "24h",                     "description": "时间范围"                 },                 "limit": {                     "type": "integer",                     "default": 10,                     "description": "返回数量限制"                 }             },             "required": ["query_type"]         }     } ) def trendradar_tool(args: dict, **kwargs) -> str:     """TrendRadar 工具函数"""     skill = TrendRadarSkill()

    query_type = args.get('query_type')

    if query_type == "hot_topics":         topics = skill.query_hot_topics(             category=args.get('category'),             keyword=args.get('keyword'),             time_range=args.get('time_range', '24h'),             limit=args.get('limit', 10)         )         return json.dumps(topics, ensure_ascii=False, indent=2)

    elif query_type == "summary_report":         return skill.get_summary_report(             time_range=args.get('time_range', '24h')         )

    else:         return json.dumps({"error": "Unknown query type"})

方案 2: Hermes 触发深度分析

2.1 深度分析服务

<h1>trendradar_analyzer.py</h1>

""" TrendRadar 深度分析服务 Hermes Agent 可以调用此服务进行特定主题的深度研究 """

import asyncio from typing import List, Dict, Optional from dataclasses import dataclass

@dataclass class AnalysisResult:     """分析结果"""     topic: str     sentiment: str     key_points: List[str]     trend_prediction: str     related_news: List[Dict]     risk_assessment: str     opportunities: List[str]

class TrendRadarAnalyzer:     """TrendRadar 深度分析器"""

    def __init__(self, llm_client):         self.llm = llm_client         self.skill = TrendRadarSkill()

    async def deep_analysis(self, keyword: str, depth: str = "standard") -> AnalysisResult:         """         对特定主题进行深度分析

        Args:             keyword: 分析主题/关键词             depth: 分析深度 (quick/standard/deep)         """         # 1. 收集相关数据         related_topics = self.skill.query_hot_topics(             keyword=keyword,             time_range="7d",             limit=50         )

        if not related_topics:             return AnalysisResult(                 topic=keyword,                 sentiment="unknown",                 key_points=["未找到相关数据"],                 trend_prediction="无法预测",                 related_news=[],                 risk_assessment="无数据",                 opportunities=[]             )

        # 2. LLM 深度分析         analysis_prompt = self._build_analysis_prompt(keyword, related_topics, depth)

        llm_response = await self.llm.analyze(analysis_prompt)

        # 3. 解析结果         return self._parse_analysis_result(keyword, related_topics, llm_response)

    def _build_analysis_prompt(self, keyword: str, topics: List[Dict], depth: str) -> str:         """构建分析提示词"""

        topic_summaries = "\n".join([             f"- {t['title']} (热度: {t['hot_score']:.1f}, 情感: {t['sentiment']})\n"             f"  摘要: {t.get('summary', 'N/A')[:200]}"             for t in topics[:20]         ])

        depth_instruction = {             "quick": "快速分析，重点关注核心要点",             "standard": "标准分析，平衡深度和效率",             "deep": "深度分析，详细研究所有方面"         }.get(depth, "standard")

        return f""" 你是一位专业的舆情分析师。请对以下主题进行深度分析：

<h2>分析主题</h2> {keyword}

<h2>相关数据 ({len(topics)} 条)</h2> {topic_summaries}

<h2>分析要求 ({depth_instruction})</h2>



  
<strong>情感分析</strong>: 整体舆论倾向 (positive/negative/neutral/mixed)

  
<strong>关键要点</strong>: 提取 5-10 个核心要点

  
<strong>趋势预测</strong>: 预测未来 7-30 天的发展趋势

  
<strong>风险评估</strong>: 识别潜在风险和危机信号

  
<strong>机会识别</strong>: 发现可能的机会点



请以结构化 JSON 格式输出： {{     "sentiment": "positive/negative/neutral/mixed",     "key_points": ["要点1", "要点2", ...],     "trend_prediction": "趋势预测内容",     "risk_assessment": "风险评估内容",     "opportunities": ["机会1", "机会2", ...] }} """

    def _parse_analysis_result(self, keyword: str, topics: List[Dict], llm_response: str) -> AnalysisResult:         """解析 LLM 分析结果"""         import json         import re

        # 尝试提取 JSON         json_match = re.search(r'

<h3>方案 3: 双向协同 (高级)</h3>

<h4>3.1 协同工作流</h4>

<h2>部署方案</h2>

<h3>部署架构</h3>

<h3>Docker Compose 配置</h3>

<h3>启动脚本</h3>

<h2>使用示例</h2>

<h3>场景 1: 追踪技术趋势并生成报告</h3>

<h3>场景 2: 竞品监控与预警</h3>

<h2>监控与维护</h2>

<h3>健康检查端点</h3>

总结

集成价值

| 维度 | 价值 | |------|------| | 数据 | Hermes 获得实时热点感知能力 | | 决策 | 基于趋势数据做出更明智的决策 | | 效率 | 自动化信息收集和分析流程 | | 响应 | 快速响应市场变化和热点事件 |

三种集成模式选择

| 模式 | 适用场景 | 复杂度 | |------|----------|--------| | 数据源模式 | Agent 需要了解热点上下文 | ⭐⭐ | | 触发分析模式 | 特定主题深度研究 | ⭐⭐⭐ | | 双向协同模式 | 全自动舆情监控 | ⭐⭐⭐⭐ |

下一步建议

1. 立即: 部署数据源模式，为 Hermes 添加热点查询能力
2. 本周: 实现 2-3 个具体使用场景（技术趋势追踪、竞品监控）
3. 本月: 根据使用反馈优化，考虑升级到双向协同模式