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

Merge pull request #98 from Madhuvod/leadgen-fix-extract

Browse source 
Updated AI Lead generation Agent
This commit is contained in:
Shubham Saboo 2025-01-23 11:30:57 -06:00 committed by GitHub
commit 05b458dc8c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 86 additions and 83 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
ai_agent_tutorials/ai_lead_generation_agent/README.md
View file

@ -1,10 +1,10 @@
## 🎯 AI Lead Generation Agent ## 🎯 AI Lead Generation Agent - Powered by Firecrawl's Extract Endpoint
The AI Lead Generation Agent automates the process of finding and qualifying potential leads from Quora. It uses Firecrawl's search and LLM extraction endpoints to identify relevant user profiles, extract valuable information, and organize it into a structured format in Google Sheets. This agent helps sales and marketing teams efficiently build targeted lead lists while saving hours of manual research. The AI Lead Generation Agent automates the process of finding and qualifying potential leads from Quora. It uses Firecrawl's search and the new Extract endpoint to identify relevant user profiles, extract valuable information, and organize it into a structured format in Google Sheets. This agent helps sales and marketing teams efficiently build targeted lead lists while saving hours of manual research.
### Features ### Features
- **Targeted Search**: Uses Firecrawl's search endpoint to find relevant Quora URLs based on your search criteria - **Targeted Search**: Uses Firecrawl's search endpoint to find relevant Quora URLs based on your search criteria
- **Intelligent Extraction**: Leverages Firecrawl's LLM extract functionality to pull user information from Quora profiles - **Intelligent Extraction**: Leverages Firecrawl's new Extract endpoint to pull user information from Quora profiles
- **Automated Processing**: Formats extracted user information into a clean, structured format - **Automated Processing**: Formats extracted user information into a clean, structured format
- **Google Sheets Integration**: Automatically creates and populates Google Sheets with lead information - **Google Sheets Integration**: Automatically creates and populates Google Sheets with lead information
- **Customizable Criteria**: Allows you to define specific search parameters to find your ideal leads for your niche - **Customizable Criteria**: Allows you to define specific search parameters to find your ideal leads for your niche

163
ai_agent_tutorials/ai_lead_generation_agent/ai_lead_generation_agent.py
View file

