Introduction

In the fast-paced world of B2B SaaS and digital services, competitive intelligence is no longer a quarterly report—it’s a real-time necessity. Waiting for a monthly newsletter to learn that a competitor has dropped prices or launched a new feature is too late.

This article details the architecture of a Real-Time Competitor Intel Agent. Unlike simple web scrapers, this system uses LangGraph to orchestrate a multi-agent workflow that:

  1. Monitors diverse data sources (websites, social media, press releases).

  2. Retrieves historical context via RAG to distinguish "noise" from "signal."

  3. Analyzes impact using LLM-based reasoning.

  4. Suggests actionable counter-strategies.

  5. Remembers past interactions to build a longitudinal view of competitor behavior.

Real-Time Use Case: The "Pricing War" in Cloud Security

Scenario

Company: "SecureCloud Inc." (Your Company) Competitor: "ShieldNet" (Main Rival) Event: ShieldNet updates their pricing page and posts a LinkedIn announcement about a "New Enterprise Tier" with a 20% discount for annual contracts.

The Problem

  1. Data Silos: Pricing changes are on the website; announcements are on LinkedIn; technical specs are in press releases.

  2. Context Gap: Is this a permanent price drop or a limited-time offer? How does it compare to our last quarter’s pricing?

  3. Action Lag: By the time sales teams hear about it, prospects may have already signed with ShieldNet.

The Autonomous Solution

The Competitor Intel Agent detects the change, retrieves historical pricing data from its vector store, analyzes the strategic intent, and immediately suggests a counter-campaign: "Launch a 'Value-Over-Price' webinar series highlighting our superior support SLAs."

System Architecture

440

Prerequisites

pip install langgraph langchain langchain-openai chromadb psycopg2-binary redis pydantic beautifulsoup4 requests

Step-by-Step Implementation

Step 1: Define State Schema and Data Models

We need a state that tracks the competitor, the detected change, historical context, and the suggested response.

from typing import TypedDict, List, Optional, Dict, Any
from pydantic import BaseModel, Field
from datetime import datetime
import uuid

class CompetitorChange(BaseModel):
    source_type: str  # "website", "social", "press_release"
    url: str
    content_summary: str
    detected_at: datetime
    change_type: str  # "pricing", "feature", "messaging", "personnel"

class HistoricalContext(BaseModel):
    previous_pricing: Optional[str] = None
    previous_messaging: Optional[str] = None
    last_similar_change: Optional[str] = None

class StrategicAnalysis(BaseModel):
    impact_level: str  # "high", "medium", "low"
    strategic_intent: str
    threat_to_us: str

class CounterStrategy(BaseModel):
    recommended_action: str
    campaign_theme: str
    key_messaging_points: List[str]
    target_channel: str

class IntelState(TypedDict):
    competitor_name: str
    detected_change: Optional[CompetitorChange]
    historical_context: Optional[HistoricalContext]
    analysis: Optional[StrategicAnalysis]
    counter_strategy: Optional[CounterStrategy]
    conversation_history: List[Dict[str, str]]
    error_message: Optional[str]

Step 2: Data Ingestion Agent (Monitoring)

This agent simulates monitoring. In production, this would connect to APIs (LinkedIn, Twitter) or use headless browsers (Playwright) for websites.

import requests
from bs4 import BeautifulSoup

class DataIngestionAgent:
    def monitor_competitor(self, competitor_name: str, source_url: str) -> CompetitorChange:
        """Simulate fetching data from a competitor source."""
        # In production: Use Playwright for JS-heavy sites or APIs for social
        try:
            response = requests.get(source_url, timeout=10)
            soup = BeautifulSoup(response.text, 'html.parser')
            
            # Simple extraction logic (demo only)
            content = soup.get_text()[:500] 
            change_type = "pricing" if "price" in content.lower() else "feature"
            
            return CompetitorChange(
                source_type="website",
                url=source_url,
                content_summary=content,
                detected_at=datetime.now(),
                change_type=change_type
            )
        except Exception as e:
            raise ValueError(f"Failed to ingest data: {str(e)}")

    def run(self, state: IntelState) -> IntelState:
        """Execute monitoring."""
        # Demo URL
        url = f"https://{state['competitor_name'].lower()}.com/pricing"
        state['detected_change'] = self.monitor_competitor(state['competitor_name'], url)
        return state

