github-starred/awesome-llm-apps
2
0
Fork 0
mirror of https://github.com/Shubhamsaboo/awesome-llm-apps.git synced 2026-03-11 17:48:31 -05:00
Code Issues 14 Packages Projects Releases Wiki Activity

refactor(devpulse): clarify agent boundaries, simplify models, fix deps

Browse Source
This commit is contained in:
STiFLeR7
2026-02-17 14:13:12 +05:30
parent f26a05d2a7
commit 076c9dd2d1
9 changed files with 620 additions and 548 deletions
Download Patch File Download Diff File Expand all files Collapse all files

284
advanced_ai_agents/multi_agent_apps/devpulse_ai/README.md
View File

@@ -1,115 +1,104 @@
## 🧠 DevPulseAI - Multi-Agent Signal Intelligence Pipeline
# 🧠 DevPulseAI — Multi-Agent Signal Intelligence
A reference implementation demonstrating a **multi-agent system** for aggregating, analyzing, and synthesizing technical signals from multiple developer-focused sources.
A reference implementation demonstrating how to build a **multi-agent pipeline** that aggregates technical signals from multiple sources, scores them for relevance, assesses risks, and synthesizes an actionable intelligence digest.
### Features
> **Design Philosophy:** Agents are used **only where reasoning is required.** Deterministic operations (collection, normalization, deduplication) are implemented as plain utilities — not agents.
- **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
## 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 │ │
│ └─────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────┐
│ DATA SOURCES │
│ GitHub · ArXiv · HackerNews · Medium · HuggingFace │
└──────────────────────┬──────────────────────────────────┘
│ raw signals
▼
┌──────────────────────────────────────────────────────────┐
│ SignalCollector (UTILITY — no LLM) │
│ • Normalizes to unified schema │
│ • Deduplicates via source:id composite key │
│ • Filters incomplete signals │
└──────────────────────┬───────────────────────────────────┘
│ normalized signals
▼
┌──────────────────────────────────────────────────────────┐
│ RelevanceAgent (AGENT — gpt-4.1-mini) │
│ • Scores each signal 0–100 for developer relevance │
│ • Considers: novelty, impact, actionability, timeliness │
│ • Falls back to heuristics if no API key │
└──────────────────────┬───────────────────────────────────┘
│ scored signals
▼
┌──────────────────────────────────────────────────────────┐
│ RiskAgent (AGENT — gpt-4.1-mini) │
│ • Assesses security vulnerabilities │
│ • Flags breaking changes and deprecations │
│ • Rates risk: LOW / MEDIUM / HIGH / CRITICAL │
└──────────────────────┬───────────────────────────────────┘
│ risk-assessed signals
▼
┌──────────────────────────────────────────────────────────┐
│ SynthesisAgent (AGENT — gpt-4.1) │
│ • Cross-references relevance + risk data │
│ • Produces executive summary │
│ • Generates actionable recommendations │
└──────────────────────┬───────────────────────────────────┘
│
▼
📄 Intelligence 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 |
## Why Signal Collection Is Not an Agent
### How to Get Started
This is an **intentional, opinionated design choice** — not a shortcut.
1. Clone the repository
Signal collection involves:
- Fetching data from HTTP APIs (deterministic)
- Normalizing fields to a unified schema (mechanical transformation)
- Deduplicating by composite key (hash comparison)
**None of these tasks require reasoning, judgment, or language understanding.**
Wrapping collection in an `Agent` class would be _decorative_ — it would have an LLM import that never gets called. This misleads readers into thinking an LLM is necessary, when the actual logic is a `for` loop with a `set()`.
> **Rule of thumb:** If you can write the logic as a pure function with no ambiguity, it's a utility. If the output depends on understanding context, making judgment calls, or generating natural language, it's an agent.
---
## Agent Roles & Model Selection
| Component | Type | Model | Why This Model |
|---|---|---|---|
| `SignalCollector` | **Utility** | _none_ | Deterministic — no reasoning required |
| `RelevanceAgent` | **Agent** | `gpt-4.1-mini` | Classification task — fast, cheap, high-volume |
| `RiskAgent` | **Agent** | `gpt-4.1-mini` | Structured analysis — careful but not expensive |
| `SynthesisAgent` | **Agent** | `gpt-4.1` | Cross-referencing & summarization — needs strongest reasoning |
**Single provider by default (OpenAI)** to reduce onboarding friction. Override per-agent via environment variables:
```bash
export MODEL_RELEVANCE=gpt-4.1-nano # cheaper, faster
export MODEL_RISK=o4-mini # deeper reasoning for risk
export MODEL_SYNTHESIS=gpt-4.1 # default, strongest
```
---
## How to Run
### Quick Verification (No API Key Required)
```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 Gemini API key (optional for live mode)
```bash
export GOOGLE_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
```
### Streamlit Demo
A modern, interactive dashboard is included to visualize the multi-agent pipeline:
1. Launch the app:
```bash
streamlit run streamlit_app.py
```
1. Configure sources and signal counts in the sidebar.
2. Provide a Gemini API key (optional) to use full LLM intelligence.
3. View real-time progress as agents collaborate.
> **Note**: The default configuration is optimized for fast demo runs.
### 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
```
This runs the full pipeline with mock data in **<1 second**. No network calls, no API keys.
Expected output:
@@ -117,51 +106,90 @@ Expected output:
[OK] DevPulseAI reference pipeline executed successfully
```
### Optional: n8n Automation
### Full Pipeline (With API Key)
An n8n workflow is included for those who want to automate the pipeline:
```bash
pip install -r requirements.txt
export OPENAI_API_KEY=sk-...
python main.py
```
- **Location**: `workflows/signal-intelligence-pipeline.json`
- **Import**: n8n → Settings → Import from File
- **Requires**: n8n instance + configured credentials
Without an API key, agents automatically fall back to heuristic scoring.
This is entirely optional - the Python implementation works standalone.
### Streamlit Dashboard
### Directory Structure
```bash
streamlit run streamlit_app.py
```
---
## Project 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
│ ├── __init__.py # Package exports + design docs
│ ├── signal_collector.py # UTILITY — normalize & dedup
│ ├── relevance_agent.py # AGENT — score relevance (gpt-4.1-mini)
│ ├── risk_agent.py # AGENT — assess risks (gpt-4.1-mini)
│ └── synthesis_agent.py # AGENT — produce digest (gpt-4.1)
├── adapters/
│ ├── github.py # GitHub trending repos
│ ├── arxiv.py # ArXiv recent papers
│ ├── hackernews.py # HackerNews top stories
│ ├── medium.py # Medium AI/ML blogs
│ └── huggingface.py # HuggingFace trending models
├── workflows/
│ └── signal-intelligence-pipeline.json
├── main.py # Full pipeline demo (CLI)
├── streamlit_app.py # Interactive dashboard (UI)
├── verify.py # Mock data verification
├── requirements.txt
└── README.md
├── main.py # Full pipeline runner
├── verify.py # Mock-data verification (<1s)
├── streamlit_app.py # Interactive dashboard
└── requirements.txt # Minimal deps (single provider)
```
### 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
## Optional Extensions (Advanced Users)
### Built With
These are **not required** for the reference implementation, but show how the architecture extends:
- [Agno](https://github.com/agno-agi/agno) - Multi-agent framework
- [Google Gemini 1.5 Flash](https://ai.google.dev/) - LLM backbone
- [httpx](https://www.python-httpx.org/) - Async HTTP client
1. **Multi-provider models** — Swap `RelevanceAgent` to use Anthropic Claude or Google Gemini by updating the model config. The `agno` framework supports multiple providers.
2. **Vector search** — Add a Pinecone or Qdrant adapter to store and retrieve signals semantically for long-term pattern detection.
3. **Streaming digests** — Use WebSocket streaming from `SynthesisAgent` for real-time intelligence feeds.
4. **Custom adapters** — Add new signal sources by implementing a `fetch_*` function that returns `List[Dict]` with the standard schema (`id`, `source`, `title`, `description`, `url`, `metadata`).
5. **Feedback loop** — Store user feedback (👍/👎) in Supabase and use it to fine-tune relevance scoring over time.
---
## Dependencies
```
agno # Agent framework
openai # LLM provider (single default)
httpx # HTTP client for adapters
feedparser # RSS/Atom parsing for Medium
streamlit>=1.30 # Interactive dashboard
```
No `google-generativeai` required. Gemini is an optional extension if users want multi-provider support — install `google-genai` (not the deprecated `google-generativeai`) separately.
---
## Design Tradeoffs
| Decision | Tradeoff | Why |
|---|---|---|
| Single provider default | Less flexibility | Reduces onboarding from 2+ keys to 1 |
| Signal collection as utility | Less "agentic" demo | Honest architecture — agents where reasoning exists |
| Heuristic fallbacks | Lower quality without API key | Pipeline always works, even for evaluation |
| 5 signals per source default | Less data | Keeps demo fast (<10s with API, <1s mock) |
| No async in agents | Less throughput | Simpler code, clearer educational value |
---
_Built as a reference implementation for [awesome-llm-apps](https://github.com/Shubhamsaboo/awesome-llm-apps)._

30
advanced_ai_agents/multi_agent_apps/devpulse_ai/agents/__init__.py
View File

@@ -1,21 +1,33 @@
"""
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
This package contains the intelligence pipeline components:
- SignalCollector: Pure utility (NOT an agent) — normalizes and deduplicates signals.
Signal collection is deterministic and intentionally not agent-driven.
- RelevanceAgent: LLM agent — scores signals 0-100 for developer relevance.
Uses fast model (gpt-4.1-mini) for high-throughput classification.
- RiskAgent: LLM agent — assesses security risks and breaking changes.
Uses structured-reasoning model (gpt-4.1-mini) for careful analysis.
- SynthesisAgent: LLM agent — produces final intelligence digest.
Uses strongest model (gpt-4.1) for cross-referencing and summarization.
Design Principle:
Agents are used ONLY where reasoning is required.
Deterministic operations (collection, normalization, dedup) are utilities.
"""
from .signal_collector import SignalCollectorAgent
from .signal_collector import SignalCollector
from .relevance_agent import RelevanceAgent
from .risk_agent import RiskAgent
from .synthesis_agent import SynthesisAgent
__all__ = [
"SignalCollectorAgent",
"RelevanceAgent",
"SignalCollector",
"RelevanceAgent",
"RiskAgent",
"SynthesisAgent"
"SynthesisAgent",
]

121
advanced_ai_agents/multi_agent_apps/devpulse_ai/agents/relevance_agent.py
View File

@@ -1,58 +1,70 @@
"""
Relevance Agent - Scores signals based on developer relevance.
Relevance Agent — Scores signals by developer relevance (0–100).
This agent uses LLM reasoning to score each signal from 0-100
based on its relevance to AI/ML developers and engineers.
This agent uses LLM reasoning to evaluate each signal's importance to
AI/ML developers. It's a legitimate agent because relevance scoring
requires judgment, context understanding, and nuanced assessment that
pure heuristics cannot capture.
Model Selection:
Uses a fast, cost-efficient model (gpt-4.1-mini by default) because
relevance scoring is high-volume and doesn't require deep reasoning —
it's a classification task, not a synthesis task.
"""
from typing import Dict, Any, Optional
import json
from typing import Dict, Any, List
from agno.agent import Agent
from agno.models.google import Gemini
from agno.models.openai import OpenAIChat
# Central model config — override via MODEL_RELEVANCE env var
import os
DEFAULT_MODEL = os.environ.get("MODEL_RELEVANCE", "gpt-4.1-mini")
class RelevanceAgent:
"""
Agent that scores signals based on relevance to developers.
Why this IS an agent (unlike SignalCollector):
Relevance scoring requires understanding context, assessing novelty,
and making judgment calls about what matters to developers. This is
inherently a reasoning task that benefits from LLM capabilities.
Responsibilities:
- Score signals 0-100 based on developer relevance
- Score signals 0–100 based on developer relevance
- Provide reasoning for each score
- Prioritize actionable, timely content
- Gracefully fall back to heuristics when LLM unavailable
"""
def __init__(self, model_id: str = "gemini-1.5-flash"):
def __init__(self, model_id: str = None):
"""
Initialize the Relevance Agent.
Args:
model_id: Model ID to use for scoring.
model_id: OpenAI model to use. Defaults to gpt-4.1-mini (fast, cheap).
"""
self.model_id = model_id
self.model_id = model_id or DEFAULT_MODEL
self.agent = Agent(
name="Relevance Scorer",
model=Gemini(id=model_id),
model=OpenAIChat(id=self.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."
"Provide brief reasoning for each score.",
],
markdown=True
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.
Score a single signal for developer relevance.
Attempts LLM scoring first, falls back to heuristics on failure.
"""
prompt = f"""
Rate the relevance of this signal for AI/ML developers.
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)
@@ -65,57 +77,48 @@ Signal:
- Description: {signal.get('description', '')[:500]}
Respond with ONLY a JSON object:
{{"score": <number>, "reasoning": "<one sentence>"}}
"""
{{"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.
"""
def score_batch(self, signals: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Score multiple signals, returning each with a 'relevance' key."""
scored = []
for signal in signals:
result = self.score(signal)
signal_with_score = {**signal, "relevance": result}
scored.append(signal_with_score)
scored.append({**signal, "relevance": result})
return scored
def _parse_response(self, content: str, signal: Dict) -> Dict[str, Any]:
"""Parse LLM response into structured output."""
import json
"""Parse LLM JSON response into structured output."""
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:
text = content.strip()
if "```" in text:
text = text.split("```")[1].replace("json", "").strip()
return json.loads(text)
except (json.JSONDecodeError, IndexError):
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
"""
Heuristic fallback when LLM is unavailable.
Uses metadata signals (stars, points) as rough relevance proxies.
This is intentionally simple — the LLM path is the real logic.
"""
score = 50 # Default: moderate
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})"
"reasoning": f"Heuristic score (LLM unavailable: {error})",
}

126
advanced_ai_agents/multi_agent_apps/devpulse_ai/agents/risk_agent.py
View File

@@ -1,63 +1,74 @@
"""
Risk Agent - Assesses security risks and breaking changes.
Risk Agent — Assesses security risks and breaking changes.
This agent analyzes signals for potential risks including:
- Security vulnerabilities
- Breaking changes in dependencies
- Deprecation notices
This agent uses LLM reasoning to analyze signals for potential risks
including security vulnerabilities, breaking API changes, and deprecation
notices. It's a legitimate agent because risk assessment requires
contextual understanding of technical implications.
Model Selection:
Uses a structured-reasoning model (gpt-4.1-mini by default) because
risk assessment benefits from careful, step-by-step analysis. For
production workloads with high-stakes decisions, consider upgrading
to gpt-4.1 or o4-mini via the MODEL_RISK environment variable.
"""
import json
from typing import Dict, Any, List
from agno.agent import Agent
from agno.models.google import Gemini
from agno.models.openai import OpenAIChat
# Central model config — override via MODEL_RISK env var
import os
DEFAULT_MODEL = os.environ.get("MODEL_RISK", "gpt-4.1-mini")
class RiskAgent:
"""
Agent that assesses risk levels in technical signals.
Why this IS an agent:
Risk assessment requires reasoning about technical implications,
understanding security contexts, and making judgment calls about
severity. This is inherently a reasoning task.
Responsibilities:
- Identify security vulnerabilities
- Flag breaking changes
- Detect deprecation notices
- Rate overall risk level
- Rate overall risk level (LOW / MEDIUM / HIGH / CRITICAL)
"""
RISK_LEVELS = ["LOW", "MEDIUM", "HIGH", "CRITICAL"]
def __init__(self, model_id: str = "gemini-1.5-flash"):
def __init__(self, model_id: str = None):
"""
Initialize the Risk Agent.
Args:
model_id: Model ID to use for risk assessment.
model_id: OpenAI model to use. Defaults to gpt-4.1-mini.
"""
self.model_id = model_id
self.model_id = model_id or DEFAULT_MODEL
self.agent = Agent(
name="Risk Assessor",
model=Gemini(id=model_id),
model=OpenAIChat(id=self.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."
"Rate risk level: LOW, MEDIUM, HIGH, or CRITICAL.",
],
markdown=True
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.
Attempts LLM assessment first, falls back to keyword heuristics.
"""
prompt = f"""
Analyze this technical signal for risks:
prompt = f"""Analyze this technical signal for risks:
Signal:
- Source: {signal.get('source', 'unknown')}
@@ -70,64 +81,57 @@ Assess for:
3. Deprecations
Respond with ONLY a JSON object:
{{"risk_level": "LOW|MEDIUM|HIGH|CRITICAL", "concerns": ["<list of concerns>"], "breaking_changes": true|false}}
"""
{{"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.
"""
def assess_batch(self, signals: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Assess multiple signals, returning each with a 'risk' key."""
assessed = []
for signal in signals:
result = self.assess(signal)
signal_with_risk = {**signal, "risk": result}
assessed.append(signal_with_risk)
assessed.append({**signal, "risk": result})
return assessed
def _parse_response(self, content: str, signal: Dict) -> Dict[str, Any]:
"""Parse LLM response into structured output."""
import json
"""Parse LLM JSON response into structured output."""
try:
content = content.strip()
if "```" in content:
content = content.split("```")[1].replace("json", "").strip()
return json.loads(content)
except:
text = content.strip()
if "```" in text:
text = text.split("```")[1].replace("json", "").strip()
return json.loads(text)
except (json.JSONDecodeError, IndexError):
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."""
"""
Keyword-based fallback when LLM is unavailable.
Simple heuristic: scan title for risk-indicating keywords.
This is intentionally conservative — better to over-flag than miss.
"""
title = signal.get("title", "").lower()
# Simple keyword-based heuristics
risk_level = "LOW"
concerns = []
risk_keywords = {
"HIGH": ["vulnerability", "exploit", "CVE", "critical", "breach"],
"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):
if any(kw 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()
"concerns": concerns or [f"Heuristic (LLM unavailable: {error})"],
"breaking_changes": "breaking" in title,
}

107
advanced_ai_agents/multi_agent_apps/devpulse_ai/agents/signal_collector.py
View File

@@ -1,100 +1,79 @@
"""
Signal Collector Agent - Aggregates signals from multiple data sources.
Signal Collector — Pure utility (not an agent).
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.
Signal collection is deterministic and intentionally not agent-driven.
This module aggregates signals from adapters, normalizes them to a unified
schema, and deduplicates deterministically. No LLM reasoning is involved
because collection/normalization is a mechanical transformation — using an
agent here would be decorative, not functional.
Design Decision:
Agents are used only where reasoning is required. Signal collection
involves no ambiguity, judgment, or language understanding — it's a
pipeline transformation. Wrapping it in an Agent class would mislead
readers into thinking an LLM call is necessary here.
"""
from typing import List, Dict, Any
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from datetime import datetime, timezone
class SignalCollectorAgent:
class SignalCollector:
"""
Agent that collects and normalizes signals from multiple sources.
Utility that collects and normalizes signals from multiple sources.
NOT an agent — no LLM calls. This is an intentional design choice.
See module docstring for rationale.
Responsibilities:
- Fetch data from configured adapters
- Normalize signals to unified schema
- Deduplicate and filter low-quality signals
- Deduplicate deterministically (source:id composite key)
- Filter incomplete 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.
Normalize and deduplicate raw signals from adapters.
Args:
signals: Raw signals from adapters.
signals: Raw signals from adapters (heterogeneous schemas).
Returns:
List of normalized signal dictionaries.
List of normalized, deduplicated signal dictionaries.
"""
normalized = []
seen_ids = set()
for signal in signals:
# Deduplicate
# Deterministic dedup key: source + external id
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 = {
# Normalize to unified schema
normalized.append({
"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)
"collected_at": datetime.now(timezone.utc).isoformat(),
})
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.
Generate a human-readable collection summary.
Pure string formatting — no LLM needed.
"""
sources = {}
sources: Dict[str, int] = {}
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)}"
parts = [f"{count} from {src}" for src, count in sources.items()]
return f"Collected {len(signals)} signals: {', '.join(parts)}"

171
advanced_ai_agents/multi_agent_apps/devpulse_ai/agents/synthesis_agent.py
View File

@@ -1,154 +1,167 @@
"""
Synthesis Agent - Produces final intelligence digest.
Synthesis Agent — Produces final intelligence digest.
This agent combines outputs from all previous agents to create
a comprehensive, actionable intelligence summary for developers.
This agent combines outputs from all previous stages to create a
comprehensive, actionable intelligence summary. It's the most reasoning-
intensive agent in the pipeline and uses the strongest available model.
Model Selection:
Uses gpt-4.1 by default — the strongest reasoning model — because
synthesis requires cross-referencing multiple signals, identifying
patterns, generating executive summaries, and producing actionable
recommendations. This is the one stage where model quality directly
impacts output quality.
Override via MODEL_SYNTHESIS env var for cost optimization.
"""
from typing import Dict, Any, List
from datetime import datetime, timezone
from agno.agent import Agent
from agno.models.google import Gemini
from agno.models.openai import OpenAIChat
# Central model config — override via MODEL_SYNTHESIS env var
import os
DEFAULT_MODEL = os.environ.get("MODEL_SYNTHESIS", "gpt-4.1")
class SynthesisAgent:
"""
Agent that synthesizes all signal intelligence into a final digest.
Why this uses the strongest model:
Synthesis is the most reasoning-intensive task in the pipeline.
It must cross-reference relevance scores, risk assessments, and
source metadata to produce coherent, actionable intelligence.
Using a weaker model here would produce generic, shallow outputs.
Responsibilities:
- Combine relevance and risk assessments
- Prioritize signals by importance
- Generate executive summary
- Produce actionable recommendations
"""
def __init__(self, model_id: str = "gemini-1.5-flash"):
def __init__(self, model_id: str = None):
"""
Initialize the Synthesis Agent.
Args:
model_id: Model ID to use for synthesis.
model_id: OpenAI model to use. Defaults to gpt-4.1 (strongest reasoning).
"""
self.model_id = model_id
self.model_id = model_id or DEFAULT_MODEL
self.agent = Agent(
name="Intelligence Synthesizer",
model=Gemini(id=model_id),
model=OpenAIChat(id=self.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."
"Provide actionable recommendations for developers.",
],
markdown=True
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.
This method uses deterministic logic for prioritization and grouping,
then delegates summary generation to either LLM or heuristics.
"""
# 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(),
"generated_at": datetime.now(timezone.utc).isoformat(),
"total_signals": len(signals),
"executive_summary": summary,
"priority_signals": prioritized[:5], # Top 5
"priority_signals": prioritized[:5],
"signals_by_source": grouped,
"recommendations": self._generate_recommendations(prioritized)
"recommendations": self._generate_recommendations(prioritized),
}
def _prioritize_signals(self, signals: List[Dict]) -> List[Dict]:
"""Sort signals by priority score."""
"""Sort signals by composite priority score (relevance × risk multiplier)."""
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)
"LOW": 0.8,
}.get(signal.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 = {}
"""Group signals by their source for categorized display."""
grouped: Dict[str, List] = {}
for signal in signals:
source = signal.get("source", "unknown")
if source not in grouped:
grouped[source] = []
grouped[source].append(signal)
grouped.setdefault(source, []).append(signal)
return grouped
def _generate_summary(self, signals: List[Dict]) -> str:
"""Generate executive summary."""
"""Generate executive summary from processed signals."""
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"]]
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.")
parts.append(
f"⚠️ {len(critical_risks)} signals with elevated risk."
)
if signals:
top = signals[0]
parts.append(f"Top signal: {top.get('title', 'Unknown')}")
parts.append(f"Top signal: {signals[0].get('title', 'Unknown')}")
return " ".join(parts)
def _generate_recommendations(self, signals: List[Dict]) -> List[str]:
"""Generate actionable recommendations."""
"""Generate actionable recommendations from signal analysis."""
recommendations = []
critical_risks = [s for s in signals
if s.get("risk", {}).get("risk_level") == "CRITICAL"]
if critical_risks:
critical = [
s
for s in signals
if s.get("risk", {}).get("risk_level") == "CRITICAL"
]
if critical:
recommendations.append(
f"🚨 Review {len(critical_risks)} critical-risk signals immediately"
f"🚨 Review {len(critical)} critical-risk signals immediately"
)
high_relevance = [s for s in signals
if s.get("relevance", {}).get("score", 0) >= 80]
if high_relevance:
high_rel = [
s for s in signals if s.get("relevance", {}).get("score", 0) >= 80
]
if high_rel:
recommendations.append(
f"📌 Prioritize {len(high_relevance)} high-relevance items"
f"📌 Prioritize {len(high_rel)} high-relevance items"
)
github_signals = [s for s in signals if s.get("source") == "github"]
if github_signals:
github = [s for s in signals if s.get("source") == "github"]
if github:
recommendations.append(
f"⭐ Explore {len(github_signals)} trending repositories"
f"⭐ Explore {len(github)} 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"

132
advanced_ai_agents/multi_agent_apps/devpulse_ai/main.py
View File

@@ -1,22 +1,32 @@
"""
DevPulseAI - Multi-Agent Signal Intelligence Pipeline
DevPulseAI — Multi-Agent Signal Intelligence Pipeline
This script demonstrates a complete multi-agent workflow for
aggregating and analyzing technical signals from multiple sources.
Demonstrates a production-style multi-agent workflow that aggregates
technical signals from multiple sources, scores them for relevance,
assesses risks, and synthesizes an actionable intelligence digest.
Architecture:
Adapters (fetch) → SignalCollector (normalize) → RelevanceAgent (score)
→ RiskAgent (assess) → SynthesisAgent (digest)
Design Decisions:
- Signal collection is a utility, not an agent (deterministic work).
- Agents are used only where reasoning is required.
- Single provider (OpenAI) by default to reduce onboarding friction.
- Models are chosen by role: fast for classification, strong for synthesis.
Usage:
export OPENAI_API_KEY=sk-...
python main.py
Requirements:
- OpenAI API key set as OPENAI_API_KEY environment variable
- Internet connection for fetching live data
Without API key: agents fall back to heuristic scoring.
"""
import os
from typing import List, Dict, Any, Optional
# Reduced default signal count for faster demo execution
DEFAULT_SIGNAL_LIMIT = 8
DEFAULT_SIGNAL_LIMIT = 5
# Import adapters
from adapters.github import fetch_github_trending
@@ -25,46 +35,41 @@ from adapters.hackernews import fetch_hackernews_stories
from adapters.medium import fetch_medium_blogs
from adapters.huggingface import fetch_huggingface_models
# Import agents
# Import pipeline components
from agents import (
SignalCollectorAgent,
RelevanceAgent,
RiskAgent,
SynthesisAgent
SignalCollector, # Utility — no LLM
RelevanceAgent, # Agent — gpt-4.1-mini
RiskAgent, # Agent — gpt-4.1-mini
SynthesisAgent, # Agent — gpt-4.1
)
def collect_signals(limit: Optional[int] = None) -> List[Dict[str, Any]]:
"""
Collect signals from all configured sources.
Args:
limit: Maximum signals to fetch per source. Defaults to DEFAULT_SIGNAL_LIMIT.
Returns:
Combined list of signals from all adapters.
This is pure data aggregation — no LLM involved.
"""
fetch_limit = limit if limit is not None else DEFAULT_SIGNAL_LIMIT
print(f"\n📡 [1/4] Collecting Signals (limit: {fetch_limit} per source)...")
signals = []
# Fetch from each source
print(" → Fetching GitHub trending repos...")
signals.extend(fetch_github_trending(limit=fetch_limit))
print(" → Fetching ArXiv papers...")
signals.extend(fetch_arxiv_papers(limit=fetch_limit))
print(" → Fetching HackerNews stories...")
signals.extend(fetch_hackernews_stories(limit=fetch_limit))
print(" → Fetching Medium blogs...")
signals.extend(fetch_medium_blogs(limit=min(fetch_limit, 3))) # Medium feeds are usually denser
signals.extend(fetch_medium_blogs(limit=min(fetch_limit, 3)))
print(" → Fetching HuggingFace models...")
signals.extend(fetch_huggingface_models(limit=fetch_limit))
print(f" ✓ Collected {len(signals)} raw signals")
return signals
@@ -72,54 +77,57 @@ def collect_signals(limit: Optional[int] = None) -> List[Dict[str, Any]]:
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
1. Signal Collection — Aggregate from sources (utility, no LLM)
2. Normalization — Deduplicate and normalize (utility, no LLM)
3. Relevance Score — Rate signals 0-100 (agent, gpt-4.1-mini)
4. Risk Assessment — Identify risks (agent, gpt-4.1-mini)
5. Synthesis — Produce digest (agent, gpt-4.1)
"""
print("=" * 60)
print("🧠 DevPulseAI - Signal Intelligence Pipeline")
print("🧠 DevPulseAI — Signal Intelligence Pipeline")
print("=" * 60)
# Check for API key
if not os.environ.get("GOOGLE_API_KEY") and not os.environ.get("GEMINI_API_KEY"):
print("\n⚠️ Warning: GOOGLE_API_KEY not set.")
print(" LLM-based agents will use fallback heuristics.\n")
# Stage 1: Collection
if not os.environ.get("OPENAI_API_KEY"):
print("\n⚠️ Warning: OPENAI_API_KEY not set.")
print(" Agents will use fallback heuristics.\n")
# Stage 1: Collect raw signals from adapters
raw_signals = collect_signals()
# Initialize agents
collector = SignalCollectorAgent()
relevance = RelevanceAgent()
risk = RiskAgent()
synthesis = SynthesisAgent()
# Stage 2: Normalize
# Stage 2: Normalize and deduplicate (utility — no LLM)
collector = SignalCollector()
print("\n🔄 [2/4] Normalizing Signals...")
normalized = collector.collect(raw_signals)
print(f" ✓ {collector.summarize_collection(normalized)}")
# Stage 3: Score for relevance
# Stage 3: Score for relevance (agent — gpt-4.1-mini)
relevance = RelevanceAgent()
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)
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
# Stage 4: Assess risks (agent — gpt-4.1-mini)
risk = RiskAgent()
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"])
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
# Stage 5: Synthesize digest (agent — gpt-4.1)
synthesis = SynthesisAgent()
print("\n📋 Generating Intelligence Digest...")
digest = synthesis.synthesize(assessed)
# Output results
print("\n" + "=" * 60)
print("📄 INTELLIGENCE DIGEST")
@@ -127,22 +135,22 @@ def run_pipeline():
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

6
advanced_ai_agents/multi_agent_apps/devpulse_ai/requirements.txt
View File

@@ -1,6 +1,8 @@
# DevPulseAI Dependencies
# Single provider (OpenAI) by default — no multi-provider setup required.
agno
httpx
openai
httpx
feedparser
streamlit>=1.30
google-generativeai

191
advanced_ai_agents/multi_agent_apps/devpulse_ai/verify.py
View File

@@ -1,19 +1,25 @@
"""
DevPulseAI Verification Script
This script verifies the pipeline works correctly using MOCK DATA ONLY.
No network calls or API keys are required.
Verifies the complete pipeline using MOCK DATA ONLY.
No network calls, no API keys, no external dependencies required.
Usage:
python verify.py
Expected output:
[OK] DevPulseAI reference pipeline executed successfully
Runs in <1s on any machine.
"""
import sys
import time
from typing import List, Dict, Any
# Mock signal data for verification
# ────────────────────────────────────────────────
# Mock signal data — representative of real adapter output
# ────────────────────────────────────────────────
MOCK_SIGNALS = [
{
"id": "mock-gh-001",
@@ -21,15 +27,15 @@ MOCK_SIGNALS = [
"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"]}
"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.",
"description": "A comprehensive analysis of transformer architectures.",
"url": "https://arxiv.org/abs/2401.00001",
"metadata": {"pdf": "https://arxiv.org/pdf/2401.00001", "published": "2024-01-15"}
"metadata": {"pdf": "https://arxiv.org/pdf/2401.00001", "published": "2024-01-15"},
},
{
"id": "mock-hn-001",
@@ -37,15 +43,15 @@ MOCK_SIGNALS = [
"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"}
"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.",
"description": "A deep dive into building scalable retrieval-augmented generation.",
"url": "https://medium.com/@techblog/building-rag",
"metadata": {"author": "TechBlog", "published": "2024-01-20"}
"metadata": {"author": "TechBlog", "published": "2024-01-20"},
},
{
"id": "mock-hf-001",
@@ -53,155 +59,172 @@ MOCK_SIGNALS = [
"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"}
}
"metadata": {"downloads": 1000000, "likes": 5000, "pipeline_tag": "text-generation"},
},
]
# ────────────────────────────────────────────────
# Verification steps
# ────────────────────────────────────────────────
def verify_imports():
"""Verify all modules can be imported."""
"""Verify all modules can be imported without errors."""
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
from agents import SignalCollector, 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
# Verify SignalCollector is NOT an agent (no .agent attribute)
collector = SignalCollector()
assert not hasattr(collector, "agent"), (
"SignalCollector should NOT have an .agent attribute — "
"it's a utility, not an agent"
)
print(" ✓ All modules imported successfully")
print(" ✓ SignalCollector confirmed as utility (no LLM)")
return True
def verify_signal_collector():
"""Verify SignalCollectorAgent works with mock data."""
print("[2/5] Verifying Signal Collector...")
from agents import SignalCollectorAgent
collector = SignalCollectorAgent()
"""Verify SignalCollector normalizes and deduplicates correctly."""
print("[2/5] Verifying Signal Collector (utility)...")
from agents import SignalCollector
collector = SignalCollector()
# Test normalization
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"
# Test deduplication — adding duplicates should not increase count
duped = MOCK_SIGNALS + [MOCK_SIGNALS[0]] # One duplicate
deduped = collector.collect(duped)
assert len(deduped) == len(MOCK_SIGNALS), "Deduplication failed"
summary = collector.summarize_collection(normalized)
print(f" ✓ {summary}")
print(" ✓ Deduplication works correctly")
return normalized
def verify_relevance_agent(signals: List[Dict]):
"""Verify RelevanceAgent works with mock data."""
print("[3/5] Verifying Relevance Agent...")
"""Verify RelevanceAgent fallback scoring works without API key."""
print("[3/5] Verifying Relevance Agent (fallback mode)...")
from agents import RelevanceAgent
# Use fallback mode (no API key needed)
agent = RelevanceAgent()
# Use fallback scoring directly (no API key needed)
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)
scored.append({**signal, "relevance": result})
assert all("relevance" in s for s in scored), "Missing relevance scores"
print(f" ✓ Scored {len(scored)} signals")
assert all(
0 <= s["relevance"]["score"] <= 100 for s in scored
), "Scores out of range"
print(f" ✓ Scored {len(scored)} signals (heuristic fallback)")
return scored
def verify_risk_agent(signals: List[Dict]):
"""Verify RiskAgent works with mock data."""
print("[4/5] Verifying Risk Agent...")
"""Verify RiskAgent fallback assessment works without API key."""
print("[4/5] Verifying Risk Agent (fallback mode)...")
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)
assessed.append({**signal, "risk": result})
assert all("risk" in s for s in assessed), "Missing risk assessments"
# Check that breaking change detection works
# Verify keyword detection: "GPT-5 Breaking Changes" should flag
breaking = [s for s in assessed if s.get("risk", {}).get("breaking_changes")]
assert len(breaking) >= 1, "Breaking change detection failed"
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."""
"""Verify SynthesisAgent produces valid digest structure."""
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 "priority_signals" in digest, "Missing priority signals"
assert digest["total_signals"] == len(signals), "Signal count mismatch"
print(f" ✓ Generated digest with {len(digest['recommendations'])} recommendations")
return digest
# ────────────────────────────────────────────────
# Main verification runner
# ────────────────────────────────────────────────
def run_verification():
"""Run complete verification suite."""
"""Run the complete verification suite with timing."""
print("=" * 60)
print("🔍 DevPulseAI Verification Suite")
print("=" * 60)
print("\nUsing MOCK DATA - No network calls or API keys required.\n")
print("\nUsing MOCK DATA — No network calls or API keys required.\n")
start = time.time()
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)
elapsed = time.time() - start
# 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(f" • Signals processed: {digest['total_signals']}")
print(f" • Summary: {digest['executive_summary']}")
print(f" • Recommendations: {len(digest['recommendations'])}")
print(f" • Time elapsed: {elapsed:.3f}s")
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}")
import traceback
traceback.print_exc()
return False
if __name__ == "__main__":
success = run_verification()
exit(0 if success else 1)
sys.exit(0 if success else 1)