# 构建智能资讯系统：AiOnly与N8N自动化流程实战解析

在信息过载的时代，精准获取AI领域最新动态成为许多从业者的核心需求。本文将详细介绍如何结合AiOnly全球大模型服务和N8N自动化工作流，搭建个性化的AI资讯早报系统，实现从信息采集到分发的全流程自动化。

## 系统架构设计与技术选型

整个资讯系统采用模块化设计，主要包含四个核心组件：数据采集层、AI处理层、内容生成层和分发推送层。AiOnly作为大模型服务提供商，负责资讯的智能摘要和内容优化；N8N作为自动化工作流引擎，串联各个环节并处理异常情况。

**技术栈组成：**

- 数据源：主流AI资讯网站API、RSS订阅源

- 核心服务：AiOnly API接口

- 工作流引擎：N8N自托管实例

- 存储：SQLite轻量级数据库

- 推送渠道：企业微信、邮件、Webhook

## 环境配置与基础搭建

首先需要完成AiOnly服务和N8N环境的初始化配置。AiOnly提供了多种大模型接入选项，根据资讯处理需求选择合适的模型版本。

```python

# AiOnly服务配置示例

class AiOnlyConfig:

    def __init__(self):

        self.api_key = "your_api_key_here"

        self.base_url = "https://api.aionly.com/v1"

        self.model = "deepseek-reasoner"  # 选择适合摘要生成的模型

        self.max_tokens = 2000

        

    def create_client(self):

        import requests

        return requests.Session()

    

    def generate_headers(self):

        return {

            "Authorization": f"Bearer {self.api_key}",

            "Content-Type": "application/json"

        }

```

N8N采用Docker方式部署，保证环境一致性：

```yaml

# docker-compose.yml 配置

version: '3.8'

services:

  n8n:

    image: n8nio/n8n

    ports<"jushengyan.dingdundoor.cn">:

      - "5678:5678"

    environment:

      - N8N_BASIC_AUTH_ACTIVE=true

      - N8N_BASIC_AUTH_USER=user

      - N8N_BASIC_AUTH_PASSWORD=password

    volumes:

      - n8n_data:/home/node/.n8n

volumes:

  n8n_data:

```

## 数据采集模块实现

数据采集是整个系统的源头，需要从多个信息源获取最新的AI资讯内容。采用混合采集策略，结合API直接调用和RSS订阅。

```python

# 多源数据采集器

class NewsCollector:

    def __init__(self):

        self.sources = [

            {

                "name": "AINewsAPI",

                "type": "api",

                "endpoint": "https://ainewsapi.com/latest",

                "params": {"category": "ai"}

            },

            {

                "name": "TechCrunchRSS",

                "type": "rss", 

                "url": "https://techcrunch.com/feed/"

            }

        ]

    

    async def fetch_news(self):

        import aiohttp

        import feedparser

        

        all_articles = []

        

        for source in self.sources:

            try:

                if source["type"] == "api":

                    async with aiohttp.ClientSession() as session:

                        async with session.get(

                            source["endpoint"], 

                            params=source.get("params", {})

                        ) as response:

                            data = await response.json()

                            articles = self.parse_api_response(data)

                            all_<"sean.dingdundoor.cn">articles.extend(articles)

                            

                elif source["type"] == "rss":

                    feed = feedparser.parse(source["url"])

                    articles = self.parse_rss_feed(feed)

                    all_articles.extend(articles)

                    

            except Exception as e:

                print(f"采集失败 {source['name']}: {str(e)}")

                continue

                

        return self.deduplicate_articles(all_articles)

    

    def parse_api_response(self, data):

        # 解析API返回的JSON数据

        articles = []

        for item in data.get('articles', []):

            article = {

                'title': item['title'],

                'content': item.get('content', ''),

                'url': item['url'],

                'published_at': item['published_at'],

                'source': item['source']['name']

            }

            articles.append(article)

        return articles

```

## AI内容处理与摘要生成

采集到的原始资讯通过AiOnly服务进行智能处理，包括关键信息提取、内容摘要生成和重要性评分。