@ -7,8 +7,8 @@ from firecrawl import FirecrawlApp
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from typing import List from typing import List
from composio_phidata import Action, ComposioToolSet from composio_phidata import Action, ComposioToolSet
import json
# Define a schema for a single user interaction (question or answer)
class QuoraUserInteractionSchema(BaseModel): class QuoraUserInteractionSchema(BaseModel):
username: str = Field(description="The username of the user who posted the question or answer") username: str = Field(description="The username of the user who posted the question or answer")
bio: str = Field(description="The bio or description of the user") bio: str = Field(description="The bio or description of the user")
@ -17,13 +17,10 @@ class QuoraUserInteractionSchema(BaseModel):
upvotes: int = Field(default=0, description="Number of upvotes received") upvotes: int = Field(default=0, description="Number of upvotes received")
links: List[str] = Field(default_factory=list, description="Any links included in the post") links: List[str] = Field(default_factory=list, description="Any links included in the post")
# Define a schema for the entire page, containing multiple interactions
class QuoraPageSchema(BaseModel): class QuoraPageSchema(BaseModel):
interactions: List[QuoraUserInteractionSchema] = Field(description="List of all user interactions (questions and answers) on the page") interactions: List[QuoraUserInteractionSchema] = Field(description="List of all user interactions (questions and answers) on the page")
# Step 1: Search for relevant URLs using Firecrawl def search_for_urls(company_description: str, firecrawl_api_key: str, num_links: int) -> List[str]:
def search_for_urls(company_description, firecrawl_api_key, num_links):
print("Step 1: Searching for relevant URLs using Firecrawl...")
url = "https://api.firecrawl.dev/v1/search" url = "https://api.firecrawl.dev/v1/search"
headers = { headers = {
"Authorization": f"Bearer {firecrawl_api_key}", "Authorization": f"Bearer {firecrawl_api_key}",
@ -32,7 +29,7 @@ def search_for_urls(company_description, firecrawl_api_key, num_links):
query1 = f"quora websites where people are looking for {company_description} services" query1 = f"quora websites where people are looking for {company_description} services"
payload = { payload = {
"query": query1, "query": query1,
"limit": num_links, # Use the num_links parameter here "limit": num_links,
"lang": "en", "lang": "en",
"location": "United States", "location": "United States",
"timeout": 60000, "timeout": 60000,
@ -42,46 +39,36 @@ def search_for_urls(company_description, firecrawl_api_key, num_links):
data = response.json() data = response.json()
if data.get("success"): if data.get("success"):
results = data.get("data", []) results = data.get("data", [])
print(f"Found {len(results)} relevant URLs.")
return [result["url"] for result in results] return [result["url"] for result in results]
else:
print("Search request was not successful.")
print(data.get("warning", "No warning provided."))
else:
print(f"Failed to retrieve data. Status code: {response.status_code}")
return [] return []
# Step 2: Extract user info from URLs using Firecrawl's LLM extract def extract_user_info_from_urls(urls: List[str], firecrawl_api_key: str) -> List[dict]:
def extract_user_info_from_urls(urls, firecrawl_api_key):
print("\nStep 2: Extracting user info from URLs using Firecrawl's LLM extract...")
user_info_list = [] user_info_list = []
firecrawl_app = FirecrawlApp(api_key=firecrawl_api_key) firecrawl_app = FirecrawlApp(api_key=firecrawl_api_key)
for website_url in urls:
print(f"Extracting user info from: {website_url}") try:
for url in urls:
# Use Firecrawl's LLM extract to get structured data response = firecrawl_app.extract(
data = firecrawl_app.scrape_url(website_url, { [url],
'formats': ['extract'], {
'extract': { 'prompt': 'Extract all user information including username, bio, post type (question/answer), timestamp, upvotes, and any links from Quora posts. Focus on identifying potential leads who are asking questions or providing answers related to the topic.',
'schema': QuoraPageSchema.model_json_schema(), 'schema': QuoraPageSchema.model_json_schema(),
} }
}) )
# Extract the interactions from the response if response.get('success') and response.get('status') == 'completed':
extracted_data = data.get("extract", {}) interactions = response.get('data', {}).get('interactions', [])
interactions = extracted_data.get("interactions", []) if interactions:
user_info_list.append({
# Store the results "website_url": url,
user_info_list.append({ "user_info": interactions
"website_url": website_url, })
"user_info": interactions except Exception:
}) pass
print(f"Extracted {len(interactions)} interactions from {website_url}.")
return user_info_list return user_info_list
# Step 3: Format the extracted user info into a flattened JSON structure def format_user_info_to_flattened_json(user_info_list: List[dict]) -> List[dict]:
def format_user_info_to_flattened_json(user_info_list):
print("\nStep 3: Formatting the extracted user info into a flattened JSON structure...")
flattened_data = [] flattened_data = []
for info in user_info_list: for info in user_info_list:
@ -96,88 +83,104 @@ def format_user_info_to_flattened_json(user_info_list):
"Post Type": interaction.get("post_type", ""), "Post Type": interaction.get("post_type", ""),
"Timestamp": interaction.get("timestamp", ""), "Timestamp": interaction.get("timestamp", ""),
"Upvotes": interaction.get("upvotes", 0), "Upvotes": interaction.get("upvotes", 0),
"Links": ", ".join(interaction.get("links", [])), # Convert list of links to a single string "Links": ", ".join(interaction.get("links", [])),
} }
flattened_data.append(flattened_interaction) flattened_data.append(flattened_interaction)
print(f"Formatted {len(flattened_data)} interactions into flattened JSON.")
return flattened_data return flattened_data
# Step 4: Create a new Phidata agent to interact with Google Sheets def create_google_sheets_agent(composio_api_key: str, openai_api_key: str) -> Agent:
def create_google_sheets_agent(composio_api_key, openai_api_key):
print("\nStep 4: Creating a new Phidata agent to interact with Google Sheets...")
# Initialize Composio Toolset
composio_toolset = ComposioToolSet(api_key=composio_api_key) composio_toolset = ComposioToolSet(api_key=composio_api_key)
# Get the Google Sheets tool
google_sheets_tool = composio_toolset.get_tools(actions=[Action.GOOGLESHEETS_SHEET_FROM_JSON])[0] google_sheets_tool = composio_toolset.get_tools(actions=[Action.GOOGLESHEETS_SHEET_FROM_JSON])[0]
# Create the agent
google_sheets_agent = Agent( google_sheets_agent = Agent(
model=OpenAIChat(id="gpt-4o-mini", api_key=openai_api_key), model=OpenAIChat(id="gpt-4o-mini", api_key=openai_api_key),
tools=[google_sheets_tool], tools=[google_sheets_tool],
show_tool_calls=True, # Enable verbose tool call output for debugging show_tool_calls=True,
system_prompt="You are an expert at creating and updating Google Sheets. You will be given user information in JSON format, and you need to write it into a new Google Sheet.", system_prompt="You are an expert at creating and updating Google Sheets. You will be given user information in JSON format, and you need to write it into a new Google Sheet.",
markdown=True markdown=True
) )
print("Google Sheets agent created successfully.")
return google_sheets_agent return google_sheets_agent
# Step 5: Write formatted user info to Google Sheets def write_to_google_sheets(flattened_data: List[dict], composio_api_key: str, openai_api_key: str) -> str:
def write_to_google_sheets(flattened_data, composio_api_key, openai_api_key):
print("\nStep 5: Writing formatted user info to Google Sheets...")
# Create the Google Sheets agent
google_sheets_agent = create_google_sheets_agent(composio_api_key, openai_api_key) google_sheets_agent = create_google_sheets_agent(composio_api_key, openai_api_key)
# Create a new Google Sheet from the flattened JSON data try:
print("Creating a new Google Sheet with the flattened JSON data...") message = (
create_sheet_response = google_sheets_agent.run( "Create a new Google Sheet with this data. "
f"Create a new Google Sheet with the following data:\n" "The sheet should have these columns: Website URL, Username, Bio, Post Type, Timestamp, Upvotes, and Links in the same order as mentioned. "
f"Title: Quora User Info\n" "Here's the data in JSON format:\n\n"
f"Sheet Name: Sheet1\n" f"{json.dumps(flattened_data, indent=2)}"
f"Sheet JSON: {flattened_data}" )
)
print("Create Sheet Response:", create_sheet_response.content) create_sheet_response = google_sheets_agent.run(message)
# Extract the Google Sheets link from the response if "https://docs.google.com/spreadsheets/d/" in create_sheet_response.content:
if "https://docs.google.com/spreadsheets/d/" in create_sheet_response.content: google_sheets_link = create_sheet_response.content.split("https://docs.google.com/spreadsheets/d/")[1].split(" ")[0]
google_sheets_link = create_sheet_response.content.split("https://docs.google.com/spreadsheets/d/")[1].split(" ")[0] return f"https://docs.google.com/spreadsheets/d/{google_sheets_link}"
google_sheets_link = f"https://docs.google.com/spreadsheets/d/{google_sheets_link}" except Exception:
return google_sheets_link pass
return None return None
# Streamlit UI def create_prompt_transformation_agent(openai_api_key: str) -> Agent:
return Agent(
model=OpenAIChat(id="gpt-4o-mini", api_key=openai_api_key),
system_prompt="""You are an expert at transforming detailed user queries into concise company descriptions.
Your task is to extract the core business/product focus in 3-4 words.
Examples:
Input: "Generate leads looking for AI-powered customer support chatbots for e-commerce stores."
Output: "AI customer support chatbots for e commerce"
Input: "Find people interested in voice cloning technology for creating audiobooks and podcasts"
Output: "voice cloning technology"
Input: "Looking for users who need automated video editing software with AI capabilities"
Output: "AI video editing software"
Input: "Need to find businesses interested in implementing machine learning solutions for fraud detection"
Output: "ML fraud detection"
Always focus on the core product/service and keep it concise but clear.""",
markdown=True
)
def main(): def main():
st.title("🎯 AI Lead Generation Agent") st.title("🎯 AI Lead Generation Agent")
st.info("This firecrawl powered agent helps you generate leads from Quora by searching for relevant posts and extracting user information.") st.info("This firecrawl powered agent helps you generate leads from Quora by searching for relevant posts and extracting user information.")
# Sidebar for API keys and number of links
with st.sidebar: with st.sidebar:
st.header("API Keys") st.header("API Keys")
firecrawl_api_key = st.text_input("Firecrawl API Key", type="password") firecrawl_api_key = st.text_input("Firecrawl API Key", type="password")
st.caption(" Get your Firecrawl API key from [Firecrawl's website](https://www.firecrawl.dev/app/api-keys)") st.caption(" Get your Firecrawl API key from [Firecrawl's website](https://www.firecrawl.dev/app/api-keys)")
openai_api_key = st.text_input("OpenAI API Key", type="password") openai_api_key = st.text_input("OpenAI API Key", type="password")
st.caption(" Get your Composio API key from [Composio's website](https://composio.ai)")
composio_api_key = st.text_input("Composio API Key", type="password")
st.caption(" Get your OpenAI API key from [OpenAI's website](https://platform.openai.com/api-keys)") st.caption(" Get your OpenAI API key from [OpenAI's website](https://platform.openai.com/api-keys)")
composio_api_key = st.text_input("Composio API Key", type="password")
st.caption(" Get your Composio API key from [Composio's website](https://composio.ai)")
# Add a numeric input for the number of links
num_links = st.number_input("Number of links to search", min_value=1, max_value=10, value=3) num_links = st.number_input("Number of links to search", min_value=1, max_value=10, value=3)
# Reset button
if st.button("Reset"): if st.button("Reset"):
st.session_state.clear() st.session_state.clear()
st.experimental_rerun() st.experimental_rerun()
# Main input for company description user_query = st.text_area(
company_description = st.text_input("Enter your company description or the niche you want to find leads in:", placeholder="e.g. AI voice cloning, Video Generation AI tools") "Describe what kind of leads you're looking for:",
placeholder="e.g., Looking for users who need automated video editing software with AI capabilities",
help="Be specific about the product/service and target audience. The AI will convert this into a focused search query."
)
if st.button("Generate Leads"): if st.button("Generate Leads"):
if not all([firecrawl_api_key, openai_api_key, composio_api_key, company_description]): if not all([firecrawl_api_key, openai_api_key, composio_api_key, user_query]):
st.error("Please fill in all the API keys and the company description.") st.error("Please fill in all the API keys and describe what leads you're looking for.")
else: else:
with st.spinner("Processing your query..."):
transform_agent = create_prompt_transformation_agent(openai_api_key)
company_description = transform_agent.run(f"Transform this query into a concise 3-4 word company description: {user_query}")
st.write("🎯 Searching for:", company_description.content)
with st.spinner("Searching for relevant URLs..."): with st.spinner("Searching for relevant URLs..."):
urls = search_for_urls(company_description, firecrawl_api_key, num_links) # Pass num_links to search_for_urls urls = search_for_urls(company_description.content, firecrawl_api_key, num_links)
if urls: if urls:
st.subheader("Quora Links Used:") st.subheader("Quora Links Used:")