REVAMP of AI Competitor Intelligence Agent Team
This commit is contained in:
Madhu
parent
5d4bd68156
commit
df0cb66b0f
3 changed files with 311 additions and 202 deletions
|
|
@ -1,6 +1,6 @@
|
|||
# 🧲 AI Competitor Intelligence Agent Team
|
||||
|
||||
The AI Competitor Intelligence Agent team is a powerful competitor analysis tool powered by Firecrawl and Phidata's AI Agent framework. This app helps businesses analyze their competitors by extracting structured data from competitor websites and generating actionable insights using AI.
|
||||
The AI Competitor Intelligence Agent Team is a powerful competitor analysis tool powered by Firecrawl and Agno's AI Agent framework. This app helps businesses analyze their competitors by extracting structured data from competitor websites and generating actionable insights using AI.
|
||||
|
||||
## Features
|
||||
|
||||
|
|
@ -29,7 +29,7 @@ The AI Competitor Intelligence Agent team is a powerful competitor analysis tool
|
|||
|
||||
The application requires the following Python libraries:
|
||||
|
||||
- `phidata`
|
||||
- `agno`
|
||||
- `exa-py`
|
||||
- `streamlit`
|
||||
- `pandas`
|
||||
|
|
|
|||
|
|
@ -1,10 +1,15 @@
|
|||
import streamlit as st
|
||||
from exa_py import Exa
|
||||
from phi.agent import Agent
|
||||
from phi.tools.firecrawl import FirecrawlTools
|
||||
from phi.model.openai import OpenAIChat
|
||||
from phi.tools.duckduckgo import DuckDuckGo
|
||||
from agno.agent import Agent
|
||||
from agno.tools.firecrawl import FirecrawlTools
|
||||
from agno.models.openai import OpenAIChat
|
||||
from agno.tools.duckduckgo import DuckDuckGoTools
|
||||
import pandas as pd
|
||||
import requests
|
||||
from firecrawl import FirecrawlApp
|
||||
from pydantic import BaseModel, Field
|
||||
from typing import List, Optional
|
||||
import json
|
||||
|
||||
# Streamlit UI
|
||||
st.set_page_config(page_title="AI Competitor Intelligence Agent Team", layout="wide")
|
||||
|
|
@ -13,15 +18,33 @@ st.set_page_config(page_title="AI Competitor Intelligence Agent Team", layout="w
|
|||
st.sidebar.title("API Keys")
|
||||
openai_api_key = st.sidebar.text_input("OpenAI API Key", type="password")
|
||||
firecrawl_api_key = st.sidebar.text_input("Firecrawl API Key", type="password")
|
||||
exa_api_key = st.sidebar.text_input("Exa API Key", type="password")
|
||||
|
||||
# Store API keys in session state
|
||||
if openai_api_key and firecrawl_api_key and exa_api_key:
|
||||
st.session_state.openai_api_key = openai_api_key
|
||||
st.session_state.firecrawl_api_key = firecrawl_api_key
|
||||
st.session_state.exa_api_key = exa_api_key
|
||||
else:
|
||||
st.sidebar.warning("Please enter all API keys to proceed.")
|
||||
# Add search engine selection before API keys
|
||||
search_engine = st.sidebar.selectbox(
|
||||
"Select Search Endpoint",
|
||||
options=["Perplexity AI - Sonar Pro", "Exa AI"],
|
||||
help="Choose which AI service to use for finding competitor URLs"
|
||||
)
|
||||
|
||||
# Show relevant API key input based on selection
|
||||
if search_engine == "Perplexity AI - Sonar Pro":
|
||||
perplexity_api_key = st.sidebar.text_input("Perplexity API Key", type="password")
|
||||
# Store API keys in session state
|
||||
if openai_api_key and firecrawl_api_key and perplexity_api_key:
|
||||
st.session_state.openai_api_key = openai_api_key
|
||||
st.session_state.firecrawl_api_key = firecrawl_api_key
|
||||
st.session_state.perplexity_api_key = perplexity_api_key
|
||||
else:
|
||||
st.sidebar.warning("Please enter all required API keys to proceed.")
|
||||
else: # Exa AI
|
||||
exa_api_key = st.sidebar.text_input("Exa API Key", type="password")
|
||||
# Store API keys in session state
|
||||
if openai_api_key and firecrawl_api_key and exa_api_key:
|
||||
st.session_state.openai_api_key = openai_api_key
|
||||
st.session_state.firecrawl_api_key = firecrawl_api_key
|
||||
st.session_state.exa_api_key = exa_api_key
|
||||
else:
|
||||
st.sidebar.warning("Please enter all required API keys to proceed.")
|
||||
|
||||
# Main UI
|
||||
st.title("🧲 AI Competitor Intelligence Agent")
|
||||
|
|
@ -32,206 +55,292 @@ st.info(
|
|||
- The app will fetch competitor URLs, extract relevant information, and generate a detailed analysis report.
|
||||
"""
|
||||
)
|
||||
st.success("For better results, provide both URL and a 5-6 word description of your company!")
|
||||
|
||||
# Input fields for URL and description
|
||||
url = st.text_input("Enter your company URL :")
|
||||
description = st.text_area("Enter a description of your company (if URL is not available):")
|
||||
|
||||
# Initialize API keys and tools
|
||||
if "openai_api_key" in st.session_state and "firecrawl_api_key" in st.session_state and "exa_api_key" in st.session_state:
|
||||
exa = Exa(api_key=st.session_state.exa_api_key)
|
||||
|
||||
firecrawl_tools = FirecrawlTools(
|
||||
api_key=st.session_state.firecrawl_api_key,
|
||||
scrape=False,
|
||||
crawl=True,
|
||||
limit=5
|
||||
)
|
||||
|
||||
firecrawl_agent = Agent(
|
||||
model=OpenAIChat(id="gpt-4o-mini", api_key=st.session_state.openai_api_key),
|
||||
tools=[firecrawl_tools, DuckDuckGo()],
|
||||
show_tool_calls=True,
|
||||
markdown=True
|
||||
)
|
||||
|
||||
analysis_agent = Agent(
|
||||
model=OpenAIChat(id="gpt-4o-mini", api_key=st.session_state.openai_api_key),
|
||||
show_tool_calls=True,
|
||||
markdown=True
|
||||
)
|
||||
|
||||
# New agent for comparing competitor data
|
||||
comparison_agent = Agent(
|
||||
model=OpenAIChat(id="gpt-4o-mini", api_key=st.session_state.openai_api_key),
|
||||
show_tool_calls=True,
|
||||
markdown=True
|
||||
)
|
||||
|
||||
def get_competitor_urls(url=None, description=None):
|
||||
if url:
|
||||
result = exa.find_similar(
|
||||
url=url,
|
||||
num_results=3,
|
||||
exclude_source_domain=True,
|
||||
category="company"
|
||||
)
|
||||
elif description:
|
||||
result = exa.search(
|
||||
description,
|
||||
type="neural",
|
||||
category="company",
|
||||
use_autoprompt=True,
|
||||
num_results=3
|
||||
)
|
||||
else:
|
||||
raise ValueError("Please provide either a URL or a description.")
|
||||
if "openai_api_key" in st.session_state and "firecrawl_api_key" in st.session_state:
|
||||
if (search_engine == "Perplexity AI - Sonar Pro" and "perplexity_api_key" in st.session_state) or \
|
||||
(search_engine == "Exa AI" and "exa_api_key" in st.session_state):
|
||||
|
||||
competitor_urls = [item.url for item in result.results]
|
||||
return competitor_urls
|
||||
# Initialize Exa only if selected
|
||||
if search_engine == "Exa AI":
|
||||
exa = Exa(api_key=st.session_state.exa_api_key)
|
||||
|
||||
def extract_competitor_info(competitor_url: str):
|
||||
try:
|
||||
crawl_response = firecrawl_agent.run(f"Crawl and summarize {competitor_url}")
|
||||
crawled_data = crawl_response.content
|
||||
|
||||
return {
|
||||
"competitor": competitor_url,
|
||||
"data": crawled_data
|
||||
}
|
||||
except Exception as e:
|
||||
st.error(f"Error extracting info for {competitor_url}: {e}")
|
||||
return {
|
||||
"competitor": competitor_url,
|
||||
"error": str(e)
|
||||
}
|
||||
|
||||
def generate_comparison_report(competitor_data: list) -> None:
|
||||
"""
|
||||
Generate and display a comparison report of competitor data.
|
||||
|
||||
Args:
|
||||
competitor_data: List of dictionaries containing competitor information
|
||||
"""
|
||||
# Combine all competitor data into a single string
|
||||
combined_data = "\n\n".join([str(data) for data in competitor_data])
|
||||
|
||||
# Updated system prompt for more structured output
|
||||
system_prompt = """
|
||||
As an expert business analyst, analyze the competitor data and create a structured comparison table.
|
||||
|
||||
Format the data in EXACTLY this markdown table structure:
|
||||
| Company | Pricing | Key Features | Tech Stack | Marketing Focus | Customer Feedback |
|
||||
|---------|---------|--------------|------------|-----------------|-------------------|
|
||||
| [Company Name 1] | ... | ... | ... | ... | ... |
|
||||
| [Company Name 2] | ... | ... | ... | ... | ... |
|
||||
| [Company Name 3] | ... | ... | ... | ... | ... |
|
||||
|
||||
Rules:
|
||||
1. Always include all columns
|
||||
2. Use the exact column names specified above
|
||||
3. Keep entries concise but informative
|
||||
4. Use pipe symbols (|) to separate columns
|
||||
5. Include the separator row (|---|) after headers
|
||||
|
||||
Competitor Data:
|
||||
{combined_data}
|
||||
"""
|
||||
|
||||
# Get comparison table from agent
|
||||
comparison_response = comparison_agent.run(
|
||||
system_prompt.format(combined_data=combined_data)
|
||||
firecrawl_tools = FirecrawlTools(
|
||||
api_key=st.session_state.firecrawl_api_key,
|
||||
scrape=False,
|
||||
crawl=True,
|
||||
limit=5
|
||||
)
|
||||
|
||||
# Display the raw markdown table first
|
||||
st.subheader("Competitor Comparison")
|
||||
st.markdown(comparison_response.content)
|
||||
|
||||
try:
|
||||
# Split the markdown table into lines and clean them
|
||||
table_lines = [
|
||||
line.strip()
|
||||
for line in comparison_response.content.split('\n')
|
||||
if line.strip() and '|' in line
|
||||
]
|
||||
|
||||
# Extract headers (first row)
|
||||
headers = [
|
||||
col.strip()
|
||||
for col in table_lines[0].split('|')
|
||||
if col.strip()
|
||||
]
|
||||
|
||||
# Extract data rows (skip header and separator rows)
|
||||
data_rows = []
|
||||
for line in table_lines[2:]: # Skip header and separator rows
|
||||
row_data = [
|
||||
cell.strip()
|
||||
for cell in line.split('|')
|
||||
if cell.strip()
|
||||
]
|
||||
if len(row_data) == len(headers):
|
||||
data_rows.append(row_data)
|
||||
|
||||
# Create DataFrame with explicit index
|
||||
df = pd.DataFrame(
|
||||
data_rows,
|
||||
columns=headers,
|
||||
index=range(len(data_rows))
|
||||
)
|
||||
|
||||
# # Display the DataFrame
|
||||
# st.subheader("Competitor Comparison Table")
|
||||
# st.table(df)
|
||||
|
||||
except Exception as e:
|
||||
st.error(f"Error converting table to DataFrame: {str(e)}")
|
||||
st.write("Raw table data for debugging:", table_lines)
|
||||
|
||||
def generate_analysis_report(competitor_data: list):
|
||||
combined_data = "\n\n".join([str(data) for data in competitor_data])
|
||||
|
||||
report = analysis_agent.run(
|
||||
f"""Analyze the following competitor data and identify market opportunities to improve my own company:
|
||||
{combined_data}
|
||||
firecrawl_agent = Agent(
|
||||
model=OpenAIChat(id="gpt-4o", api_key=st.session_state.openai_api_key),
|
||||
tools=[firecrawl_tools, DuckDuckGoTools()],
|
||||
show_tool_calls=True,
|
||||
markdown=True
|
||||
)
|
||||
|
||||
Tasks:
|
||||
1. Identify market gaps and opportunities based on competitor offerings
|
||||
2. Analyze competitor weaknesses that we can capitalize on
|
||||
3. Recommend unique features or capabilities we should develop
|
||||
4. Suggest pricing and positioning strategies to gain competitive advantage
|
||||
5. Outline specific growth opportunities in underserved market segments
|
||||
6. Provide actionable recommendations for product development and go-to-market strategy
|
||||
analysis_agent = Agent(
|
||||
model=OpenAIChat(id="gpt-4o", api_key=st.session_state.openai_api_key),
|
||||
show_tool_calls=True,
|
||||
markdown=True
|
||||
)
|
||||
|
||||
Focus on finding opportunities where we can differentiate and do better than competitors.
|
||||
Highlight any unmet customer needs or pain points we can address.
|
||||
# New agent for comparing competitor data
|
||||
comparison_agent = Agent(
|
||||
model=OpenAIChat(id="gpt-4o", api_key=st.session_state.openai_api_key),
|
||||
show_tool_calls=True,
|
||||
markdown=True
|
||||
)
|
||||
|
||||
def get_competitor_urls(url: str = None, description: str = None) -> list[str]:
|
||||
if not url and not description:
|
||||
raise ValueError("Please provide either a URL or a description.")
|
||||
|
||||
if search_engine == "Perplexity AI - Sonar Pro":
|
||||
perplexity_url = "https://api.perplexity.ai/chat/completions"
|
||||
|
||||
content = "Find me 3 competitor company URLs similar to the company with "
|
||||
if url and description:
|
||||
content += f"URL: {url} and description: {description}"
|
||||
elif url:
|
||||
content += f"URL: {url}"
|
||||
else:
|
||||
content += f"description: {description}"
|
||||
content += ". ONLY RESPOND WITH THE URLS, NO OTHER TEXT."
|
||||
|
||||
payload = {
|
||||
"model": "sonar-pro",
|
||||
"messages": [
|
||||
{
|
||||
"role": "system",
|
||||
"content": "Be precise and only return 3 company URLs ONLY."
|
||||
},
|
||||
{
|
||||
"role": "user",
|
||||
"content": content
|
||||
}
|
||||
],
|
||||
"max_tokens": 1000,
|
||||
"temperature": 0.2,
|
||||
}
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Bearer {st.session_state.perplexity_api_key}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.post(perplexity_url, json=payload, headers=headers)
|
||||
response.raise_for_status()
|
||||
urls = response.json()['choices'][0]['message']['content'].strip().split('\n')
|
||||
return [url.strip() for url in urls if url.strip()]
|
||||
except Exception as e:
|
||||
st.error(f"Error fetching competitor URLs from Perplexity: {str(e)}")
|
||||
return []
|
||||
|
||||
else: # Exa AI
|
||||
try:
|
||||
if url:
|
||||
result = exa.find_similar(
|
||||
url=url,
|
||||
num_results=3,
|
||||
exclude_source_domain=True,
|
||||
category="company"
|
||||
)
|
||||
else:
|
||||
result = exa.search(
|
||||
description,
|
||||
type="neural",
|
||||
category="company",
|
||||
use_autoprompt=True,
|
||||
num_results=3
|
||||
)
|
||||
return [item.url for item in result.results]
|
||||
except Exception as e:
|
||||
st.error(f"Error fetching competitor URLs from Exa: {str(e)}")
|
||||
return []
|
||||
|
||||
class CompetitorDataSchema(BaseModel):
|
||||
company_name: str = Field(description="Name of the company")
|
||||
pricing: str = Field(description="Pricing details, tiers, and plans")
|
||||
key_features: List[str] = Field(description="Main features and capabilities of the product/service")
|
||||
tech_stack: List[str] = Field(description="Technologies, frameworks, and tools used")
|
||||
marketing_focus: str = Field(description="Main marketing angles and target audience")
|
||||
customer_feedback: str = Field(description="Customer testimonials, reviews, and feedback")
|
||||
|
||||
def extract_competitor_info(competitor_url: str) -> Optional[dict]:
|
||||
try:
|
||||
# Initialize FirecrawlApp with API key
|
||||
app = FirecrawlApp(api_key=st.session_state.firecrawl_api_key)
|
||||
|
||||
# Add wildcard to crawl subpages
|
||||
url_pattern = f"{competitor_url}/*"
|
||||
|
||||
extraction_prompt = """
|
||||
Extract detailed information about the company's offerings, including:
|
||||
- Company name and basic information
|
||||
- Pricing details, plans, and tiers
|
||||
- Key features and main capabilities
|
||||
- Technology stack and technical details
|
||||
- Marketing focus and target audience
|
||||
- Customer feedback and testimonials
|
||||
|
||||
Analyze the entire website content to provide comprehensive information for each field.
|
||||
"""
|
||||
|
||||
response = app.extract(
|
||||
[url_pattern],
|
||||
{
|
||||
'prompt': extraction_prompt,
|
||||
'schema': CompetitorDataSchema.model_json_schema(),
|
||||
}
|
||||
)
|
||||
|
||||
if response.get('success') and response.get('data'):
|
||||
extracted_info = response['data']
|
||||
|
||||
# Create JSON structure
|
||||
competitor_json = {
|
||||
"competitor_url": competitor_url,
|
||||
"company_name": extracted_info.get('company_name', 'N/A'),
|
||||
"pricing": extracted_info.get('pricing', 'N/A'),
|
||||
"key_features": extracted_info.get('key_features', [])[:5], # Top 5 features
|
||||
"tech_stack": extracted_info.get('tech_stack', [])[:5], # Top 5 tech stack items
|
||||
"marketing_focus": extracted_info.get('marketing_focus', 'N/A'),
|
||||
"customer_feedback": extracted_info.get('customer_feedback', 'N/A')
|
||||
}
|
||||
|
||||
return competitor_json
|
||||
|
||||
else:
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
return None
|
||||
|
||||
def generate_comparison_report(competitor_data: list) -> None:
|
||||
# Format the competitor data for the prompt
|
||||
formatted_data = json.dumps(competitor_data, indent=2)
|
||||
print(formatted_data)
|
||||
|
||||
# Updated system prompt for more structured output
|
||||
system_prompt = f"""
|
||||
As an expert business analyst, analyze the following competitor data in JSON format and create a structured comparison.
|
||||
Extract and summarize the key information into concise points.
|
||||
|
||||
{formatted_data}
|
||||
|
||||
Return the data in a structured format with EXACTLY these columns:
|
||||
Company, Pricing, Key Features, Tech Stack, Marketing Focus, Customer Feedback
|
||||
|
||||
Rules:
|
||||
1. For Company: Include company name and URL
|
||||
2. For Key Features: List top 3 most important features only
|
||||
3. For Tech Stack: List top 3 most relevant technologies only
|
||||
4. Keep all entries clear and concise
|
||||
5. Format feedback as brief quotes
|
||||
6. Return ONLY the structured data, no additional text
|
||||
"""
|
||||
)
|
||||
return report.content
|
||||
|
||||
# Run analysis when the user clicks the button
|
||||
if st.button("Analyze Competitors"):
|
||||
if url or description:
|
||||
with st.spinner("Fetching competitor URLs..."):
|
||||
competitor_urls = get_competitor_urls(url=url, description=description)
|
||||
st.write(f"Competitor URLs: {competitor_urls}")
|
||||
# Get comparison data from agent
|
||||
comparison_response = comparison_agent.run(system_prompt)
|
||||
|
||||
competitor_data = []
|
||||
for url in competitor_urls:
|
||||
with st.spinner(f"Analyzing Competitor: {url}..."):
|
||||
competitor_info = extract_competitor_info(url)
|
||||
competitor_data.append(competitor_info)
|
||||
try:
|
||||
# Split the response into lines and clean them
|
||||
table_lines = [
|
||||
line.strip()
|
||||
for line in comparison_response.content.split('\n')
|
||||
if line.strip() and '|' in line
|
||||
]
|
||||
|
||||
# Extract headers (first row)
|
||||
headers = [
|
||||
col.strip()
|
||||
for col in table_lines[0].split('|')
|
||||
if col.strip()
|
||||
]
|
||||
|
||||
# Extract data rows (skip header and separator rows)
|
||||
data_rows = []
|
||||
for line in table_lines[2:]: # Skip header and separator rows
|
||||
row_data = [
|
||||
cell.strip()
|
||||
for cell in line.split('|')
|
||||
if cell.strip()
|
||||
]
|
||||
if len(row_data) == len(headers):
|
||||
data_rows.append(row_data)
|
||||
|
||||
# Create DataFrame
|
||||
df = pd.DataFrame(
|
||||
data_rows,
|
||||
columns=headers
|
||||
)
|
||||
|
||||
# Display the table
|
||||
st.subheader("Competitor Comparison")
|
||||
st.table(df)
|
||||
|
||||
except Exception as e:
|
||||
st.error(f"Error creating comparison table: {str(e)}")
|
||||
st.write("Raw comparison data for debugging:", comparison_response.content)
|
||||
|
||||
def generate_analysis_report(competitor_data: list):
|
||||
# Format the competitor data for the prompt
|
||||
formatted_data = json.dumps(competitor_data, indent=2)
|
||||
print("Analysis Data:", formatted_data) # For debugging
|
||||
|
||||
# Generate and display comparison report
|
||||
with st.spinner("Generating comparison table..."):
|
||||
generate_comparison_report(competitor_data)
|
||||
|
||||
# Generate and display final analysis report
|
||||
with st.spinner("Generating analysis report..."):
|
||||
analysis_report = generate_analysis_report(competitor_data)
|
||||
st.subheader("Competitor Analysis Report")
|
||||
st.markdown(analysis_report)
|
||||
|
||||
st.success("Analysis complete!")
|
||||
else:
|
||||
st.error("Please provide either a URL or a description.")
|
||||
report = analysis_agent.run(
|
||||
f"""Analyze the following competitor data in JSON format and identify market opportunities to improve my own company:
|
||||
|
||||
{formatted_data}
|
||||
|
||||
Tasks:
|
||||
1. Identify market gaps and opportunities based on competitor offerings
|
||||
2. Analyze competitor weaknesses that we can capitalize on
|
||||
3. Recommend unique features or capabilities we should develop
|
||||
4. Suggest pricing and positioning strategies to gain competitive advantage
|
||||
5. Outline specific growth opportunities in underserved market segments
|
||||
6. Provide actionable recommendations for product development and go-to-market strategy
|
||||
|
||||
Focus on finding opportunities where we can differentiate and do better than competitors.
|
||||
Highlight any unmet customer needs or pain points we can address.
|
||||
"""
|
||||
)
|
||||
return report.content
|
||||
|
||||
# Run analysis when the user clicks the button
|
||||
if st.button("Analyze Competitors"):
|
||||
if url or description:
|
||||
with st.spinner("Fetching competitor URLs..."):
|
||||
competitor_urls = get_competitor_urls(url=url, description=description)
|
||||
st.write(f"Competitor URLs: {competitor_urls}")
|
||||
|
||||
competitor_data = []
|
||||
for comp_url in competitor_urls:
|
||||
with st.spinner(f"Analyzing Competitor: {comp_url}..."):
|
||||
competitor_info = extract_competitor_info(comp_url)
|
||||
if competitor_info is not None:
|
||||
competitor_data.append(competitor_info)
|
||||
|
||||
if competitor_data:
|
||||
# Generate and display comparison report
|
||||
with st.spinner("Generating comparison table..."):
|
||||
generate_comparison_report(competitor_data)
|
||||
|
||||
# Generate and display final analysis report
|
||||
with st.spinner("Generating analysis report..."):
|
||||
analysis_report = generate_analysis_report(competitor_data)
|
||||
st.subheader("Competitor Analysis Report")
|
||||
st.markdown(analysis_report)
|
||||
|
||||
st.success("Analysis complete!")
|
||||
else:
|
||||
st.error("Could not extract data from any competitor URLs")
|
||||
else:
|
||||
st.error("Please provide either a URL or a description.")
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
exa-py==1.7.1
|
||||
firecrawl-py==1.9.0
|
||||
duckduckgo-search==7.2.1
|
||||
phidata==2.7.3
|
||||
agno
|
||||
streamlit==1.41.1
|
||||
Loading…
Reference in a new issue