import os 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 from crewai import Agent, Task, Crew, Process, LLM, TaskOutput, TaskOutput from crewai.tools import tool from langchain_community.tools import DuckDuckGoSearchRun from datetime import datetime import os ########################################################################################### #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 # ========================================== # Simple search tool for the Research Agent @tool("web_search") def web_search_tool(query: str) -> str: """Search the web for current events, news, or factual information.""" ddg = DuckDuckGoSearchRun() return ddg.invoke(query) 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): competitors: List[Dict[str, Any]] = PydanticField(..., description="List of competitors with pricing, features, and target audience.") market_trends: List[str] = PydanticField(..., description="Key macro trends observed.") evidence_citations: List[ResearchEvidence] = PydanticField(..., description="Valid source citations for facts collected.") # --------------------------------------------------------- #Create llm instance # --------------------------------------------------------- if config.provider == "OpenAI": llm = LLM( model=config.model, api_key=os.getenv("OPENAI_API_KEY"), temperature=config.temperature, 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...", #goal='Orchestrate the GTM workflow, delegate tasks to specialized agents, and synthesize all insights into a flawless final GTM document.', #backstory='You are a Principal Operations Strategy Director. You excel at managing cross-functional teams, ensuring strict alignment, and formatting messy raw data into executive-ready corporate documentation.', allow_delegation=False, verbose=verbose, llm=llm ) # 2. researcher researcher_agent = Agent( role="Market Research Specialist", goal="Gather raw competitive intelligence, target audience signals, and pricing tiers with structural citations.", backstory="An elite OSINT researcher who tracks digital signals. You ignore marketing fluff and extract verifiable numbers, product capabilities, and structural URLs.", #role='Research Agent', #goal='Gather comprehensive market data, identify industry trends, and map out the competitive landscape.', #backstory='You are an elite Market Research Analyst. You are an expert at mining deep web data, extracting customer pain points, and finding precise statistical facts from modern industry reports.', verbose=verbose, tools=[web_search_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.", #role='Analyst Agent', #goal='Synthesize raw market data into structured frameworks like SWOT, 4P, 7P, and target user personas.', #backstory='You are a data-driven Strategic Analyst. You look past surface-level facts to find underlying market gaps, evaluate competitive positioning, and model risks.', verbose=verbose, 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.", #role='Strategy Agent', #goal='Formulate the actionable Go-To-Market blueprint, positioning, pricing strategies, and launch timelines.', #backstory='You are a veteran GTM Growth Strategist. You specialize in crafting commercial launch playbooks, positioning products uniquely against rivals, and designing user-acquisition loops.', verbose=verbose, 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=( "Search the web for the top 3-5 players in the target market: '{query}'. " "Extract pricing tiers, core feature modules, and target customer profiles. " "Every single metric or claim MUST have a URL source tracked inside the JSON structure. " "Do not fabricate any data and do not make assumptions. Only include verifiable facts using working URLs." "If you cannot find a specific metric, leave it blank and do not fabricate data." ), expected_output="A structured JSON file matching the schema with validated competitive rows and strict citation tracking.", output_json=ResearchOutputSchema, agent=researcher_agent ) # Task 2: Analyst Agent creates comparative structures analysis_task = Task( description=( "Review the structured JSON output provided by the Research Agent. " "Construct a detailed markdown competitive landscape table and a pricing matrix. " "Perform a comprehensive SWOT, 4P, and 7P analysis based strictly on the evidence compiled." ), expected_output="A deep-dive analytical report featuring clean markdown tables, a SWOT quadrant, and an exhaustive 4P/7P matrix analysis.", 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.", 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 that cleary identifies the target market and a subtitle that highlights the core value proposition. " ), 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