feat(devpulse_ai): add multi-agent signal intelligence reference implementation

2026-09-21 05:10:55 +08:00 · 2026-02-03 13:35:07 +05:30
parent 60b72c6479
commit 92f2803136
25 changed files with 1660 additions and 0 deletions
@@ -0,0 +1,150 @@
+## 🧠 DevPulseAI - Multi-Agent Signal Intelligence Pipeline
+
+A reference implementation demonstrating a **multi-agent system** for aggregating, analyzing, and synthesizing technical signals from multiple developer-focused sources.
+
+### Features
+
+- **Multi-Source Signal Collection** - Aggregates data from GitHub, ArXiv, HackerNews, Medium, and HuggingFace
+- **LLM-Powered Analysis** - Four specialized agents working in concert
+- **Structured Intelligence Output** - Prioritized digest with actionable recommendations
+
+### Architecture
+
+```
+┌─────────────────────────────────────────────────────────────────┐
+│                    Signal Intelligence Pipeline                  │
+├─────────────────────────────────────────────────────────────────┤
+│                                                                  │
+│  ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐         │
+│  │ GitHub │ │ ArXiv  │ │  HN    │ │ Medium │ │   HF   │ ← Data  │
+│  └───┬────┘ └───┬────┘ └───┬────┘ └───┬────┘ └───┬────┘         │
+│      └──────────┴──────────┼──────────┴──────────┘               │
+│       │             │             │                              │
+│       └─────────────┼─────────────┘                              │
+│                     ▼                                            │
+│           ┌─────────────────┐                                    │
+│           │ Signal Collector│   ← Agent 1: Ingestion            │
+│           └────────┬────────┘                                    │
+│                    ▼                                             │
+│           ┌─────────────────┐                                    │
+│           │ Relevance Agent │   ← Agent 2: Scoring (0-100)      │
+│           └────────┬────────┘                                    │
+│                    ▼                                             │
+│           ┌─────────────────┐                                    │
+│           │   Risk Agent    │   ← Agent 3: Security Assessment  │
+│           └────────┬────────┘                                    │
+│                    ▼                                             │
+│           ┌─────────────────┐                                    │
+│           │ Synthesis Agent │   ← Agent 4: Final Digest         │
+│           └────────┬────────┘                                    │
+│                    ▼                                             │
+│           ┌─────────────────┐                                    │
+│           │ Intelligence    │   ← Prioritized Output            │
+│           │ Digest          │                                    │
+│           └─────────────────┘                                    │
+└─────────────────────────────────────────────────────────────────┘
+```
+
+### Agent Responsibilities
+
+| Agent | Role | Output |
+|-------|------|--------|
+| **SignalCollectorAgent** | Aggregates & normalizes signals | Unified signal list |
+| **RelevanceAgent** | Scores developer relevance (0-100) | Score + reasoning |
+| **RiskAgent** | Identifies security/breaking changes | Risk level + concerns |
+| **SynthesisAgent** | Produces final intelligence digest | Prioritized recommendations |
+
+### How to Get Started
+
+1. Clone the repository
+
+```bash
+git clone https://github.com/Shubhamsaboo/awesome-llm-apps.git
+cd advanced_ai_agents/multi_agent_apps/devpulse_ai
+```
+
+1. Install dependencies
+
+```bash
+pip install -r requirements.txt
+```
+
+1. Set your OpenAI API key (optional for live mode)
+
+```bash
+export OPENAI_API_KEY=your_api_key
+```
+
+1. Run the verification script (no API key needed)
+
+```bash
+python verify.py
+```
+
+1. Run the full pipeline (requires API key for LLM agents)
+
+```bash
+python main.py
+```
+
+### Verification Script
+
+The `verify.py` script tests the entire pipeline using **mock data only** - no network calls or API keys required:
+
+```bash
+python verify.py
+```
+
+Expected output:
+
+```
+[OK] DevPulseAI reference pipeline executed successfully
+```
+
+### Optional: n8n Automation
+
+An n8n workflow is included for those who want to automate the pipeline:
+
+- **Location**: `workflows/signal-intelligence-pipeline.json`
+- **Import**: n8n → Settings → Import from File
+- **Requires**: n8n instance + configured credentials
+
+This is entirely optional - the Python implementation works standalone.
+
+### Directory Structure
+
+```
+devpulse_ai/
+├── adapters/
+│   ├── github.py       # GitHub trending repos
+│   ├── arxiv.py        # AI/ML research papers
+│   ├── hackernews.py   # Tech news stories
+│   ├── medium.py       # Tech blog RSS feeds
+│   └── huggingface.py  # HuggingFace models
+├── agents/
+│   ├── __init__.py
+│   ├── signal_collector.py
+│   ├── relevance_agent.py
+│   ├── risk_agent.py
+│   └── synthesis_agent.py
+├── workflows/
+│   └── signal-intelligence-pipeline.json
+├── main.py             # Full pipeline demo
+├── verify.py           # Mock data verification
+├── requirements.txt
+└── README.md
+```
+
+### How It Works
+
+1. **Signal Collection**: Adapters fetch data from GitHub, ArXiv, HackerNews, Medium, and HuggingFace
+2. **Normalization**: SignalCollectorAgent unifies signals to a common schema
+3. **Relevance Scoring**: RelevanceAgent rates each signal 0-100 for developer relevance
+4. **Risk Assessment**: RiskAgent flags security issues and breaking changes
+5. **Synthesis**: SynthesisAgent produces a prioritized intelligence digest
+
+### Built With
+
+- [Agno](https://github.com/agno-agi/agno) - Multi-agent framework
+- [OpenAI GPT-4o-mini](https://openai.com/) - LLM backbone
+- [httpx](https://www.python-httpx.org/) - Async HTTP client
@@ -0,0 +1,86 @@
+"""
+ArXiv Adapter - Fetches recent AI/ML research papers.
+
+This is a simplified, stateless adapter for the DevPulseAI reference implementation.
+ArXiv API is public and requires no authentication.
+"""
+
+import httpx
+import xml.etree.ElementTree as ET
+from typing import List, Dict, Any
+
+
+def fetch_arxiv_papers(limit: int = 5) -> List[Dict[str, Any]]:
+    """
+    Fetch recent AI/ML papers from ArXiv.
+    
+    Args:
+        limit: Maximum number of papers to return.
+        
+    Returns:
+        List of signal dictionaries with standardized schema.
+    """
+    base_url = "https://export.arxiv.org/api/query"
+    params = {
+        "search_query": "cat:cs.AI OR cat:cs.LG",
+        "start": 0,
+        "max_results": limit,
+        "sortBy": "submittedDate",
+        "sortOrder": "descending"
+    }
+    
+    signals = []
+    
+    try:
+        response = httpx.get(base_url, params=params, timeout=15.0)
+        response.raise_for_status()
+        
+        # Parse Atom XML response
+        root = ET.fromstring(response.content)
+        ns = {"atom": "http://www.w3.org/2005/Atom"}
+        
+        for entry in root.findall("atom:entry", ns):
+            title_elem = entry.find("atom:title", ns)
+            summary_elem = entry.find("atom:summary", ns)
+            id_elem = entry.find("atom:id", ns)
+            published_elem = entry.find("atom:published", ns)
+            
+            title = title_elem.text.strip() if title_elem is not None else "Untitled"
+            summary = summary_elem.text.strip() if summary_elem is not None else ""
+            arxiv_id = id_elem.text.strip() if id_elem is not None else ""
+            published = published_elem.text if published_elem is not None else ""
+            
+            # Get PDF link
+            pdf_link = arxiv_id
+            link_elem = entry.find("atom:link[@title='pdf']", ns)
+            if link_elem is not None:
+                pdf_link = link_elem.attrib.get("href", arxiv_id)
+            
+            signal = {
+                "id": arxiv_id,
+                "source": "arxiv",
+                "title": title,
+                "description": summary[:500] + "..." if len(summary) > 500 else summary,
+                "url": arxiv_id,
+                "metadata": {
+                    "pdf": pdf_link,
+                    "published": published
+                }
+            }
+            signals.append(signal)
+            
+    except httpx.HTTPError as e:
+        print(f"[ArXiv Adapter] HTTP error: {e}")
+    except ET.ParseError as e:
+        print(f"[ArXiv Adapter] XML parse error: {e}")
+    except Exception as e:
+        print(f"[ArXiv Adapter] Error: {e}")
+    
+    return signals
+
+
+if __name__ == "__main__":
+    # Quick test
+    results = fetch_arxiv_papers(limit=3)
+    for r in results:
+        print(f"- {r['title'][:60]}...")
@@ -0,0 +1,65 @@
+"""
+GitHub Adapter - Fetches trending repositories from GitHub.
+
+This is a simplified, stateless adapter for the DevPulseAI reference implementation.
+No authentication required for basic public API access.
+"""
+
+import httpx
+from datetime import datetime, timedelta
+from typing import List, Dict, Any
+
+
+def fetch_github_trending(limit: int = 5) -> List[Dict[str, Any]]:
+    """
+    Fetch trending GitHub repositories created in the last 24 hours.
+    
+    Args:
+        limit: Maximum number of repositories to return.
+        
+    Returns:
+        List of signal dictionaries with standardized schema.
+    """
+    base_url = "https://api.github.com/search/repositories"
+    date_query = (datetime.utcnow() - timedelta(days=1)).strftime("%Y-%m-%d")
+    
+    params = {
+        "q": f"created:>{date_query} sort:stars",
+        "per_page": limit
+    }
+    
+    signals = []
+    
+    try:
+        response = httpx.get(base_url, params=params, timeout=10.0)
+        response.raise_for_status()
+        data = response.json()
+        
+        for item in data.get("items", []):
+            signal = {
+                "id": str(item["id"]),
+                "source": "github",
+                "title": item["full_name"],
+                "description": item.get("description") or "No description",
+                "url": item["html_url"],
+                "metadata": {
+                    "stars": item["stargazers_count"],
+                    "language": item.get("language"),
+                    "topics": item.get("topics", [])
+                }
+            }
+            signals.append(signal)
+            
+    except httpx.HTTPError as e:
+        print(f"[GitHub Adapter] HTTP error: {e}")
+    except Exception as e:
+        print(f"[GitHub Adapter] Error: {e}")
+    
+    return signals
+
+
+if __name__ == "__main__":
+    # Quick test
+    results = fetch_github_trending(limit=3)
+    for r in results:
+        print(f"- {r['title']}: {r['metadata']['stars']} stars")
@@ -0,0 +1,72 @@
+"""
+HackerNews Adapter - Fetches top AI/ML stories from HackerNews.
+
+This is a simplified, stateless adapter for the DevPulseAI reference implementation.
+Uses the Algolia HN API for better search capabilities.
+"""
+
+import httpx
+from typing import List, Dict, Any
+
+
+def fetch_hackernews_stories(limit: int = 5) -> List[Dict[str, Any]]:
+    """
+    Fetch recent AI/ML related stories from HackerNews.
+    
+    Args:
+        limit: Maximum number of stories to return.
+        
+    Returns:
+        List of signal dictionaries with standardized schema.
+    """
+    base_url = "https://hn.algolia.com/api/v1/search_by_date"
+    params = {
+        "query": "AI OR LLM OR Machine Learning OR GPT",
+        "tags": "story",
+        "hitsPerPage": limit,
+        "numericFilters": "points>5"
+    }
+    
+    signals = []
+    
+    try:
+        response = httpx.get(base_url, params=params, timeout=10.0)
+        response.raise_for_status()
+        data = response.json()
+        
+        for hit in data.get("hits", []):
+            # Skip stories without URLs (Ask HN, etc.)
+            if not hit.get("url") and not hit.get("story_text"):
+                continue
+            
+            external_id = str(hit.get("objectID", ""))
+            hn_url = f"https://news.ycombinator.com/item?id={external_id}"
+            
+            signal = {
+                "id": external_id,
+                "source": "hackernews",
+                "title": hit.get("title", "Untitled"),
+                "description": hit.get("story_text", "")[:300] if hit.get("story_text") else "",
+                "url": hit.get("url") or hn_url,
+                "metadata": {
+                    "points": hit.get("points", 0),
+                    "comments": hit.get("num_comments", 0),
+                    "author": hit.get("author", "unknown"),
+                    "hn_url": hn_url
+                }
+            }
+            signals.append(signal)
+            
+    except httpx.HTTPError as e:
+        print(f"[HackerNews Adapter] HTTP error: {e}")
+    except Exception as e:
+        print(f"[HackerNews Adapter] Error: {e}")
+    
+    return signals
+
+
+if __name__ == "__main__":
+    # Quick test
+    results = fetch_hackernews_stories(limit=3)
+    for r in results:
+        print(f"- {r['title']}: {r['metadata']['points']} points")
@@ -0,0 +1,80 @@
+"""
+HuggingFace Adapter - Fetches trending models from HuggingFace Hub.
+
+This is a simplified, stateless adapter for the DevPulseAI reference implementation.
+Uses the public HuggingFace API (no authentication required for basic access).
+"""
+
+import httpx
+from typing import List, Dict, Any
+
+
+def fetch_huggingface_models(limit: int = 5) -> List[Dict[str, Any]]:
+    """
+    Fetch trending/popular models from HuggingFace Hub.
+    
+    Args:
+        limit: Maximum number of models to return.
+        
+    Returns:
+        List of signal dictionaries with standardized schema.
+    """
+    base_url = "https://huggingface.co/api/models"
+    params = {
+        "sort": "likes",
+        "direction": "-1",
+        "limit": limit
+    }
+    
+    signals = []
+    
+    try:
+        response = httpx.get(base_url, params=params, timeout=10.0)
+        response.raise_for_status()
+        data = response.json()
+        
+        for item in data:
+            model_id = item.get("modelId", item.get("id", "unknown"))
+            
+            # Build description from model metadata
+            tags = item.get("tags", [])
+            pipeline = item.get("pipeline_tag", "")
+            description_parts = []
+            
+            if pipeline:
+                description_parts.append(f"Pipeline: {pipeline}")
+            if tags:
+                description_parts.append(f"Tags: {', '.join(tags[:5])}")
+            
+            description_parts.append(f"Downloads: {item.get('downloads', 0):,}")
+            description_parts.append(f"Likes: {item.get('likes', 0):,}")
+            
+            signal = {
+                "id": model_id,
+                "source": "huggingface",
+                "title": f"HF Model: {model_id}",
+                "description": " | ".join(description_parts),
+                "url": f"https://huggingface.co/{model_id}",
+                "metadata": {
+                    "downloads": item.get("downloads", 0),
+                    "likes": item.get("likes", 0),
+                    "pipeline_tag": pipeline,
+                    "tags": tags[:10],
+                    "author": item.get("author", "")
+                }
+            }
+            signals.append(signal)
+            
+    except httpx.HTTPError as e:
+        print(f"[HuggingFace Adapter] HTTP error: {e}")
+    except Exception as e:
+        print(f"[HuggingFace Adapter] Error: {e}")
+    
+    return signals
+
+
+if __name__ == "__main__":
+    # Quick test
+    results = fetch_huggingface_models(limit=3)
+    for r in results:
+        print(f"- {r['title']}: {r['metadata']['likes']} likes")
@@ -0,0 +1,70 @@
+"""
+Medium Adapter - Fetches tech blogs from Medium and other RSS feeds.
+
+This is a simplified, stateless adapter for the DevPulseAI reference implementation.
+Uses feedparser to fetch from RSS/Atom feeds.
+"""
+
+import feedparser
+from typing import List, Dict, Any
+
+
+# Tech blog feeds to monitor
+FEEDS = [
+    "https://medium.com/feed/tag/artificial-intelligence",
+    "https://medium.com/feed/tag/machine-learning",
+    "https://medium.com/feed/@netflixtechblog",
+    "https://engineering.fb.com/feed/",
+]
+
+
+def fetch_medium_blogs(limit: int = 5) -> List[Dict[str, Any]]:
+    """
+    Fetch recent tech blogs from Medium and engineering blogs.
+    
+    Args:
+        limit: Maximum number of entries per feed.
+        
+    Returns:
+        List of signal dictionaries with standardized schema.
+    """
+    signals = []
+    
+    for feed_url in FEEDS:
+        try:
+            feed = feedparser.parse(feed_url)
+            
+            for entry in feed.entries[:limit]:
+                # Get summary or description
+                summary = getattr(entry, "summary", "") or getattr(entry, "description", "")
+                
+                # Clean HTML tags from summary (simple approach)
+                if summary:
+                    import re
+                    summary = re.sub(r'<[^>]+>', '', summary)[:500]
+                
+                signal = {
+                    "id": entry.get("id", entry.link),
+                    "source": "medium",
+                    "title": entry.title,
+                    "description": summary,
+                    "url": entry.link,
+                    "metadata": {
+                        "published": getattr(entry, "published", ""),
+                        "author": getattr(entry, "author", "Unknown"),
+                        "feed": feed_url
+                    }
+                }
+                signals.append(signal)
+                
+        except Exception as e:
+            print(f"[Medium Adapter] Error fetching {feed_url}: {e}")
+    
+    return signals
+
+
+if __name__ == "__main__":
+    # Quick test
+    results = fetch_medium_blogs(limit=2)
+    for r in results:
+        print(f"- {r['title'][:60]}...")
@@ -0,0 +1,21 @@
+"""
+DevPulseAI Agents Package
+
+This package contains four specialized agents for the signal intelligence pipeline:
+- SignalCollectorAgent: Aggregates signals from multiple sources
+- RelevanceAgent: Scores signals based on developer relevance  
+- RiskAgent: Assesses security risks and breaking changes
+- SynthesisAgent: Produces final intelligence digest
+"""
+
+from .signal_collector import SignalCollectorAgent
+from .relevance_agent import RelevanceAgent
+from .risk_agent import RiskAgent
+from .synthesis_agent import SynthesisAgent
+
+__all__ = [
+    "SignalCollectorAgent",
+    "RelevanceAgent", 
+    "RiskAgent",
+    "SynthesisAgent"
+]
@@ -0,0 +1,121 @@
+"""
+Relevance Agent - Scores signals based on developer relevance.
+
+This agent uses LLM reasoning to score each signal from 0-100
+based on its relevance to AI/ML developers and engineers.
+"""
+
+from typing import Dict, Any, Optional
+from agno.agent import Agent
+from agno.models.openai import OpenAIChat
+
+
+class RelevanceAgent:
+    """
+    Agent that scores signals based on relevance to developers.
+    
+    Responsibilities:
+    - Score signals 0-100 based on developer relevance
+    - Provide reasoning for each score
+    - Prioritize actionable, timely content
+    """
+    
+    def __init__(self, model_id: str = "gpt-4o-mini"):
+        """
+        Initialize the Relevance Agent.
+        
+        Args:
+            model_id: OpenAI model to use for scoring.
+        """
+        self.model_id = model_id
+        self.agent = Agent(
+            name="Relevance Scorer",
+            model=OpenAIChat(id=model_id),
+            role="Scores technical signals based on developer relevance",
+            instructions=[
+                "Score each signal from 0-100 based on relevance.",
+                "Consider: novelty, impact, actionability, and timeliness.",
+                "Prioritize signals relevant to AI/ML engineers.",
+                "Provide brief reasoning for each score."
+            ],
+            markdown=True
+        )
+    
+    def score(self, signal: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Score a signal for relevance.
+        
+        Args:
+            signal: Signal dictionary to score.
+            
+        Returns:
+            Dictionary with score and reasoning.
+        """
+        prompt = f"""
+Rate the relevance of this signal for AI/ML developers.
+Score from 0-100 where:
+- 0-30: Low relevance (noise, off-topic)
+- 31-60: Moderate relevance (interesting but not urgent)
+- 61-80: High relevance (important for developers to know)
+- 81-100: Critical relevance (must-know, actionable)
+
+Signal:
+- Source: {signal.get('source', 'unknown')}
+- Title: {signal.get('title', 'Untitled')}
+- Description: {signal.get('description', '')[:500]}
+
+Respond with ONLY a JSON object:
+{{"score": <number>, "reasoning": "<one sentence>"}}
+"""
+        
+        try:
+            response = self.agent.run(prompt, stream=False)
+            # Parse response - in real use, would parse JSON
+            return self._parse_response(response.content, signal)
+        except Exception as e:
+            return self._fallback_score(signal, str(e))
+    
+    def score_batch(self, signals: list) -> list:
+        """
+        Score multiple signals.
+        
+        Args:
+            signals: List of signal dictionaries.
+            
+        Returns:
+            List of signals with scores added.
+        """
+        scored = []
+        for signal in signals:
+            result = self.score(signal)
+            signal_with_score = {**signal, "relevance": result}
+            scored.append(signal_with_score)
+        return scored
+    
+    def _parse_response(self, content: str, signal: Dict) -> Dict[str, Any]:
+        """Parse LLM response into structured output."""
+        import json
+        try:
+            # Try to extract JSON from response
+            content = content.strip()
+            if "```" in content:
+                content = content.split("```")[1].replace("json", "").strip()
+            return json.loads(content)
+        except:
+            return self._fallback_score(signal, "Parse error")
+    
+    def _fallback_score(self, signal: Dict, error: str) -> Dict[str, Any]:
+        """Provide fallback score when LLM call fails."""
+        # Simple heuristic based on metadata
+        score = 50  # Default moderate score
+        metadata = signal.get("metadata", {})
+        
+        if metadata.get("stars", 0) > 100:
+            score += 20
+        if metadata.get("points", 0) > 50:
+            score += 15
+            
+        return {
+            "score": min(score, 100),
+            "reasoning": f"Heuristic score (LLM unavailable: {error})"
+        }
@@ -0,0 +1,133 @@
+"""
+Risk Agent - Assesses security risks and breaking changes.
+
+This agent analyzes signals for potential risks including:
+- Security vulnerabilities
+- Breaking changes in dependencies
+- Deprecation notices
+"""
+
+from typing import Dict, Any, List
+from agno.agent import Agent
+from agno.models.openai import OpenAIChat
+
+
+class RiskAgent:
+    """
+    Agent that assesses risk levels in technical signals.
+    
+    Responsibilities:
+    - Identify security vulnerabilities
+    - Flag breaking changes
+    - Detect deprecation notices
+    - Rate overall risk level
+    """
+    
+    RISK_LEVELS = ["LOW", "MEDIUM", "HIGH", "CRITICAL"]
+    
+    def __init__(self, model_id: str = "gpt-4o-mini"):
+        """
+        Initialize the Risk Agent.
+        
+        Args:
+            model_id: OpenAI model to use for risk assessment.
+        """
+        self.model_id = model_id
+        self.agent = Agent(
+            name="Risk Assessor",
+            model=OpenAIChat(id=model_id),
+            role="Assesses security and breaking change risks in technical signals",
+            instructions=[
+                "Analyze signals for security vulnerabilities.",
+                "Identify breaking changes that may affect developers.",
+                "Flag deprecation notices and migration requirements.",
+                "Rate risk level: LOW, MEDIUM, HIGH, or CRITICAL."
+            ],
+            markdown=True
+        )
+    
+    def assess(self, signal: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Assess risk level of a signal.
+        
+        Args:
+            signal: Signal dictionary to assess.
+            
+        Returns:
+            Dictionary with risk assessment.
+        """
+        prompt = f"""
+Analyze this technical signal for risks:
+
+Signal:
+- Source: {signal.get('source', 'unknown')}
+- Title: {signal.get('title', 'Untitled')}
+- Description: {signal.get('description', '')[:500]}
+
+Assess for:
+1. Security vulnerabilities
+2. Breaking changes
+3. Deprecations
+
+Respond with ONLY a JSON object:
+{{"risk_level": "LOW|MEDIUM|HIGH|CRITICAL", "concerns": ["<list of concerns>"], "breaking_changes": true|false}}
+"""
+        
+        try:
+            response = self.agent.run(prompt, stream=False)
+            return self._parse_response(response.content, signal)
+        except Exception as e:
+            return self._fallback_assessment(signal, str(e))
+    
+    def assess_batch(self, signals: list) -> list:
+        """
+        Assess multiple signals for risk.
+        
+        Args:
+            signals: List of signal dictionaries.
+            
+        Returns:
+            List of signals with risk assessments added.
+        """
+        assessed = []
+        for signal in signals:
+            result = self.assess(signal)
+            signal_with_risk = {**signal, "risk": result}
+            assessed.append(signal_with_risk)
+        return assessed
+    
+    def _parse_response(self, content: str, signal: Dict) -> Dict[str, Any]:
+        """Parse LLM response into structured output."""
+        import json
+        try:
+            content = content.strip()
+            if "```" in content:
+                content = content.split("```")[1].replace("json", "").strip()
+            return json.loads(content)
+        except:
+            return self._fallback_assessment(signal, "Parse error")
+    
+    def _fallback_assessment(self, signal: Dict, error: str) -> Dict[str, Any]:
+        """Provide fallback assessment when LLM call fails."""
+        title = signal.get("title", "").lower()
+        
+        # Simple keyword-based heuristics
+        risk_level = "LOW"
+        concerns = []
+        
+        risk_keywords = {
+            "HIGH": ["vulnerability", "exploit", "CVE", "critical", "breach"],
+            "MEDIUM": ["breaking", "deprecated", "removed", "migration"],
+        }
+        
+        for level, keywords in risk_keywords.items():
+            if any(kw.lower() in title for kw in keywords):
+                risk_level = level
+                concerns.append(f"Keyword match: {level}")
+                break
+        
+        return {
+            "risk_level": risk_level,
+            "concerns": concerns if concerns else [f"Heuristic (LLM unavailable: {error})"],
+            "breaking_changes": "breaking" in title.lower()
+        }
@@ -0,0 +1,100 @@
+"""
+Signal Collector Agent - Aggregates signals from multiple data sources.
+
+This agent is responsible for the ingestion phase of the pipeline.
+It collects signals from GitHub, ArXiv, and HackerNews, then normalizes
+them into a unified schema for downstream processing.
+"""
+
+from typing import List, Dict, Any
+from agno.agent import Agent
+from agno.models.openai import OpenAIChat
+
+
+class SignalCollectorAgent:
+    """
+    Agent that collects and normalizes signals from multiple sources.
+    
+    Responsibilities:
+    - Fetch data from configured adapters
+    - Normalize signals to unified schema
+    - Deduplicate and filter low-quality signals
+    """
+    
+    def __init__(self, model_id: str = "gpt-4o-mini"):
+        """
+        Initialize the Signal Collector Agent.
+        
+        Args:
+            model_id: OpenAI model to use for signal processing.
+        """
+        self.model_id = model_id
+        self.agent = Agent(
+            name="Signal Collector",
+            model=OpenAIChat(id=model_id),
+            role="Collects and normalizes technical signals from multiple sources",
+            instructions=[
+                "You aggregate signals from GitHub, ArXiv, and HackerNews.",
+                "Normalize all signals to a consistent format.",
+                "Filter out low-quality or duplicate signals.",
+                "Prioritize signals relevant to AI/ML developers."
+            ],
+            markdown=True
+        )
+    
+    def collect(self, signals: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """
+        Process and normalize collected signals.
+        
+        Args:
+            signals: Raw signals from adapters.
+            
+        Returns:
+            List of normalized signal dictionaries.
+        """
+        normalized = []
+        seen_ids = set()
+        
+        for signal in signals:
+            # Deduplicate
+            signal_id = f"{signal.get('source', 'unknown')}:{signal.get('id', '')}"
+            if signal_id in seen_ids:
+                continue
+            seen_ids.add(signal_id)
+            
+            # Ensure required fields
+            normalized_signal = {
+                "id": signal.get("id", ""),
+                "source": signal.get("source", "unknown"),
+                "title": signal.get("title", "Untitled"),
+                "description": signal.get("description", ""),
+                "url": signal.get("url", ""),
+                "metadata": signal.get("metadata", {}),
+                "collected_at": self._get_timestamp()
+            }
+            normalized.append(normalized_signal)
+        
+        return normalized
+    
+    def _get_timestamp(self) -> str:
+        """Get current UTC timestamp."""
+        from datetime import datetime
+        return datetime.utcnow().isoformat() + "Z"
+    
+    def summarize_collection(self, signals: List[Dict[str, Any]]) -> str:
+        """
+        Generate a summary of collected signals.
+        
+        Args:
+            signals: List of collected signals.
+            
+        Returns:
+            Summary string.
+        """
+        sources = {}
+        for s in signals:
+            src = s.get("source", "unknown")
+            sources[src] = sources.get(src, 0) + 1
+        
+        summary_parts = [f"{count} from {src}" for src, count in sources.items()]
+        return f"Collected {len(signals)} signals: {', '.join(summary_parts)}"
@@ -0,0 +1,154 @@
+"""
+Synthesis Agent - Produces final intelligence digest.
+
+This agent combines outputs from all previous agents to create
+a comprehensive, actionable intelligence summary for developers.
+"""
+
+from typing import Dict, Any, List
+from agno.agent import Agent
+from agno.models.openai import OpenAIChat
+
+
+class SynthesisAgent:
+    """
+    Agent that synthesizes all signal intelligence into a final digest.
+    
+    Responsibilities:
+    - Combine relevance and risk assessments
+    - Prioritize signals by importance
+    - Generate executive summary
+    - Produce actionable recommendations
+    """
+    
+    def __init__(self, model_id: str = "gpt-4o-mini"):
+        """
+        Initialize the Synthesis Agent.
+        
+        Args:
+            model_id: OpenAI model to use for synthesis.
+        """
+        self.model_id = model_id
+        self.agent = Agent(
+            name="Intelligence Synthesizer",
+            model=OpenAIChat(id=model_id),
+            role="Synthesizes technical signals into actionable intelligence digests",
+            instructions=[
+                "Combine relevance scores and risk assessments.",
+                "Prioritize by: high relevance + critical risks first.",
+                "Generate an executive summary.",
+                "Provide actionable recommendations for developers."
+            ],
+            markdown=True
+        )
+    
+    def synthesize(self, signals: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """
+        Synthesize signals into a final intelligence digest.
+        
+        Args:
+            signals: List of signals with relevance and risk data.
+            
+        Returns:
+            Complete intelligence digest.
+        """
+        # Sort by priority (high relevance + high risk first)
+        prioritized = self._prioritize_signals(signals)
+        
+        # Group by category
+        grouped = self._group_by_source(prioritized)
+        
+        # Generate summary
+        summary = self._generate_summary(prioritized)
+        
+        return {
+            "generated_at": self._get_timestamp(),
+            "total_signals": len(signals),
+            "executive_summary": summary,
+            "priority_signals": prioritized[:5],  # Top 5
+            "signals_by_source": grouped,
+            "recommendations": self._generate_recommendations(prioritized)
+        }
+    
+    def _prioritize_signals(self, signals: List[Dict]) -> List[Dict]:
+        """Sort signals by priority score."""
+        def priority_score(signal):
+            relevance = signal.get("relevance", {}).get("score", 50)
+            risk = signal.get("risk", {})
+            risk_multiplier = {
+                "CRITICAL": 2.0,
+                "HIGH": 1.5,
+                "MEDIUM": 1.0,
+                "LOW": 0.8
+            }.get(risk.get("risk_level", "LOW"), 1.0)
+            return relevance * risk_multiplier
+        
+        return sorted(signals, key=priority_score, reverse=True)
+    
+    def _group_by_source(self, signals: List[Dict]) -> Dict[str, List]:
+        """Group signals by their source."""
+        grouped = {}
+        for signal in signals:
+            source = signal.get("source", "unknown")
+            if source not in grouped:
+                grouped[source] = []
+            grouped[source].append(signal)
+        return grouped
+    
+    def _generate_summary(self, signals: List[Dict]) -> str:
+        """Generate executive summary."""
+        if not signals:
+            return "No signals to summarize."
+        
+        high_priority = [s for s in signals 
+                        if s.get("relevance", {}).get("score", 0) >= 70]
+        critical_risks = [s for s in signals 
+                         if s.get("risk", {}).get("risk_level") in ["HIGH", "CRITICAL"]]
+        
+        parts = [f"Analyzed {len(signals)} signals."]
+        
+        if high_priority:
+            parts.append(f"{len(high_priority)} high-relevance items detected.")
+        
+        if critical_risks:
+            parts.append(f"⚠️ {len(critical_risks)} signals with elevated risk.")
+        
+        if signals:
+            top = signals[0]
+            parts.append(f"Top signal: {top.get('title', 'Unknown')}")
+        
+        return " ".join(parts)
+    
+    def _generate_recommendations(self, signals: List[Dict]) -> List[str]:
+        """Generate actionable recommendations."""
+        recommendations = []
+        
+        critical_risks = [s for s in signals 
+                         if s.get("risk", {}).get("risk_level") == "CRITICAL"]
+        if critical_risks:
+            recommendations.append(
+                f"🚨 Review {len(critical_risks)} critical-risk signals immediately"
+            )
+        
+        high_relevance = [s for s in signals 
+                         if s.get("relevance", {}).get("score", 0) >= 80]
+        if high_relevance:
+            recommendations.append(
+                f"📌 Prioritize {len(high_relevance)} high-relevance items"
+            )
+        
+        github_signals = [s for s in signals if s.get("source") == "github"]
+        if github_signals:
+            recommendations.append(
+                f"⭐ Explore {len(github_signals)} trending repositories"
+            )
+        
+        if not recommendations:
+            recommendations.append("✅ No urgent actions required")
+        
+        return recommendations
+    
+    def _get_timestamp(self) -> str:
+        """Get current UTC timestamp."""
+        from datetime import datetime
+        return datetime.utcnow().isoformat() + "Z"
@@ -0,0 +1,143 @@
+"""
+DevPulseAI - Multi-Agent Signal Intelligence Pipeline
+
+This script demonstrates a complete multi-agent workflow for
+aggregating and analyzing technical signals from multiple sources.
+
+Usage:
+    python main.py
+
+Requirements:
+    - OpenAI API key set as OPENAI_API_KEY environment variable
+    - Internet connection for fetching live data
+"""
+
+import os
+from typing import List, Dict, Any
+
+# Import adapters
+from adapters.github import fetch_github_trending
+from adapters.arxiv import fetch_arxiv_papers
+from adapters.hackernews import fetch_hackernews_stories
+from adapters.medium import fetch_medium_blogs
+from adapters.huggingface import fetch_huggingface_models
+
+# Import agents
+from agents import (
+    SignalCollectorAgent,
+    RelevanceAgent,
+    RiskAgent,
+    SynthesisAgent
+)
+
+
+def collect_signals() -> List[Dict[str, Any]]:
+    """
+    Collect signals from all configured sources.
+    
+    Returns:
+        Combined list of signals from all adapters.
+    """
+    print("\n📡 [1/4] Collecting Signals...")
+    
+    signals = []
+    
+    # Fetch from each source
+    print("  → Fetching GitHub trending repos...")
+    signals.extend(fetch_github_trending(limit=5))
+    
+    print("  → Fetching ArXiv papers...")
+    signals.extend(fetch_arxiv_papers(limit=5))
+    
+    print("  → Fetching HackerNews stories...")
+    signals.extend(fetch_hackernews_stories(limit=5))
+    
+    print("  → Fetching Medium blogs...")
+    signals.extend(fetch_medium_blogs(limit=3))
+    
+    print("  → Fetching HuggingFace models...")
+    signals.extend(fetch_huggingface_models(limit=5))
+    
+    print(f"  ✓ Collected {len(signals)} raw signals")
+    return signals
+
+
+def run_pipeline():
+    """
+    Execute the full signal intelligence pipeline.
+    
+    Pipeline stages:
+    1. Signal Collection - Aggregate from multiple sources
+    2. Relevance Scoring - Rate signals 0-100
+    3. Risk Assessment - Identify security/breaking changes
+    4. Synthesis - Produce final intelligence digest
+    """
+    print("=" * 60)
+    print("🧠 DevPulseAI - Signal Intelligence Pipeline")
+    print("=" * 60)
+    
+    # Check for API key
+    if not os.environ.get("OPENAI_API_KEY"):
+        print("\n⚠️  Warning: OPENAI_API_KEY not set.")
+        print("   LLM-based agents will use fallback heuristics.\n")
+    
+    # Stage 1: Collection
+    raw_signals = collect_signals()
+    
+    # Initialize agents
+    collector = SignalCollectorAgent()
+    relevance = RelevanceAgent()
+    risk = RiskAgent()
+    synthesis = SynthesisAgent()
+    
+    # Stage 2: Normalize
+    print("\n🔄 [2/4] Normalizing Signals...")
+    normalized = collector.collect(raw_signals)
+    print(f"  ✓ {collector.summarize_collection(normalized)}")
+    
+    # Stage 3: Score for relevance
+    print("\n📊 [3/4] Scoring Relevance...")
+    scored = relevance.score_batch(normalized)
+    high_relevance = sum(1 for s in scored 
+                         if s.get("relevance", {}).get("score", 0) >= 70)
+    print(f"  ✓ {high_relevance}/{len(scored)} signals rated high-relevance")
+    
+    # Stage 4: Assess risks
+    print("\n⚠️  [4/4] Assessing Risks...")
+    assessed = risk.assess_batch(scored)
+    critical = sum(1 for s in assessed 
+                   if s.get("risk", {}).get("risk_level") in ["HIGH", "CRITICAL"])
+    print(f"  ✓ {critical}/{len(assessed)} signals with elevated risk")
+    
+    # Stage 5: Synthesize
+    print("\n📋 Generating Intelligence Digest...")
+    digest = synthesis.synthesize(assessed)
+    
+    # Output results
+    print("\n" + "=" * 60)
+    print("📄 INTELLIGENCE DIGEST")
+    print("=" * 60)
+    print(f"\n🕐 Generated: {digest['generated_at']}")
+    print(f"📦 Total Signals: {digest['total_signals']}")
+    print(f"\n📝 Summary: {digest['executive_summary']}")
+    
+    print("\n🎯 Top Priority Signals:")
+    for i, signal in enumerate(digest.get("priority_signals", [])[:3], 1):
+        score = signal.get("relevance", {}).get("score", "?")
+        risk_level = signal.get("risk", {}).get("risk_level", "?")
+        print(f"   {i}. [{signal['source']}] {signal['title'][:50]}...")
+        print(f"      Relevance: {score} | Risk: {risk_level}")
+    
+    print("\n💡 Recommendations:")
+    for rec in digest.get("recommendations", []):
+        print(f"   • {rec}")
+    
+    print("\n" + "=" * 60)
+    print("✅ Pipeline completed successfully!")
+    print("=" * 60)
+    
+    return digest
+
+
+if __name__ == "__main__":
+    run_pipeline()
@@ -0,0 +1,4 @@
+agno
+httpx
+openai
+feedparser
@@ -0,0 +1,207 @@
+"""
+DevPulseAI Verification Script
+
+This script verifies the pipeline works correctly using MOCK DATA ONLY.
+No network calls or API keys are required.
+
+Usage:
+    python verify.py
+
+Expected output:
+    [OK] DevPulseAI reference pipeline executed successfully
+"""
+
+from typing import List, Dict, Any
+
+# Mock signal data for verification
+MOCK_SIGNALS = [
+    {
+        "id": "mock-gh-001",
+        "source": "github",
+        "title": "awesome-llm-apps",
+        "description": "A curated collection of awesome LLM apps built with RAG and AI agents.",
+        "url": "https://github.com/Shubhamsaboo/awesome-llm-apps",
+        "metadata": {"stars": 5000, "language": "Python", "topics": ["llm", "ai"]}
+    },
+    {
+        "id": "mock-arxiv-001",
+        "source": "arxiv",
+        "title": "Attention Is All You Need: Revisited",
+        "description": "A comprehensive analysis of transformer architectures and their evolution over the past years.",
+        "url": "https://arxiv.org/abs/2401.00001",
+        "metadata": {"pdf": "https://arxiv.org/pdf/2401.00001", "published": "2024-01-15"}
+    },
+    {
+        "id": "mock-hn-001",
+        "source": "hackernews",
+        "title": "GPT-5 Breaking Changes in API",
+        "description": "OpenAI announces breaking changes to the Chat Completions API.",
+        "url": "https://news.ycombinator.com/item?id=12345",
+        "metadata": {"points": 500, "comments": 200, "author": "techwriter"}
+    },
+    {
+        "id": "mock-medium-001",
+        "source": "medium",
+        "title": "Building Production RAG Systems",
+        "description": "A deep dive into building scalable retrieval-augmented generation pipelines.",
+        "url": "https://medium.com/@techblog/building-rag",
+        "metadata": {"author": "TechBlog", "published": "2024-01-20"}
+    },
+    {
+        "id": "mock-hf-001",
+        "source": "huggingface",
+        "title": "HF Model: meta-llama/Llama-3-8B",
+        "description": "Pipeline: text-generation | Downloads: 1,000,000 | Likes: 5,000",
+        "url": "https://huggingface.co/meta-llama/Llama-3-8B",
+        "metadata": {"downloads": 1000000, "likes": 5000, "pipeline_tag": "text-generation"}
+    }
+]
+
+
+def verify_imports():
+    """Verify all modules can be imported."""
+    print("[1/5] Verifying imports...")
+    
+    try:
+        from agents import (
+            SignalCollectorAgent,
+            RelevanceAgent,
+            RiskAgent,
+            SynthesisAgent
+        )
+        from adapters.github import fetch_github_trending
+        from adapters.arxiv import fetch_arxiv_papers
+        from adapters.hackernews import fetch_hackernews_stories
+        from adapters.medium import fetch_medium_blogs
+        from adapters.huggingface import fetch_huggingface_models
+        print("  ✓ All modules imported successfully")
+        return True
+    except ImportError as e:
+        print(f"  ✗ Import error: {e}")
+        return False
+
+
+def verify_signal_collector():
+    """Verify SignalCollectorAgent works with mock data."""
+    print("[2/5] Verifying Signal Collector...")
+    
+    from agents import SignalCollectorAgent
+    
+    collector = SignalCollectorAgent()
+    normalized = collector.collect(MOCK_SIGNALS)
+    
+    assert len(normalized) == len(MOCK_SIGNALS), "Signal count mismatch"
+    assert all("collected_at" in s for s in normalized), "Missing timestamp"
+    
+    summary = collector.summarize_collection(normalized)
+    print(f"  ✓ {summary}")
+    return normalized
+
+
+def verify_relevance_agent(signals: List[Dict]):
+    """Verify RelevanceAgent works with mock data."""
+    print("[3/5] Verifying Relevance Agent...")
+    
+    from agents import RelevanceAgent
+    
+    # Use fallback mode (no API key needed)
+    agent = RelevanceAgent()
+    
+    scored = []
+    for signal in signals:
+        # Use fallback scoring directly
+        result = agent._fallback_score(signal, "Mock mode")
+        signal_with_score = {**signal, "relevance": result}
+        scored.append(signal_with_score)
+    
+    assert all("relevance" in s for s in scored), "Missing relevance scores"
+    print(f"  ✓ Scored {len(scored)} signals")
+    return scored
+
+
+def verify_risk_agent(signals: List[Dict]):
+    """Verify RiskAgent works with mock data."""
+    print("[4/5] Verifying Risk Agent...")
+    
+    from agents import RiskAgent
+    
+    agent = RiskAgent()
+    
+    assessed = []
+    for signal in signals:
+        # Use fallback assessment directly
+        result = agent._fallback_assessment(signal, "Mock mode")
+        signal_with_risk = {**signal, "risk": result}
+        assessed.append(signal_with_risk)
+    
+    assert all("risk" in s for s in assessed), "Missing risk assessments"
+    
+    # Check that breaking change detection works
+    breaking = [s for s in assessed if s.get("risk", {}).get("breaking_changes")]
+    print(f"  ✓ Assessed {len(assessed)} signals ({len(breaking)} with breaking changes)")
+    return assessed
+
+
+def verify_synthesis_agent(signals: List[Dict]):
+    """Verify SynthesisAgent produces valid digest."""
+    print("[5/5] Verifying Synthesis Agent...")
+    
+    from agents import SynthesisAgent
+    
+    agent = SynthesisAgent()
+    digest = agent.synthesize(signals)
+    
+    assert "generated_at" in digest, "Missing timestamp"
+    assert "executive_summary" in digest, "Missing summary"
+    assert "recommendations" in digest, "Missing recommendations"
+    assert digest["total_signals"] == len(signals), "Signal count mismatch"
+    
+    print(f"  ✓ Generated digest with {len(digest['recommendations'])} recommendations")
+    return digest
+
+
+def run_verification():
+    """Run complete verification suite."""
+    print("=" * 60)
+    print("🔍 DevPulseAI Verification Suite")
+    print("=" * 60)
+    print("\nUsing MOCK DATA - No network calls or API keys required.\n")
+    
+    try:
+        # Step 1: Import verification
+        if not verify_imports():
+            raise AssertionError("Import verification failed")
+        
+        # Step 2: Signal collection
+        normalized = verify_signal_collector()
+        
+        # Step 3: Relevance scoring
+        scored = verify_relevance_agent(normalized)
+        
+        # Step 4: Risk assessment
+        assessed = verify_risk_agent(scored)
+        
+        # Step 5: Synthesis
+        digest = verify_synthesis_agent(assessed)
+        
+        # Final summary
+        print("\n" + "=" * 60)
+        print("📊 Verification Summary")
+        print("=" * 60)
+        print(f"  • Signals processed: {digest['total_signals']}")
+        print(f"  • Summary: {digest['executive_summary']}")
+        print(f"  • Recommendations: {len(digest['recommendations'])}")
+        
+        print("\n" + "=" * 60)
+        print("[OK] DevPulseAI reference pipeline executed successfully")
+        print("=" * 60)
+        return True
+        
+    except Exception as e:
+        print(f"\n[FAIL] Verification failed: {e}")
+        return False
+
+
+if __name__ == "__main__":
+    success = run_verification()
+    exit(0 if success else 1)
@@ -0,0 +1,254 @@
+{
+    "name": "Signal Intelligence Ingestion Pipeline",
+    "description": "Optional n8n workflow for automating the DevPulseAI signal intelligence pipeline. Import into n8n to schedule daily digest generation.",
+    "nodes": [
+        {
+            "parameters": {},
+            "id": "trigger-cron",
+            "name": "Daily Trigger",
+            "type": "n8n-nodes-base.scheduleTrigger",
+            "typeVersion": 1.1,
+            "position": [
+                0,
+                300
+            ],
+            "notes": "Triggers the pipeline daily at configured time"
+        },
+        {
+            "parameters": {
+                "httpMethod": "POST",
+                "path": "trigger-pulse",
+                "responseMode": "responseNode"
+            },
+            "id": "webhook-trigger",
+            "name": "Webhook Trigger",
+            "type": "n8n-nodes-base.webhook",
+            "typeVersion": 2,
+            "position": [
+                0,
+                500
+            ],
+            "webhookId": "devpulse-trigger",
+            "notes": "Manual trigger via POST request"
+        },
+        {
+            "parameters": {
+                "url": "https://api.github.com/search/repositories?q=stars:>1000&sort=stars&order=desc&per_page=10",
+                "options": {
+                    "response": {
+                        "response": {
+                            "responseFormat": "json"
+                        }
+                    }
+                }
+            },
+            "id": "github-adapter",
+            "name": "GitHub Trending Repos",
+            "type": "n8n-nodes-base.httpRequest",
+            "typeVersion": 4.2,
+            "position": [
+                400,
+                100
+            ],
+            "notes": "Fetches trending GitHub repositories"
+        },
+        {
+            "parameters": {
+                "url": "http://export.arxiv.org/api/query?search_query=cat:cs.AI+OR+cat:cs.LG&sortBy=submittedDate&sortOrder=descending&max_results=10"
+            },
+            "id": "arxiv-adapter",
+            "name": "ArXiv Papers",
+            "type": "n8n-nodes-base.httpRequest",
+            "typeVersion": 4.2,
+            "position": [
+                400,
+                400
+            ],
+            "notes": "Fetches latest AI/ML research papers"
+        },
+        {
+            "parameters": {
+                "url": "https://hn.algolia.com/api/v1/search_by_date?query=AI&tags=story&hitsPerPage=10"
+            },
+            "id": "hackernews-adapter",
+            "name": "HackerNews Top Stories",
+            "type": "n8n-nodes-base.httpRequest",
+            "typeVersion": 4.2,
+            "position": [
+                400,
+                550
+            ],
+            "notes": "Fetches top HackerNews stories"
+        },
+        {
+            "parameters": {
+                "mode": "multiplex"
+            },
+            "id": "merge-signals",
+            "name": "Aggregate Signals",
+            "type": "n8n-nodes-base.merge",
+            "typeVersion": 3,
+            "position": [
+                650,
+                400
+            ],
+            "notes": "Combines all signal sources into unified stream"
+        },
+        {
+            "parameters": {
+                "resource": "chat",
+                "model": "gpt-4o-mini",
+                "prompt": {
+                    "messages": [
+                        {
+                            "role": "system",
+                            "content": "You are a Relevance Scoring Agent. Score the following content from 0-100 based on its relevance to AI/ML developers. Return ONLY a JSON object: {\"score\": <number>, \"reason\": \"<1 sentence>\"}"
+                        },
+                        {
+                            "role": "user",
+                            "content": "Title: {{ $json.title }}\nDescription: {{ $json.description }}"
+                        }
+                    ]
+                },
+                "options": {
+                    "temperature": 0.1,
+                    "maxOutputTokens": 100
+                }
+            },
+            "id": "relevance-agent",
+            "name": "Relevance Agent",
+            "type": "@n8n/n8n-nodes-langchain.lmChatOpenAi",
+            "typeVersion": 1,
+            "position": [
+                1100,
+                400
+            ],
+            "notes": "Scores content 0-100 based on developer relevance"
+        },
+        {
+            "parameters": {
+                "resource": "chat",
+                "model": "gpt-4o-mini",
+                "prompt": {
+                    "messages": [
+                        {
+                            "role": "system",
+                            "content": "You are a Risk Assessment Agent. Analyze for: breaking changes, security vulnerabilities, or deprecations. Return ONLY a JSON object: {\"risk_level\": \"HIGH|MEDIUM|LOW\", \"concerns\": [\"<list>\"]}"
+                        },
+                        {
+                            "role": "user",
+                            "content": "Title: {{ $json.title }}\nDescription: {{ $json.description }}"
+                        }
+                    ]
+                },
+                "options": {
+                    "temperature": 0.1,
+                    "maxOutputTokens": 150
+                }
+            },
+            "id": "risk-agent",
+            "name": "Risk Agent",
+            "type": "@n8n/n8n-nodes-langchain.lmChatOpenAi",
+            "typeVersion": 1,
+            "position": [
+                1100,
+                600
+            ],
+            "notes": "Flags breaking changes and security vulnerabilities"
+        }
+    ],
+    "connections": {
+        "Daily Trigger": {
+            "main": [
+                [
+                    {
+                        "node": "GitHub Trending Repos",
+                        "type": "main",
+                        "index": 0
+                    }
+                ]
+            ]
+        },
+        "Webhook Trigger": {
+            "main": [
+                [
+                    {
+                        "node": "GitHub Trending Repos",
+                        "type": "main",
+                        "index": 0
+                    }
+                ]
+            ]
+        },
+        "GitHub Trending Repos": {
+            "main": [
+                [
+                    {
+                        "node": "Aggregate Signals",
+                        "type": "main",
+                        "index": 0
+                    }
+                ]
+            ]
+        },
+        "ArXiv Papers": {
+            "main": [
+                [
+                    {
+                        "node": "Aggregate Signals",
+                        "type": "main",
+                        "index": 1
+                    }
+                ]
+            ]
+        },
+        "HackerNews Top Stories": {
+            "main": [
+                [
+                    {
+                        "node": "Aggregate Signals",
+                        "type": "main",
+                        "index": 2
+                    }
+                ]
+            ]
+        },
+        "Aggregate Signals": {
+            "main": [
+                [
+                    {
+                        "node": "Relevance Agent",
+                        "type": "main",
+                        "index": 0
+                    },
+                    {
+                        "node": "Risk Agent",
+                        "type": "main",
+                        "index": 0
+                    }
+                ]
+            ]
+        }
+    },
+    "settings": {
+        "executionOrder": "v1"
+    },
+    "tags": [
+        {
+            "name": "signal-intelligence"
+        },
+        {
+            "name": "automation"
+        },
+        {
+            "name": "developer-tools"
+        },
+        {
+            "name": "ai-agents"
+        }
+    ],
+    "meta": {
+        "templateCredsSetupCompleted": false,
+        "instanceId": "devpulse-reference"
+    }
+}