Introduction: What Is a Competitor Monitoring Agent?
A Competitor Monitoring Agent is an autonomous AI-powered system that continuously tracks, analyzes, and reports on your competitors' digital footprint. Built on LangChain, it combines large language models (LLMs) with specialized tools ā web scrapers, search APIs, news aggregators, and social media monitors ā to create an intelligent watchtower that never sleeps.
Unlike traditional competitive intelligence dashboards that simply aggregate data, a LangChain-powered agent actively reasons about what it finds. It connects dots between disparate signals, interprets the strategic implications of a competitor's pricing change, identifies patterns in product launches, and delivers actionable insights in natural language ā exactly the way a skilled market analyst would, but at machine scale and speed.
In this comprehensive guide, you'll build a fully functional competitor monitoring agent from scratch. You'll learn how to set up the LangChain environment, create specialized monitoring tools, wire up an agent with reasoning capabilities, implement scheduled monitoring pipelines, and apply production-grade best practices. Every code example is complete and ready to run.
Why Competitor Monitoring Matters for Modern Businesses
Competitive intelligence has evolved from a periodic manual effort into a continuous strategic necessity. Here's why an automated agent approach is transformative:
- Speed of detection ā A competitor can launch a new feature, change pricing, or publish a damaging comparison page at any moment. Manual monitoring means you might discover this days or weeks late. An agent catches it within minutes.
- Signal vs. noise ā The internet produces overwhelming amounts of data. An LLM-powered agent filters out noise and surfaces only what's strategically relevant, with context and recommended actions.
- Connecting dots autonomously ā A competitor's job posting for a specific role + a trademark filing + a cryptic tweet might individually mean nothing. Together, they signal a new product line. Agents excel at this synthesis.
- Cost efficiency ā A single agent replaces multiple full-time analysts for routine monitoring, freeing human experts for high-level strategy.
- Customizable intelligence ā You define what matters: pricing thresholds, keyword triggers, sentiment shifts, specific page changes. The agent adapts to your competitive landscape.
For product managers, founders, marketers, and strategy teams, a competitor monitoring agent is no longer a luxury ā it's a competitive necessity in fast-moving markets.
Setting Up Your LangChain Environment
Before building the agent, let's set up a complete development environment. We'll need LangChain itself, along with supporting libraries for web scraping, search, scheduling, and embeddings.
Installing Required Dependencies
# Core LangChain packages
pip install langchain langchain-openai langchain-community
# Web scraping and HTML parsing
pip install httpx beautifulsoup4 lxml selenium-wire
# Search APIs
pip install google-search-results serpapi google-search-results
# Vector store and embeddings for long-term memory
pip install chromadb sentence-transformers
# Scheduling for production monitoring
pip install celery redis schedule
# Optional: browser automation for JavaScript-heavy sites
pip install playwright
playwright install chromiumAPI Keys and Environment Configuration
Create a .env file to store your API keys securely. At minimum, you'll need an OpenAI API key for the LLM and optionally a SerpAPI key for web search.
# .env file
OPENAI_API_KEY=sk-your-openai-api-key-here
SERPAPI_API_KEY=your-serpapi-key-here
# Optional: for direct Google Search access
GOOGLE_API_KEY=your-google-api-key
GOOGLE_CSE_ID=your-custom-search-engine-idLoad these in your Python application:
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
SERPAPI_API_KEY = os.getenv("SERPAPI_API_KEY")
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
GOOGLE_CSE_ID = os.getenv("GOOGLE_CSE_ID")
# Model configuration
LLM_MODEL = "gpt-4-turbo" # or "gpt-3.5-turbo" for cost savings
TEMPERATURE = 0.1 # low temperature for factual consistencyBuilding the Core Competitor Monitoring Agent
Now we'll build the agent step by step. The architecture follows LangChain's agent pattern: we define tools (what the agent can do), a system prompt (how it should think), and an agent executor that orchestrates reasoning and tool use.
Step 1: Define Custom Monitoring Tools
Tools are the agent's hands. Each tool wraps a specific capability ā web search, page scraping, content extraction ā into a function the LLM can invoke. Let's build a comprehensive toolset.
# tools/competitor_tools.py
import json
import httpx
from bs4 import BeautifulSoup
from typing import List, Optional, Dict
from langchain.tools import tool
from langchain_community.tools import GoogleSearchResults
from datetime import datetime
# --- Tool 1: Web Search for Competitor Mentions ---
@tool
def search_competitor_news(query: str) -> str:
"""
Search the web for recent news, announcements, or mentions about a competitor.
Input: A search query like "Acme Corp latest product launch 2024"
Returns: Structured results with titles, snippets, and URLs from the last 30 days.
"""
search_tool = GoogleSearchResults(
api_key=os.getenv("SERPAPI_API_KEY"),
params={
"q": query,
"tbs": "qdr:m", # last month
"num": 10,
"hl": "en",
}
)
results = search_tool.run(query)
# Parse and structure results
parsed = []
for item in results.get("organic_results", [])[:10]:
parsed.append({
"title": item.get("title"),
"snippet": item.get("snippet"),
"url": item.get("link"),
"date": item.get("date", "Unknown date")
})
return json.dumps(parsed, indent=2)
# --- Tool 2: Scrape a Specific Competitor Page ---
@tool
def scrape_competitor_page(url: str) -> str:
"""
Scrape and extract the full text content from a competitor's web page.
Input: A complete URL like 'https://competitor.com/pricing'
Returns: The cleaned text content of the page, stripped of HTML tags and scripts.
Use this to check pricing changes, new features, or messaging updates.
"""
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36"
}
try:
response = httpx.get(url, headers=headers, timeout=15, follow_redirects=True)
response.raise_for_status()
soup = BeautifulSoup(response.text, "lxml")
# Remove script, style, and nav elements
for tag in soup(["script", "style", "nav", "footer", "header"]):
tag.decompose()
# Extract text from main content areas
main_content = soup.find("main") or soup.find("article") or soup.body
if main_content:
text = main_content.get_text(separator="\n", strip=True)
else:
text = soup.get_text(separator="\n", strip=True)
# Clean up excessive whitespace
lines = [line.strip() for line in text.split("\n") if line.strip()]
cleaned_text = "\n".join(lines[:200]) # limit to 200 lines
return f"URL: {url}\nTimestamp: {datetime.now().isoformat()}\n\nContent:\n{cleaned_text}"
except Exception as e:
return f"Error scraping {url}: {str(e)}"
# --- Tool 3: Check Specific Page Elements ---
@tool
def check_page_element(url: str, css_selector: str) -> str:
"""
Extract specific elements from a competitor page using CSS selectors.
Input: URL and CSS selector like 'https://competitor.com/pricing' and '.price-tier'
Returns: The text content of matching elements.
Use this to monitor specific pricing tables, feature lists, or call-to-action buttons.
"""
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
}
try:
response = httpx.get(url, headers=headers, timeout=15, follow_redirects=True)
response.raise_for_status()
soup = BeautifulSoup(response.text, "lxml")
elements = soup.select(css_selector)
if not elements:
return f"No elements found matching selector: {css_selector}"
results = []
for i, el in enumerate(elements[:20], 1):
text = el.get_text(strip=True)
if text:
results.append(f"Element {i}: {text}")
return "\n".join(results) if results else "Elements found but contained no text"
except Exception as e:
return f"Error checking element at {url}: {str(e)}"
# --- Tool 4: Compare Two Pages for Changes ---
@tool
def diff_competitor_pages(url: str, previous_content_hash: str) -> str:
"""
Check if a competitor page has changed since last observation.
Input: URL and a SHA256 hash of previously observed content.
Returns: A detailed diff summary or confirmation that nothing changed.
"""
import hashlib
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
}
try:
response = httpx.get(url, headers=headers, timeout=15, follow_redirects=True)
response.raise_for_status()
# Clean and hash current content
soup = BeautifulSoup(response.text, "lxml")
for tag in soup(["script", "style"]):
tag.decompose()
current_text = soup.get_text(strip=True)
current_hash = hashlib.sha256(current_text.encode()).hexdigest()
if current_hash == previous_content_hash:
return json.dumps({
"changed": False,
"message": "Page content is identical to previous observation.",
"current_hash": current_hash,
"timestamp": datetime.now().isoformat()
})
else:
# Return first 3000 chars for LLM analysis
return json.dumps({
"changed": True,
"message": "Page content has changed since last observation.",
"current_hash": current_hash,
"previous_hash": previous_content_hash,
"current_content_preview": current_text[:3000],
"timestamp": datetime.now().isoformat()
})
except Exception as e:
return json.dumps({"error": str(e)})Step 2: Create the Competitor Database for Memory
For effective monitoring over time, the agent needs memory ā a way to remember what it observed yesterday so it can detect changes today. We'll use a lightweight ChromaDB vector store to persist competitor profiles and historical observations.
# memory/competitor_store.py
import chromadb
from chromadb.config import Settings
from langchain_openai import OpenAIEmbeddings
from datetime import datetime
import json
class CompetitorMemoryStore:
"""
Persistent memory for competitor profiles and historical observations.
Uses ChromaDB with OpenAI embeddings for semantic retrieval.
"""
def __init__(self, persist_directory: str = "./competitor_db"):
self.client = chromadb.Client(
Settings(
chroma_db_impl="duckdb+parquet",
persist_directory=persist_directory,
anonymized_telemetry=False
)
)
self.embeddings = OpenAIEmbeddings(
model="text-embedding-3-small",
openai_api_key=os.getenv("OPENAI_API_KEY")
)
# Create or get collections
self.profiles = self.client.get_or_create_collection(
name="competitor_profiles",
metadata={"description": "Core competitor information and profiles"}
)
self.observations = self.client.get_or_create_collection(
name="competitor_observations",
metadata={"description": "Historical monitoring observations"}
)
def store_profile(self, competitor_name: str, profile_data: Dict) -> str:
"""Store or update a competitor's profile."""
doc_id = f"profile_{competitor_name.lower().replace(' ', '_')}"
embedding = self.embeddings.embed_query(json.dumps(profile_data))
self.profiles.upsert(
ids=[doc_id],
embeddings=[embedding],
metadatas=[{
"competitor_name": competitor_name,
"updated_at": datetime.now().isoformat(),
"data": json.dumps(profile_data)
}],
documents=[json.dumps(profile_data)]
)
return doc_id
def store_observation(self, competitor_name: str, observation: str, metadata: Dict = None):
"""Record a new monitoring observation."""
doc_id = f"obs_{competitor_name.lower().replace(' ', '_')}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
embedding = self.embeddings.embed_query(observation)
meta = metadata or {}
meta.update({
"competitor_name": competitor_name,
"timestamp": datetime.now().isoformat(),
"observation_text": observation
})
self.observations.add(
ids=[doc_id],
embeddings=[embedding],
metadatas=[meta],
documents=[observation]
)
return doc_id
def get_recent_observations(self, competitor_name: str, days: int = 7) -> List[str]:
"""Retrieve observations from the last N days for a competitor."""
results = self.observations.get(
where={"competitor_name": competitor_name},
limit=20
)
observations = []
for doc, meta in zip(results.get("documents", []), results.get("metadatas", [])):
observations.append(f"[{meta.get('timestamp', 'Unknown')}] {doc}")
return observations
def get_profile(self, competitor_name: str) -> Optional[Dict]:
"""Retrieve a competitor's stored profile."""
doc_id = f"profile_{competitor_name.lower().replace(' ', '_')}"
results = self.profiles.get(ids=[doc_id])
if results and results["metadatas"]:
return json.loads(results["metadatas"][0].get("data", "{}"))
return NoneStep 3: Craft the System Prompt
The system prompt defines the agent's persona, reasoning framework, and operational boundaries. A well-crafted prompt transforms a general-purpose LLM into a specialized competitive intelligence analyst.
# prompts/system_prompt.py
COMPETITOR_ANALYST_PROMPT = """You are an elite Competitive Intelligence Analyst AI. Your mission is to monitor, analyze,
and report on competitor activity with precision, depth, and strategic insight.
## Your Capabilities
You have access to these tools:
- search_competitor_news: Find recent news about competitors
- scrape_competitor_page: Extract full content from any competitor web page
- check_page_element: Extract specific elements (pricing, features, etc.) from pages
- diff_competitor_pages: Detect changes in competitor pages since last check
## Your Monitoring Targets
The following competitors are being tracked:
{competitor_list}
## Your Analysis Framework
For every observation, apply this structured analysis:
1. **FACTUAL OBSERVATION** - What exactly was observed? Quote specifics.
2. **STRATEGIC SIGNIFICANCE** - Why does this matter? Rate impact: LOW / MEDIUM / HIGH / CRITICAL
3. **COMPETITIVE RESPONSE** - What should we do? Suggest concrete actions.
4. **TREND CONNECTION** - Does this connect to previous observations or broader market trends?
5. **CONFIDENCE LEVEL** - How certain are you? (High/Medium/Low) and why.
## Output Format
Structure every report as:
---
**Date:** [current date]
**Competitor:** [name]
**Signal Type:** [Product Launch / Pricing Change / Marketing Shift / Hiring / Partnership / Other]
**Observation:**
[Detailed factual observation]
**Strategic Assessment:**
[Impact rating and rationale]
**Recommended Action:**
[Specific, actionable recommendations]
**Connections:**
[Links to previous observations or market context]
---
## Critical Rules
- Never fabricate information. If you cannot verify something, state your uncertainty clearly.
- Prioritize HIGH and CRITICAL signals for immediate attention.
- When you detect a pricing change, ALWAYS flag it as HIGH or CRITICAL.
- If a competitor launches a feature similar to our roadmap, flag it as CRITICAL.
- Maintain a professional, analytical tone. Avoid speculation without evidence.
- When in doubt, use scrape_competitor_page to verify before reporting.
"""
# Shorter prompt variant for quick scans
QUICK_SCAN_PROMPT = """You are a Competitive Intelligence Analyst. Quickly scan the provided content
for any significant competitor signals. Focus on: pricing changes, new features, leadership changes,
partnerships, funding news, and marketing shifts. Report only what's noteworthy with a 1-line impact assessment.
Content to analyze:
{content}"""Step 4: Assemble the Agent Executor
Now we wire everything together. The agent executor manages the conversation loop: it receives a task, reasons about which tool to use, calls the tool, feeds the result back to the LLM, and repeats until it has a complete answer.
# agent/competitor_agent.py
import os
from typing import List, Dict
from datetime import datetime
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.schema import SystemMessage, HumanMessage
from tools.competitor_tools import (
search_competitor_news,
scrape_competitor_page,
check_page_element,
diff_competitor_pages,
)
from memory.competitor_store import CompetitorMemoryStore
from prompts.system_prompt import COMPETITOR_ANALYST_PROMPT
class CompetitorMonitoringAgent:
"""
Complete competitor monitoring agent with tools, memory, and reasoning.
"""
def __init__(self, competitors: List[Dict], model: str = "gpt-4-turbo"):
self.competitors = competitors
self.memory = CompetitorMemoryStore()
# Initialize LLM
self.llm = ChatOpenAI(
model=model,
temperature=0.1,
openai_api_key=os.getenv("OPENAI_API_KEY")
)
# Register tools
self.tools = [
search_competitor_news,
scrape_competitor_page,
check_page_element,
diff_competitor_pages,
]
# Build the agent
self.agent = create_openai_tools_agent(
llm=self.llm,
tools=self.tools,
prompt=self._build_prompt()
)
self.executor = AgentExecutor(
agent=self.agent,
tools=self.tools,
verbose=True,
max_iterations=8,
max_execution_time=120,
handle_parsing_errors=True,
return_intermediate_steps=True,
)
def _build_prompt(self) -> ChatPromptTemplate:
"""Build the agent's prompt with competitor context."""
competitor_str = "\n".join([
f"- **{c['name']}** ({c.get('website', 'N/A')}): {c.get('description', '')}"
for c in self.competitors
])
system_content = COMPETITOR_ANALYST_PROMPT.format(
competitor_list=competitor_str
)
return ChatPromptTemplate.from_messages([
("system", system_content),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
def run_monitoring_cycle(self, competitor_name: str, focus_areas: List[str] = None) -> Dict:
"""
Run a complete monitoring cycle for a single competitor.
Args:
competitor_name: Name of the competitor to monitor
focus_areas: Specific areas to focus on (pricing, products, hiring, etc.)
Returns:
Dict with full analysis report and metadata
"""
focus = focus_areas or ["pricing", "new products", "leadership changes",
"partnerships", "marketing campaigns", "website changes"]
# Build the monitoring task
task = f"""
Perform a comprehensive competitive intelligence scan on {competitor_name}.
Focus specifically on these areas:
{chr(10).join([f'- {area}' for area in focus])}
Steps:
1. First, use search_competitor_news to find recent mentions and news about {competitor_name}
2. For their main website, use scrape_competitor_page to check their homepage, pricing page,
and any product pages you discover
3. Use check_page_element to examine specific pricing tiers or feature lists
4. If you have previous observations from memory, use diff_competitor_pages to check for changes
Produce a structured analysis report following the framework in your system prompt.
Flag any CRITICAL or HIGH impact signals immediately.
"""
# Execute
result = self.executor.invoke({
"input": task
})
# Store the observation in memory
self.memory.store_observation(
competitor_name=competitor_name,
observation=result.get("output", ""),
metadata={
"focus_areas": focus,
"cycle_timestamp": datetime.now().isoformat()
}
)
return {
"competitor": competitor_name,
"timestamp": datetime.now().isoformat(),
"report": result.get("output"),
"intermediate_steps": result.get("intermediate_steps", []),
"tool_calls_count": len(result.get("intermediate_steps", []))
}
def run_full_monitoring(self) -> List[Dict]:
"""
Run monitoring for ALL tracked competitors.
Returns a list of reports.
"""
reports = []
for competitor in self.competitors:
print(f"\n{'='*60}")
print(f"Monitoring: {competitor['name']}")
print(f"{'='*60}")
report = self.run_monitoring_cycle(competitor['name'])
reports.append(report)
return reports
def generate_daily_briefing(self) -> str:
"""Generate an executive briefing from today's observations."""
all_obs = []
for competitor in self.competitors:
obs = self.memory.get_recent_observations(competitor['name'], days=1)
all_obs.extend(obs)
if not all_obs:
return "No observations recorded today."
briefing_prompt = f"""
You are an executive briefing specialist. Synthesize the following competitive intelligence
observations into a concise executive briefing. Group signals by competitor and by strategic
impact level. Highlight the top 3 most critical signals that require immediate attention.
Observations:
{chr(10).join(all_obs)}
Format:
# Executive Daily Briefing - {datetime.now().strftime('%B %d, %Y')}
## Critical Signals (Immediate Action Required)
[Top signals]
## Competitor-by-Competitor Summary
[Organized by competitor]
## Trend Watch
[Emerging patterns across competitors]
"""
briefing_llm = ChatOpenAI(
model="gpt-4-turbo",
temperature=0.2,
openai_api_key=os.getenv("OPENAI_API_KEY")
)
response = briefing_llm.invoke(briefing_prompt)
return response.contentStep 5: Initialize and Run the Agent
Here's how to put everything together and execute your first monitoring run:
# main.py - Complete working example
import os
from dotenv import load_dotenv
from agent.competitor_agent import CompetitorMonitoringAgent
load_dotenv()
# Define your competitor watchlist
COMPETITORS = [
{
"name": "Stripe",
"website": "https://stripe.com",
"description": "Payment processing and financial infrastructure platform",
"key_pages": ["/pricing", "/products", "/docs", "/blog"]
},
{
"name": "Adyen",
"website": "https://www.adyen.com",
"description": "Global payments platform for enterprises",
"key_pages": ["/pricing", "/products", "/resources"]
},
{
"name": "Square",
"website": "https://squareup.com",
"description": "Payment solutions for small to large businesses",
"key_pages": ["/pricing", "/products", "/news"]
}
]
def main():
"""Run a complete competitor monitoring cycle."""
print("š Initializing Competitor Monitoring Agent...")
print(f"Tracking {len(COMPETITORS)} competitors\n")
# Initialize the agent
agent = CompetitorMonitoringAgent(
competitors=COMPETITORS,
model="gpt-4-turbo" # Use gpt-3.5-turbo for lower cost
)
# Optional: Pre-populate competitor profiles
for comp in COMPETITORS:
agent.memory.store_profile(
comp['name'],
{
"name": comp['name'],
"website": comp['website'],
"description": comp['description'],
"key_pages": comp['key_pages'],
"first_tracked": "2024-01-01"
}
)
# Run a single competitor deep-dive
print("š Running deep-dive on Stripe...")
stripe_report = agent.run_monitoring_cycle(
competitor_name="Stripe",
focus_areas=["pricing changes", "new product features", "API updates",
"partnership announcements", "leadership moves"]
)
print("\n" + "="*60)
print("STRIPE MONITORING REPORT")
print("="*60)
print(stripe_report["report"])
print(f"\nTool calls made: {stripe_report['tool_calls_count']}")
# Run full monitoring on all competitors
print("\n\nš Running full monitoring cycle on all competitors...")
all_reports = agent.run_full_monitoring()
# Generate executive briefing
print("\n\nš Generating Executive Daily Briefing...")
briefing = agent.generate_daily_briefing()
print(briefing)
# Save reports to disk
import json
with open("monitoring_output.json", "w") as f:
json.dump(all_reports, f, indent=2, default=str)
print("\nā
Monitoring complete. Reports saved to monitoring_output.json")
if __name__ == "__main__":
main()Advanced Features and Customization
Once the core agent is running, you can extend it with advanced capabilities that transform it from a simple monitor into a strategic intelligence platform.
Adding Social Media Monitoring
Competitor activity on Twitter/X, LinkedIn, and other platforms often contains early signals. Here's a tool to track social presence:
# tools/social_monitor.py
import httpx
from langchain.tools import tool
from datetime import datetime, timedelta
@tool
def search_social_mentions(competitor_name: str, platform: str = "twitter") -> str:
"""
Search for recent social media mentions of a competitor on specified platforms.
Input: Competitor name and platform ('twitter', 'linkedin', 'reddit')
Returns: Recent posts and discussions mentioning the competitor.
"""
# Use a social search API (e.g., Social Searcher, or Twitter/X API via RapidAPI)
# This example uses a generic search approach
search_query = f"{competitor_name} site:{platform}.com (announcement OR launch OR new OR update)"
# For production, use dedicated social media APIs
# Example: Twitter API v2 via bearer token
headers = {
"Authorization": f"Bearer {os.getenv('TWITTER_BEARER_TOKEN')}",
"Content-Type": "application/json"
}
# Simplified implementation - in production, use proper API endpoints
# Return structured social data
return f"Social mentions found for {competitor_name} on {platform}. [Implement with real API]"
@tool
def monitor_linkedin_jobs(competitor_name: str) -> str:
"""
Monitor competitor's job postings on LinkedIn to detect hiring trends.
Input: Competitor company name
Returns: Recent job postings with titles, departments, and locations.
Hiring spikes in specific departments often signal product expansion.
"""
# In production, integrate with LinkedIn Jobs API or scraping
# For now, use web search as proxy
from tools.competitor_tools import search_competitor_news
query = f"{competitor_name} hiring jobs careers new roles 2024"
results = search_competitor_news(query)
analysis_prompt = f"""
Analyze these job-related search results for {competitor_name}.
Identify: departments hiring, seniority levels, approximate hiring volume,
and what product areas the hiring suggests they're investing in.
Results: {results}
"""
return analysis_promptAutomated Pricing Intelligence
Pricing changes are among the most critical competitive signals. This specialized module does deep pricing analysis:
# tools/pricing_intel.py
from langchain.tools import tool
from typing import Dict
import json
import re
@tool
def extract_pricing_data(page_url: str) -> str:
"""
Extract and structure pricing information from a competitor's pricing page.
Input: URL of the competitor's pricing page
Returns: Structured JSON with plan names, prices, features, and billing cycles.
"""
from tools.competitor_tools import scrape_competitor_page
page_content = scrape_competitor_page(page_url)
# Use LLM to extract structured pricing data
extraction_prompt = f"""
Extract ALL pricing information from this page content.
Return a structured JSON object with this exact schema:
{{
"currency": "USD",
"plans": [
{{
"name": "plan name",
"price": numeric_price,
"billing_cycle": "monthly/annual/one-time",
"features": ["feature1", "feature2"],
"limits": {{"users": null, "storage": "unlimited", ...}},
"target_segment": "startup/enterprise/smb"
}}
],
"free_tier": true/false,
"enterprise_custom_pricing": true/false,
"last_updated": "date if visible"
}}
Page content:
{page_content[:5000]}
"""
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0,
openai_api_key=os.getenv("OPENAI_API_KEY"))
response = llm.invoke(extraction_prompt)
# Try to parse and validate JSON
try:
structured = json.loads(response.content)
return json.dumps(structured, indent=2)
except json.JSONDecodeError:
return response.content
@tool
def compare_pricing_across_competitors(competitor_pages: Dict[str, str]) -> str:
"""
Compare pricing across multiple competitors.
Input: Dict mapping competitor names to their pricing page URLs
Returns: Comparative pricing analysis with positioning insights.
Example input: {{"Stripe": "https://stripe.com/pricing", "Adyen": "https://adyen.com/pricing"}}
"""
all_pricing = {}
for name, url in competitor_pages.items():
pricing_data = extract_pricing_data(url)
all_pricing[name] = pricing_data
comparison_prompt = f"""
Perform a detailed pricing comparison across these competitors:
{json.dumps(all_pricing, indent=2)}
Analyze:
1. Price positioning (who's cheapest, most expensive)
2. Feature differentiation at each tier
3. Target segment alignment
4. Pricing strategy signals (usage-based vs. tiered vs. flat-rate)
5. Our competitive pricing recommendations
Format as a structured comparison report.
"""
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.2,
openai_api_key=os.getenv("OPENAI_API_KEY"))
return llm.invoke(comparison_prompt).contentSetting Up Scheduled Production Monitoring
For production use, the agent needs to run on a schedule. Here's a robust scheduling setup using Celery with Redis, suitable for deployment:
# scheduler/monitoring_scheduler.py
from celery import Celery
from celery.schedules import crontab
from datetime import datetime
import json
import os
# Initialize Celery app
app = Celery(
'competitor_monitoring',
broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1'
)
app.conf.update(
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='UTC',
enable_utc=True,
)
# Configure periodic tasks
app.conf.beat_schedule = {
# Daily comprehensive monitoring at 6 AM
'daily-full-monitoring': {
'task': 'scheduler.monitoring_scheduler.run_full_monitoring',
'schedule': crontab(hour=6, minute=0),
'args':