```python

# AI内容处理器

class ContentProcessor:

    def __init__(self, ai_config):

        self.ai_config = ai_config

        self.client = ai_config.create_client()

        

    async def process_articles(self, articles):

        processed_articles<"throw.dingdundoor.cn"> = []

        

        for article in articles:

            # 生成智能摘要

            summary = await self.generate_summary(article)

            

            # 提取关键信息

            keywords = await self.extract_keywords(article)

            

            # 内容重要性评分

            score = await self.rate_importance(article)

            

            processed_article = {

                **article,

                'ai_summary': summary,

                'keywords': keywords,

                'importance_score': score,

                'processed_at': datetime.now().isoformat()

            }

            processed_articles.append(processed_article)

            

        return sorted(processed_articles, 

                     key=lambda x: x['importance_score'], reverse=True)

    

    async def generate_summary(self, article):

        prompt = f"""

        请为以下AI相关新闻生成简洁摘要，要求：

        1. 突出技术突破或重要事件

        2. 包含关键数据和影响分析

        3. 限制在200字以内

        

        新闻标题：{article['title']}

        新闻内容：{article['content'][:1000]}...

        """

        

        response = self.client.post(

            f"{self.ai_config.base_url}/chat/completions",

            headers=self.ai_<"liangry.dingdundoor.cn">config.generate_headers(),

            json={

                "model": self.ai_config.model,

                "messages": [{"role": "user", "content": prompt}],

                "max_tokens": self.ai_config.max_tokens

            }

        )

        

        return response.json()['choices'][0]['message']['content']

```

## N8N工作流设计与实现

N8N工作流是整个系统的协调中枢，负责调度各个模块的执行顺序和异常处理。

```javascript

// N8N工作流节点配置示例

const workflowNodes = [

  {

    "name": "定时触发器",

    "type": "n8n-nodes-base.cron",

    "parameters": {

      "triggerTimes": {

        "hour": 6,

        "minute": 0,

        "mode": "everyDay"

      }

    }

  },

  {

    "name": "执行数据采集",

    "type": "n8n-nodes-base.httpRequest",

    "parameters": {

      "url": "http://localhost:8000/collect-news",

      "method": "POST"

    }

  },

  {

    "name": "AI内容处理",

    "type": "n8n-nodes-base.httpRequest", 

    "parameters": {

      "url": "http://localhost:8000/process-content",

      "method": "POST",

      "body": {

        "articles": "{{ $json.articles }}"

      }

    }

  },

  {

    "name": "生成早报内容",

    "type": "n8n-nodes-base.function",

    "parameters": {

      "functionCode": `

        const articles = $input.all();

        const topArticles = articles.slice(0, 5);

        

        let newsletter = "AI资讯早报\\\\n\\\\n";

        newsletter += "今日重点内容：\\\\n\\\\n";

        

        topArticles.forEach((article, index) => {

          newsletter += \\`\\${index + 1}. 【\\${article.source}】\\${article.title}\\\\n\\`;

          newsletter += \\`   摘要：\\${article.ai_summary}\\\\n\\\\n\\`;

        });

        

        newsletter += "---\\\\n";

        newsletter += <"sweet.dingdundoor.cn">"由AiOnly + N8N系统自动生成";

        

        return [{ newsletter }];

      `

    }

  }

];

```

## 内容模板与个性化定制

早报内容模板支持个性化定制，根据不同受众群体调整内容风格和重点。

```python

# 早报模板引擎

class NewsletterTemplate:

    def __init__(self, style="technical"):

        self.templates = {

            "technical": {

                "header": "AI技术早报 - {date}",

                "section_format": """

## {title}

**来源**：{source} | **重要度**：{score}/10

{summary}

**关键点**：{keywords}

                """,

                "footer": "生成时间：{generate_time}"

            },

            "business": {

                "header": "AI商业动态 - {date}",

                "section_format": """

### {title}

*{source}*

{summary}

🏢 商业影响：{business_impact}

                """,

                "footer": "数据来源：AiOnly分析"

            }

        }

        self.current_style = style

    

    def render_newsletter(self, articles, date):

        template = self.templates[self.current_style]

        

        content = template["header"].format(date=date) + "\n\n"

        

        for i, article in enumerate(articles[:8], 1):

            content += f"--- 资讯 {i} ---\n"

            content += template["section_format"].format(

                title=article['title'],

                source=article['source'],

                score=article['importance_score'],

                summary=article['ai_summary'],

                keywords=', '.join(article['keywords'][:3]),

                business_impact=self.analyze_business_impact(article)

            ) + "\n\n"

        

        content += template["footer"]

        return content

    

    def analyze_business_impact(self, article):

        # 简单的商业影响分析

        impact_keywords = {

            '融资': '💰 融资动态',

            '合作': '🤝 战略合作', 

            '发布': '🚀 产品发布',

            '收购': '🏢 企业收购'

        }

        

        for keyword, impact in impact_keywords.items():

            if keyword in article<"mobile.dingdundoor.cn">['title'] or keyword in article['content']:

                return impact

        return '📊 行业动态'

```