Step 3: Context Retriever Agent (RAG)

This agent retrieves historical data to provide context. Is this price drop normal?

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document

class ContextRetrieverAgent:
    def __init__(self, vector_db_path: str = "./competitor_history_db"):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.vector_db = Chroma(
            persist_directory=vector_db_path,
            embedding_function=self.embeddings,
            collection_name="competitor_history"
        )
        self._seed_history()

    def _seed_history(self):
        """Seed with historical data."""
        if self.vector_db._collection.count() == 0:
            docs = [
                Document(
                    page_content="ShieldNet lowered prices by 10% in Q3 2025 during holiday season.",
                    metadata={"competitor": "ShieldNet", "type": "pricing", "date": "2025-12-01"}
                ),
                Document(
                    page_content="ShieldNet launched 'Enterprise Lite' tier in Jan 2026.",
                    metadata={"competitor": "ShieldNet", "type": "feature", "date": "2026-01-15"}
                )
            ]
            self.vector_db.add_documents(docs)

    def retrieve_context(self, competitor_name: str, change_type: str) -> HistoricalContext:
        """Retrieve relevant historical context."""
        query = f"{competitor_name} {change_type} history"
        docs = self.vector_db.similarity_search(query, k=2)
        
        prev_pricing = None
        last_change = None
        if docs:
            last_change = docs[0].page_content
            if change_type == "pricing":
                prev_pricing = "Previous drop was 10% seasonal."

        return HistoricalContext(
            previous_pricing=prev_pricing,
            last_similar_change=last_change
        )

    def run(self, state: IntelState) -> IntelState:
        """Execute retrieval."""
        if state['detected_change']:
            state['historical_context'] = self.retrieve_context(
                state['competitor_name'], 
                state['detected_change'].change_type
            )
        return state

Step 4: Impact Analyzer Agent (LLM Reasoning)

This agent analyzes the strategic intent behind the change.

from langchain_openai import ChatOpenAI
import json

class ImpactAnalyzerAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.2)

    def analyze_impact(self, state: IntelState) -> IntelState:
        """Analyze the strategic impact of the detected change."""
        change = state['detected_change']
        context = state['historical_context']
        
        prompt = f"""
        You are a Competitive Intelligence Analyst.
        
        Detected Change:
        - Type: {change.change_type}
        - Summary: {change.content_summary[:200]}
        
        Historical Context:
        - Last Similar Change: {context.last_similar_change}
        - Previous Pricing Info: {context.previous_pricing}
        
        Task:
        1. Determine Impact Level (High/Medium/Low).
        2. Infer Strategic Intent (e.g., "Market Share Grab", "Product Launch Support").
        3. Assess Threat to Our Company.
        
        Return ONLY valid JSON:
        {{
            "impact_level": "string",
            "strategic_intent": "string",
            "threat_to_us": "string"
        }}
        """
        
        try:
            response = self.llm.invoke(prompt)
            analysis_data = json.loads(response.content)
            state['analysis'] = StrategicAnalysis(**analysis_data)
        except Exception as e:
            state['error_message'] = f"Analysis failed: {str(e)}"
        
        return state

Step 5: Strategy Generator Agent (Counter-Campaign)

This agent suggests specific actions.

class StrategyGeneratorAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

    def generate_counter_strategy(self, state: IntelState) -> IntelState:
        """Generate a counter-campaign or positioning update."""
        analysis = state['analysis']
        change = state['detected_change']
        
        prompt = f"""
        You are a Chief Marketing Officer.
        
        Competitor Action: {change.change_type} change detected.
        Strategic Intent: {analysis.strategic_intent}
        Threat Level: {analysis.threat_to_us}
        
        Task: Suggest a immediate counter-strategy.
        1. Recommended Action (e.g., "Update Sales Deck", "Launch Email Blast").
        2. Campaign Theme.
        3. 3 Key Messaging Points.
        4. Target Channel.
        
        Return ONLY valid JSON:
        {{
            "recommended_action": "string",
            "campaign_theme": "string",
            "key_messaging_points": ["point1", "point2"],
            "target_channel": "string"
        }}
        """
        
        try:
            response = self.llm.invoke(prompt)
            strategy_data = json.loads(response.content)
            state['counter_strategy'] = CounterStrategy(**strategy_data)
        except Exception as e:
            state['error_message'] = f"Strategy generation failed: {str(e)}"
        
        return state

