377 lines
19 KiB
Python
377 lines
19 KiB
Python
import json
|
|
import os
|
|
from langchain_community.utilities import SerpAPIWrapper
|
|
from pydantic import BaseModel, Field as PydanticField # Alias it to avoid conflicts
|
|
from crewai.tools import tool
|
|
from typing import List, Dict, Any
|
|
from shared_config import verbose, logger, useSerpAPI
|
|
from crewai import Agent, Task, Crew, Process, LLM, TaskOutput, TaskOutput
|
|
from crewai.tools import tool
|
|
from langchain_community.utilities import DuckDuckGoSearchAPIWrapper
|
|
from datetime import datetime
|
|
import requests
|
|
|
|
import asyncio
|
|
from fastmcp.client import Client
|
|
from fastmcp.client.transports import StreamableHttpTransport
|
|
|
|
|
|
###########################################################################################
|
|
#Define a callback function to log CrewAI step task
|
|
###########################################################################################
|
|
def log_task_to_file(output:TaskOutput):
|
|
"""Callback function to append task details into the log file."""
|
|
|
|
formatted_str = f"""
|
|
{'\n'}CrewAI Task Info
|
|
{"=" * 50}
|
|
Timestamp: {datetime.now()}
|
|
Task Description: {output.description}
|
|
Agent Assigned: {output.agent}
|
|
{"-" * 30}
|
|
Raw Output:{'\n'}
|
|
{output.raw}{'\n'}
|
|
{"=" * 50}
|
|
"""
|
|
logger.info(formatted_str)
|
|
|
|
def create_crew(config):
|
|
|
|
# ==========================================
|
|
# 1. DEFINE CUSTOM TOOLS & OUTPUT SCHEMAS
|
|
# ==========================================
|
|
|
|
class CompetitorDetail(BaseModel):
|
|
company_name: str = PydanticField(..., description="The name of the competitor company.")
|
|
pricing_range: str = PydanticField(..., description="Pricing tiers or estimated cost ranges.")
|
|
core_features: List[str] = PydanticField(..., description="Key modules, tools, or features offered.")
|
|
target_audience: str = PydanticField(..., description="The primary customer profile or ideal user persona.")
|
|
|
|
class ResearchEvidence(BaseModel):
|
|
source_url: str = PydanticField(..., description="The source URL of the evidence.")
|
|
claim: str = PydanticField(..., description="The factual finding or data point.")
|
|
context: str = PydanticField(..., description="The context or time frame of this data point.")
|
|
|
|
class ResearchOutputSchema(BaseModel):
|
|
# CHANGED: Replaced List[Dict[str, Any]] with List[CompetitorDetail]
|
|
competitors: List[CompetitorDetail] = PydanticField(..., description="List of competitors with products pricing range, features, and target audience.")
|
|
market_trends: List[str] = PydanticField(..., description="Market Analysis & Growth Trends")
|
|
evidence_citations: List[ResearchEvidence] = PydanticField(..., description="Valid source citations for facts collected.")
|
|
|
|
if useSerpAPI:
|
|
#Define SerpAPI Search Tool
|
|
@tool("Web Search Tool")
|
|
def web_search_tool(search_query: str) -> str:
|
|
"""
|
|
Search Google for real-time information, news, and facts using SerpAPI.
|
|
Input should be a simple text search query string.
|
|
"""
|
|
# Strip potential quotes added by the agent to clean up the query
|
|
search_query = search_query.strip("'\"")
|
|
if not search_query:
|
|
logger.info(f"Web Search tool used. Error: Empty query")
|
|
return "Error: Received an empty search query."
|
|
|
|
url = "https://serpapi.com/search"
|
|
payload = {
|
|
"q": search_query,
|
|
"api_key": os.getenv("SERPAPI_API_KEY"),
|
|
"engine": "google",
|
|
"num": 10
|
|
}
|
|
|
|
try:
|
|
response = requests.get(url, params=payload, timeout=10)
|
|
|
|
# 1. Check HTTP Status (e.g., 403 Forbidden, 400 Bad Request)
|
|
if response.status_code != 200:
|
|
logger.info(f"Web Search tool used. API Error (Status {response.status_code}): {response.text}")
|
|
return f"API Error (Status {response.status_code}): {response.text}"
|
|
|
|
# 2. Safely parse JSON
|
|
try:
|
|
data = response.json()
|
|
except ValueError:
|
|
logger.info(f"Web Search tool used. Error: Server returned non-JSON data. Raw Response: {response.text[:200]}")
|
|
return f"Error: Server returned non-JSON data. Raw Response: {response.text[:200]}"
|
|
|
|
# 3. Handle backend errors sent inside valid JSON
|
|
if "error" in data:
|
|
logger.info(f"Web Search tool used. SerpAPI Error: {data['error']}")
|
|
return f"SerpAPI Error: {data['error']}"
|
|
|
|
organic_results = data.get("organic_results", [])
|
|
if not organic_results:
|
|
logger.info(f"Web Search tool used. No organic results found for query: '{search_query}'")
|
|
return f"No organic results found for query: '{search_query}'"
|
|
|
|
results_summary = []
|
|
for result in organic_results:
|
|
title = result.get("title")
|
|
link = result.get("link")
|
|
snippet = result.get("snippet")
|
|
results_summary.append(f"Title: {title}\nLink: {link}\nSnippet: {snippet}\n---")
|
|
|
|
strResult = "\n".join(results_summary)
|
|
logger.info(f"""Web Search tool used.\nQuery: {search_query}\nResult:\n{strResult}""")
|
|
return strResult
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logger.info(f"Web Search tool used. Network connection error occurred: {str(e)}")
|
|
return f"Network connection error occurred: {str(e)}"
|
|
else:
|
|
#Define DuckDuckGo Search Tool
|
|
@tool("Web Search Tool")
|
|
def web_search_tool(query: str) -> str:
|
|
"""
|
|
Search the web for real-time information, news, and facts using DuckDuckGo and return the structured Title, Link, and Snippet for each result.
|
|
|
|
Args:
|
|
query (str): The search query text.
|
|
|
|
Returns:
|
|
str: Text containing Title, Link, and Snippet fields for each search result.
|
|
"""
|
|
# Max results set to 5 for optimal context length
|
|
api_wrapper = DuckDuckGoSearchAPIWrapper()
|
|
|
|
try:
|
|
# Fetch structured list of dictionaries
|
|
results = api_wrapper.results(query, 15)
|
|
if not results:
|
|
logger.info(f"""Web Search tool used. No search results found for""")
|
|
return f"No search results found for: '{query}'"
|
|
|
|
output = []
|
|
for i, res in enumerate(results, start=1):
|
|
item_str = (
|
|
f"Result #{i}\n"
|
|
f"Title: {res.get('title', 'N/A')}\n"
|
|
f"Link: {res.get('link', 'N/A')}\n"
|
|
f"Snippet: {res.get('snippet', 'N/A')}\n"
|
|
f"{'-'*40}"
|
|
)
|
|
output.append(item_str)
|
|
|
|
logger.info(f"""Web Search tool used.\nQuery: {query}\nResult:\n{"\n".join(output)}""")
|
|
return "\n".join(output)
|
|
|
|
except Exception as e:
|
|
logger.info(f"""Web Search tool used. Error executing search: {str(e)}""")
|
|
return f"Error executing search: {str(e)}"
|
|
|
|
@tool("FastMCP Batch Web Content extraction Tool")
|
|
def fastmcp_batch_web_content_extraction_tool(urls: list[str], query: str) -> str:
|
|
"""
|
|
Connects directly to the FastMCP HTTP server to scrape and extract text from
|
|
a list of multiple URLs simultaneously in parallel.
|
|
"""
|
|
async def call_fast_mcp():
|
|
# Initialize client using FastMCP's precise matching transport protocol
|
|
transport = StreamableHttpTransport("http://localhost:8000/mcp/")
|
|
|
|
async with Client(transport) as client:
|
|
# call_tool abstracts away deep JSON-RPC structures
|
|
result = await client.call_tool("batch_web_content_extraction", {"urls": urls, "query": query})
|
|
return result
|
|
|
|
try:
|
|
# Spin up clean execution loop for this worker thread
|
|
mcp_response = asyncio.run(call_fast_mcp())
|
|
|
|
# FastMCP Client response objects expose string contents naturally via .content
|
|
# Ensure we safely extract the raw text string from the MCP content blocks
|
|
if hasattr(mcp_response, "content") and mcp_response.content:
|
|
# The text is inside the first content block object
|
|
raw_content = mcp_response.content[0].text
|
|
else:
|
|
# Fallback to string casting if it's already a plain string
|
|
raw_content = str(mcp_response)
|
|
|
|
# Parse JSON and pretty-format the string back to the CrewAI Agent
|
|
try:
|
|
parsed_data = json.loads(raw_content)
|
|
formatted_output = []
|
|
for item in parsed_data:
|
|
formatted_output.append(f"=== SOURCE URL: {item['url']} ===")
|
|
formatted_output.append(f"STATUS: {item['status']}")
|
|
formatted_output.append(f"CONTENT:\n{item['content']}\n")
|
|
formatted_output.append("=" * 40 + "\n")
|
|
|
|
logger.info(f"""FastMCP Batch Web Content extraction tool used.\nURLs:\n{"\n".join(urls)}\nQuery: {query}\nResult:\n{"\n".join(formatted_output)}""")
|
|
return "\n".join(formatted_output)
|
|
except json.JSONDecodeError:
|
|
logger.info(f"""FastMCP Batch Web Content extraction tool used. Error in JSON formatting so raw data returned:\nURLs:\n{"\n".join(urls)}\nQuery: {query}\nResult:\n{raw_content}""")
|
|
return raw_content
|
|
|
|
except Exception as e:
|
|
logger.info(f"""FastMCP Batch Web Content extraction tool used. Error executing the tool and no data returned.""")
|
|
return f"Error executing Parallel FastMCP tool over Streamable HTTP: {str(e)}"
|
|
|
|
|
|
# ---------------------------------------------------------
|
|
#Create llm instance
|
|
# ---------------------------------------------------------
|
|
if config.provider == "OpenAI":
|
|
llm = LLM(
|
|
model=config.model,
|
|
api_key=os.getenv("OPENAI_API_KEY"),
|
|
temperature=config.temperature if config.model != "gpt-5-mini" else 1,
|
|
#reasoning_effort="low",
|
|
max_completion_tokens=config.max_tokens,
|
|
)
|
|
else:
|
|
llm = LLM(
|
|
model=f"ollama/{config.model}",
|
|
api_key="ollama",
|
|
base_url="http://localhost:11434",
|
|
temperature=config.temperature
|
|
)
|
|
|
|
logger.info(f"""LLM Model: {config.model}, Provider: {config.provider}, Temperature: {config.temperature}""")
|
|
|
|
# ---------------------------------------------------------
|
|
# 1. Agent Definitions
|
|
# ---------------------------------------------------------
|
|
|
|
# 1. Orchestrator & Documenter (Manager)
|
|
head_planner = Agent(
|
|
role='Head Planner and Orchestrator',
|
|
goal="Synthesize intermediary outputs and assemble the final master GTM document.",
|
|
backstory="A precise compiler and documentarian...",
|
|
allow_delegation=False,
|
|
verbose=verbose,
|
|
llm=llm
|
|
)
|
|
|
|
# 2. researcher
|
|
researcher_agent = Agent(
|
|
role="Market Research Specialist",
|
|
goal ="Gather search market data and extract multiple deep resources simultaneously to speed up workflows. Identify industry trends, pricing tiers, core features, and target customer profiles, while ensuring all data is verifiable with source URLs.",
|
|
backstory="""You are an elite Market Research Specialist. You are an expert at mining deep web data and finding precise statistical facts from modern industry reports.
|
|
You use your Web Search tool to find matching links, and then pass all relevant URLs at the exact same time into your batch web content extraction tool for immediate, parallel analysis.""",
|
|
verbose=verbose,
|
|
tools=[web_search_tool, fastmcp_batch_web_content_extraction_tool],
|
|
use_system_prompt=False,
|
|
llm=llm,
|
|
)
|
|
|
|
# 3. analyst
|
|
analyst_agent = Agent(
|
|
role="Strategic Business Analyst",
|
|
goal="Transform unstructured research data into rigorous market models, SWOT matrixes, and 4P/7P frameworks.",
|
|
backstory="A former MBB consultant who specializes in corporate strategy. You find patterns in chaotic data, flag market gaps, and synthesize pricing elasticities.",
|
|
verbose=verbose,
|
|
use_system_prompt=False,
|
|
llm=llm
|
|
)
|
|
|
|
# 4. strategist
|
|
strategist_agent = Agent(
|
|
role="Go-To-Market (GTM) Strategist",
|
|
goal="Draft the final actionable GTM blueprint including ICP maps, messaging layers, and launch milestone phases.",
|
|
backstory="A legendary growth marketing executive. You turn raw analytical data into high-converting messaging matrixes, channel frameworks, and scalable product launch schedules.",
|
|
verbose=verbose,
|
|
use_system_prompt=False,
|
|
llm=llm
|
|
)
|
|
# ---------------------------------------------------------
|
|
# 2. Task Definitions
|
|
# ---------------------------------------------------------
|
|
|
|
#Fetch User query from config
|
|
query = config.user_query
|
|
|
|
# Task 1: Research Agent gathers raw intelligence
|
|
research_task = Task(
|
|
description=("""
|
|
Research for competitor compagnies in the target market: {query}
|
|
Use your Web Search tool to find relevant and recent company, product, industry, and market sources. Do not add a year to the search query. Remove broken, duplicate, or irrelevant URLs.
|
|
Create a focused MCP query that describes the information needed, including products, pricing, key features, target customers, market trends, growth, and competitive information.
|
|
Send the selected URLs and the focused query to the FastMCP Batch Web Content Extraction Tool in a single call.
|
|
Analyze the returned content and produce:
|
|
Companies/products: company, products, price range, key features, and target audience. Do not duplicate companies.
|
|
Market analysis: market characteristics, growth, trends, drivers, and competitive developments.
|
|
Evidence: every factual claim or metric must have a source URL, factual finding, and context/timeframe.
|
|
Prefer primary and authoritative sources. Do not invent information. If information is unavailable, state that it was not found.
|
|
Critical: Every factual claim, metric, price, feature, market statistic, or growth figure must be traceable to a source URL.
|
|
Return valid JSON:
|
|
"""
|
|
),
|
|
expected_output="A structured JSON payload containing validated competitor profiles, trend metrics, and verifiable source citations.",
|
|
output_json=ResearchOutputSchema,
|
|
agent=researcher_agent
|
|
)
|
|
|
|
# Task 2: Analyst Agent creates comparative structures
|
|
analysis_task = Task(
|
|
description=(
|
|
"1. Extract and read the text/JSON-formatted market research data passed to you from the Research Agent in the context below.\n"
|
|
"2. Using that data, construct a detailed markdown competitive landscape table and a pricing matrix. Ensure every row includes its respective source URL.\n"
|
|
"3. Perform a comprehensive SWOT, 4P, and 7P analysis based strictly on the facts and evidence compiled by the Research Agent.\n"
|
|
"4. Do not speculate or invent market details not present in the research context data."
|
|
),
|
|
expected_output="A deep-dive analytical report featuring clean markdown tables, a SWOT quadrant, and an exhaustive 4P/7P matrix analysis.",
|
|
context=[research_task],
|
|
agent=analyst_agent
|
|
)
|
|
|
|
# Task 3: Strategy Agent plans market entry
|
|
strategy_task = Task(
|
|
description=(
|
|
"Take the analytical matrices and SWOT outputs to build a comprehensive GTM document. "
|
|
"Define 2 distinct Ideal Customer Profiles (ICPs). Outline the core value proposition. "
|
|
"Create a messaging framework (Hook, Problem, Solution) per ICP. "
|
|
"Identify high-ROI acquisition channels and detail a 30-60-90 day milestone launch plan." \
|
|
),
|
|
expected_output="An actionable, comprehensive market entry strategy blueprint divided into logical execution phases.",
|
|
context=[analysis_task],
|
|
agent=strategist_agent
|
|
)
|
|
|
|
# Task 4: Head Planner synthesizes, formats, and exports
|
|
compilation_task = Task(
|
|
description=(
|
|
"""Synthesize the outputs from the Research, Analyst, and Strategy agents into a seamless, high-caliber Master GTM Report.
|
|
Include all research data tables, citations, SWOT/7P strategic matrices, and execution timelines.
|
|
Include a title
|
|
Inlude small summary that cleary identifies the target market including the input query '{query}' for reference.
|
|
Include a table of contents.
|
|
Include a conclusion and Evidence Citations section at the end.
|
|
Ensure the document is formatted for executive-level presentation.
|
|
[OUTPUT FORMAT SPECIFICATION]:
|
|
Format your entire document exclusively in clean, standard Markdown.
|
|
|
|
Strict Structural Formatting Rules:
|
|
1. DOCUMENT TITLE & SUMMARY: Start immediately with the document title as a level-1 heading (# Title) and your summary sections. Do NOT include numbers on these meta-sections.
|
|
2. MAIN SECTIONS & SUBSECTIONS: You must explicitly number all research content headings manually (e.g., "# 1. Competitive Landscape Analysis", "## 1.1 Competitive Landscape Table", "# 2. Go-To-Market Strategy").
|
|
3. TABLE OF CONTENTS: Directly after your summary block, include a manually written, clean Markdown Table of Contents listing your explicitly numbered headings. Example:
|
|
### Table of Contents
|
|
- [1. Competitive Landscape Analysis](#1-competitive-landscape-analysis)
|
|
- [1.1 Competitive Landscape Table](#11-competitive-landscape-table)
|
|
- [1.2 Pricing Matrix](#12-pricing-matrix)
|
|
- [2. Go-To-Market (GTM) Strategy Blueprint](#2-go-to-market-gtm-strategy-blueprint)
|
|
4. NO MARKDOWN TOC MARKERS: Do NOT output the text string "[TOC]". Write out the table of contents yourself using bullet points and standard markdown links."""
|
|
),
|
|
expected_output="Final master GTM document.",
|
|
context=[research_task, analysis_task, strategy_task],
|
|
agent=head_planner
|
|
)
|
|
|
|
# ---------------------------------------------------------
|
|
# 3. Crew Setup
|
|
# ---------------------------------------------------------
|
|
|
|
crew= Crew(
|
|
agents=[researcher_agent, analyst_agent, strategist_agent, head_planner],
|
|
tasks=[research_task, analysis_task, strategy_task, compilation_task],
|
|
process=Process.sequential, # Tasks execute in exact sequence
|
|
#manager_agent=head_planner
|
|
task_callback=log_task_to_file,
|
|
verbose=verbose
|
|
)
|
|
|
|
logger.info(f"Crew setup complete with {len(crew.agents)} agents and {len(crew.tasks)} tasks.")
|
|
return crew
|
|
|