## 多渠道分发系统

生成后的早报内容通过多个渠道进行分发，确保用户能够及时收到。

```python

# 分发管理器

class DistributionManager:

    def __init__(self):

        self.channels = [

            WeChatWorkChannel(),

            EmailChannel(),

            WebhookChannel()

        ]

    

    async def distribute(self, newsletter_content):

        results = []

        

        for channel in self.channels:

            try:

                result = await channel.send(newsletter_content)

                results.append({

                    'channel': channel.name,

                    'status': 'success',

                    'response': result

                })

            except Exception as e:

                results.append({

                    'channel': channel.name, 

                    'status': 'failed',

                    'error': str(e)

                })

        

        return results

# 企业微信通道实现

class WeChatWorkChannel:

    def __init__(self):

        self.name = "企业微信"

        self.webhook_url =<"house.dingdundoor.cn"> "https://qyapi.weixin.qq.com/cgi-bin/webhook/send"

    

    async def send(self, content):

        import aiohttp

        

        payload = {

            "msgtype": "markdown",

            "markdown": {

                "content": f"**AI资讯早报**\n\n{content}"

            }

        }

        

        async with aiohttp.ClientSession() as session:

            async with session.post(

                self.webhook_url, 

                json=payload,

                params={"key": "your_webhook_key"}

            ) as response:

                return await response.json()

```

## 监控与优化机制

建立完善的监控体系，确保系统稳定运行并持续优化。

```python

# 系统监控器

class SystemMonitor:

    def __init__(self):

        self.metrics = {

            'collection_success': 0,

            'collection_failed': 0,

            'processing_time': [],

            'distribution_status': []

        }

    

    def record_metric(self, metric_type, value=None):

        if metric_type in ['collection_success', 'collection_failed']:

            self.metrics[metric_type] += 1

        elif metric_type == 'processing_time':

            self.metrics['processing_time'].append(value)

        elif metric_type == 'distribution_status':

            self.metrics['distribution_status'].append(value)

    

    def generate_report(self):

        avg_processing_time = (

            sum(self.metrics['processing_time']) / 

            len(self.metrics['processing_time'])

            if self.metrics['processing_time'] else 0

        )

        

        success_rate = (

            self.metrics['collection_success'] / 

            (self.metrics['collection_success'] + self.metrics['collection_failed'])

            if (self.metrics['collection_success'] + self.metrics['collection_failed']) > 0 

            else 0

        )

        

        return {

            'success_rate': round(success_rate * 100, 2),

            'avg_processing_time_seconds': round(avg_processing_time, 2),

            'total_articles_processed': self.metrics['collection_success'],

            'distribution_success_count': len(

                [s for s in self.metrics<"fan.dingdundoor.cn">['distribution_status'] if s]

            )

        }

```

## 部署与维护实践

系统采用容器化部署，便于扩展和维护。使用Docker Compose管理所有服务组件。

```yaml

version: '3.8'

services:

  n8n:

    image: n8nio/n8n

    ports: ["5678:5678"]

    environment: [...]

    

  news-collector:

    build: ./news-collector

    environment:

      - AiONLY_API_KEY=${AIONLY_API_KEY}

    

  redis:

    image: redis:alpine

    ports: ["6379:6379"]

    

  monitor:

    image: prom/prometheus

    ports: ["9090:9090"]

```

通过这套系统，用户能够每天自动获取精心筛选和整理的AI领域最新动态，大大提升了信息获取效率。系统的模块化设计也便于后续的功能扩展和优化调整。