Step 6: Assemble the LangGraph Workflow

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# Initialize Agents
ingestion_agent = DataIngestionAgent()
retriever_agent = ContextRetrieverAgent()
analyzer_agent = ImpactAnalyzerAgent()
strategy_agent = StrategyGeneratorAgent()

# Define Nodes
def ingest_node(state: IntelState) -> IntelState:
    return ingestion_agent.run(state)

def retrieve_node(state: IntelState) -> IntelState:
    return retriever_agent.run(state)

def analyze_node(state: IntelState) -> IntelState:
    return analyzer_agent.analyze_impact(state)

def strategy_node(state: IntelState) -> IntelState:
    return strategy_agent.generate_counter_strategy(state)

# Build Graph
workflow = StateGraph(IntelState)

workflow.add_node("ingest", ingest_node)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("analyze", analyze_node)
workflow.add_node("strategy", strategy_node)

workflow.set_entry_point("ingest")
workflow.add_edge("ingest", "retrieve")
workflow.add_edge("retrieve", "analyze")
workflow.add_edge("analyze", "strategy")
workflow.add_edge("strategy", END)

# Compile with Memory
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

Step 7: Execute the Workflow

def run_competitor_intel(competitor_name: str) -> Dict:
    """Main entry point."""
    
    initial_state = IntelState(
        competitor_name=competitor_name,
        detected_change=None,
        historical_context=None,
        analysis=None,
        counter_strategy=None,
        conversation_history=[],
        error_message=None
    )
    
    thread_id = f"intel_{competitor_name}_{uuid.uuid4().hex[:8]}"
    config = {"configurable": {"thread_id": thread_id}}
    
    result = app.invoke(initial_state, config=config)
    
    if result['error_message']:
        raise Exception(result['error_message'])
        
    return {
        "change": result['detected_change'].dict() if result['detected_change'] else None,
        "analysis": result['analysis'].dict() if result['analysis'] else None,
        "strategy": result['counter_strategy'].dict() if result['counter_strategy'] else None
    }

# Example Usage
if __name__ == "__main__":
    try:
        output = run_competitor_intel("ShieldNet")
        print(json.dumps(output, indent=2, default=str))
    except Exception as e:
        print(f"Error: {e}")

Sample Output

{
  "change": {
    "source_type": "website",
    "url": "https://shieldnet.com/pricing",
    "content_summary": "New Enterprise Tier available at 20% discount...",
    "detected_at": "2026-08-19T10:00:00",
    "change_type": "pricing"
  },
  "analysis": {
    "impact_level": "high",
    "strategic_intent": "Market Share Grab before Q3 end",
    "threat_to_us": "High risk of churn among price-sensitive mid-market clients."
  },
  "strategy": {
    "recommended_action": "Launch 'Value-Over-Price' Email Campaign",
    "campaign_theme": "Security That Pays for Itself",
    "key_messaging_points": [
      "Our 99.99% SLA vs. their 99.9%",
      "Included 24/7 Support vs. their paid add-on",
      "Zero hidden implementation fees"
    ],
    "target_channel": "Email & LinkedIn Sales Navigator"
  }
}

Memory and State Management

Why Persistence Matters

  1. Longitudinal Tracking: By saving state to Redis/Postgres, we can track how often "ShieldNet" changes pricing. If they do it every quarter, it’s a pattern, not an anomaly.

  2. Avoiding Redundancy: The MemorySaver ensures we don’t re-analyze the same URL if the crawler hits it twice in an hour.

  3. Audit Trail: Every strategic suggestion is linked to the specific data that triggered it, ensuring accountability.

Conclusion

This Real-Time Competitor Intel Agent transforms competitive intelligence from a reactive, manual task into a proactive, automated strategic asset. By combining:

✅ Real-Time Monitoring (Data Ingestion)

✅ Contextual Awareness (RAG)

✅ Strategic Reasoning (LLM Analysis)

✅ Actionable Outputs (Counter-Strategies)

Enterprises can stay ahead of market shifts with speed and precision. This system doesn’t just tell you what happened; it tells you what to do about it.