admin-valentin/awesome-llm-apps
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: improve competitor agent

Browse source 
- Comment out unused exa_py import
- Improve error handling for Firecrawl API responses
- Enhance competitor analysis UI with better feedback
- Replace agent-based comparison with DataFrame approach
- Add extraction success/failure tracking
This commit is contained in:
Madhu 2025-07-26 20:19:40 +05:30
parent b7a21406b1
commit 55b680e0ee
1 changed files with 75 additions and 88 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

153
advanced_ai_agents/multi_agent_apps/agent_teams/ai_competitor_intelligence_agent_team/competitor_agent_team.py
View file

@ -1,5 +1,5 @@
import streamlit as st import streamlit as st
from exa_py import Exa # from exa_py import Exa
from agno.agent import Agent from agno.agent import Agent
from agno.tools.firecrawl import FirecrawlTools from agno.tools.firecrawl import FirecrawlTools
from agno.models.openai import OpenAIChat from agno.models.openai import OpenAIChat
@ -195,100 +195,68 @@ if "openai_api_key" in st.session_state and "firecrawl_api_key" in st.session_st
response = app.extract( response = app.extract(
[url_pattern], [url_pattern],
{ prompt=extraction_prompt,
'prompt': extraction_prompt, schema=CompetitorDataSchema.model_json_schema()
'schema': CompetitorDataSchema.model_json_schema(),
}
) )
if response.get('success') and response.get('data'): # Handle ExtractResponse object
extracted_info = response['data'] try:
if hasattr(response, 'success') and response.success:
if hasattr(response, 'data') and response.data:
extracted_info = response.data
# Create JSON structure # Create JSON structure
competitor_json = { competitor_json = {
"competitor_url": competitor_url, "competitor_url": competitor_url,
"company_name": extracted_info.get('company_name', 'N/A'), "company_name": extracted_info.get('company_name', 'N/A') if isinstance(extracted_info, dict) else getattr(extracted_info, 'company_name', 'N/A'),
"pricing": extracted_info.get('pricing', 'N/A'), "pricing": extracted_info.get('pricing', 'N/A') if isinstance(extracted_info, dict) else getattr(extracted_info, 'pricing', 'N/A'),
"key_features": extracted_info.get('key_features', [])[:5], # Top 5 features "key_features": extracted_info.get('key_features', [])[:5] if isinstance(extracted_info, dict) and extracted_info.get('key_features') else getattr(extracted_info, 'key_features', [])[:5] if hasattr(extracted_info, 'key_features') else ['N/A'],
"tech_stack": extracted_info.get('tech_stack', [])[:5], # Top 5 tech stack items "tech_stack": extracted_info.get('tech_stack', [])[:5] if isinstance(extracted_info, dict) and extracted_info.get('tech_stack') else getattr(extracted_info, 'tech_stack', [])[:5] if hasattr(extracted_info, 'tech_stack') else ['N/A'],
"marketing_focus": extracted_info.get('marketing_focus', 'N/A'), "marketing_focus": extracted_info.get('marketing_focus', 'N/A') if isinstance(extracted_info, dict) else getattr(extracted_info, 'marketing_focus', 'N/A'),
"customer_feedback": extracted_info.get('customer_feedback', 'N/A') "customer_feedback": extracted_info.get('customer_feedback', 'N/A') if isinstance(extracted_info, dict) else getattr(extracted_info, 'customer_feedback', 'N/A')
} }
return competitor_json return competitor_json
else:
return None
else:
return None
else: except Exception as response_error:
return None return None
except Exception as e: except Exception as e:
return None return None
def generate_comparison_report(competitor_data: list) -> None: def generate_comparison_report(competitor_data: list) -> None:
# Format the competitor data for the prompt # Create DataFrame directly from competitor data
formatted_data = json.dumps(competitor_data, indent=2) if not competitor_data:
print(formatted_data) st.error("No competitor data available for comparison")
return
# Updated system prompt for more structured output # Prepare data for DataFrame
system_prompt = f""" table_data = []
As an expert business analyst, analyze the following competitor data in JSON format and create a structured comparison. for competitor in competitor_data:
Extract and summarize the key information into concise points. row = {
'Company': f"{competitor.get('company_name', 'N/A')} ({competitor.get('competitor_url', 'N/A')})",
'Pricing': competitor.get('pricing', 'N/A')[:100] + '...' if len(competitor.get('pricing', '')) > 100 else competitor.get('pricing', 'N/A'),
'Key Features': ', '.join(competitor.get('key_features', [])[:3]) if competitor.get('key_features') else 'N/A',
'Tech Stack': ', '.join(competitor.get('tech_stack', [])[:3]) if competitor.get('tech_stack') else 'N/A',
'Marketing Focus': competitor.get('marketing_focus', 'N/A')[:100] + '...' if len(competitor.get('marketing_focus', '')) > 100 else competitor.get('marketing_focus', 'N/A'),
'Customer Feedback': competitor.get('customer_feedback', 'N/A')[:100] + '...' if len(competitor.get('customer_feedback', '')) > 100 else competitor.get('customer_feedback', 'N/A')
}
table_data.append(row)
{formatted_data} # Create DataFrame
df = pd.DataFrame(table_data)
Return the data in a structured format with EXACTLY these columns: # Display the table
Company, Pricing, Key Features, Tech Stack, Marketing Focus, Customer Feedback st.subheader("Competitor Comparison")
st.dataframe(df, use_container_width=True)
Rules: # Also show raw data for debugging
1. For Company: Include company name and URL with st.expander("View Raw Competitor Data"):
2. For Key Features: List top 3 most important features only st.json(competitor_data)
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
"""
# Get comparison data from agent
comparison_response = comparison_agent.run(system_prompt)
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): def generate_analysis_report(competitor_data: list):
# Format the competitor data for the prompt # Format the competitor data for the prompt
@ -319,16 +287,31 @@ if "openai_api_key" in st.session_state and "firecrawl_api_key" in st.session_st
if url or description: if url or description:
with st.spinner("Fetching competitor URLs..."): with st.spinner("Fetching competitor URLs..."):
competitor_urls = get_competitor_urls(url=url, description=description) competitor_urls = get_competitor_urls(url=url, description=description)
st.write(f"Competitor URLs: {competitor_urls}") st.write(f"Found {len(competitor_urls)} competitor URLs")
if not competitor_urls:
st.error("No competitor URLs found!")
st.stop()
competitor_data = [] competitor_data = []
for comp_url in competitor_urls: successful_extractions = 0
with st.spinner(f"Analyzing Competitor: {comp_url}..."): failed_extractions = 0
for i, comp_url in enumerate(competitor_urls):
with st.spinner(f"Analyzing Competitor {i+1}/{len(competitor_urls)}: {comp_url}"):
competitor_info = extract_competitor_info(comp_url) competitor_info = extract_competitor_info(comp_url)
if competitor_info is not None: if competitor_info is not None:
competitor_data.append(competitor_info) competitor_data.append(competitor_info)
successful_extractions += 1
st.success(f"✓ Successfully analyzed {comp_url}")
else:
failed_extractions += 1
st.error(f"✗ Failed to analyze {comp_url}")
if competitor_data: if competitor_data:
st.success(f"Successfully analyzed {successful_extractions}/{len(competitor_urls)} competitors!")
# Generate and display comparison report # Generate and display comparison report
with st.spinner("Generating comparison table..."): with st.spinner("Generating comparison table..."):
generate_comparison_report(competitor_data) generate_comparison_report(competitor_data)
@ -342,5 +325,9 @@ if "openai_api_key" in st.session_state and "firecrawl_api_key" in st.session_st
st.success("Analysis complete!") st.success("Analysis complete!")
else: else:
st.error("Could not extract data from any competitor URLs") st.error("Could not extract data from any competitor URLs")
st.write("This might be due to:")
st.write("- API rate limits (try again in a few minutes)")
st.write("- Website access issues (some sites block automated access)")
st.write("- Invalid URLs (try with a different company description)")
else: else:
st.error("Please provide either a URL or a description.") st.error("Please provide either a URL or a description.")