CrewAI Agent in Machine Learning

from crewai import Agent, Task, Crew, Process

# Define agents with role, goal, and backstory
researcher = Agent(
    role="Senior Market Research Analyst",
    goal="Uncover emerging trends and competitive insights in the Indian fintech market",
    backstory=(
        "You are a seasoned market analyst with 12 years of experience tracking "
        "the Indian fintech ecosystem. You've covered the rise of UPI, the growth "
        "of Razorpay and PhonePe, and the regulatory shifts by RBI. You are known "
        "for your data-driven, contrarian insights."
    ),
    verbose=True,
    allow_delegation=False,
    llm="gpt-4o",
)

writer = Agent(
    role="Content Strategist",
    goal="Transform research findings into a compelling market brief for investors",
    backstory=(
        "You are a financial content strategist who has written for Bloomberg Quint "
        "and The Ken. You excel at distilling complex fintech data into crisp, "
        "actionable narratives that appeal to Series A and B investors."
    ),
    verbose=True,
    allow_delegation=False,
    llm="gpt-4o",
)

# Define tasks
research_task = Task(
    description=(
        "Research the top 5 emerging trends in Indian fintech for 2026. "
        "Focus on UPI market share shifts, BNPL regulation impacts, "
        "and cross-border payment innovations. Use recent data from RBI "
        "and NPCI reports."
    ),
    expected_output="A detailed research brief with 5 trends, each supported by data points and sources.",
    agent=researcher,
)

writing_task = Task(
    description=(
        "Using the research brief, write a 1500-word investor-facing market "
        "analysis of Indian fintech trends. Include a competitive landscape "
        "table and actionable investment themes."
    ),
    expected_output="A polished market analysis document in markdown format.",
    agent=writer,
    context=[research_task],  # This task receives the researcher's output
)

# Create and run the crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print(result)

# config/agents.yaml
researcher:
  role: "Senior Market Research Analyst"
  goal: "Uncover emerging trends and competitive insights in {industry}"
  backstory: >
    You are a seasoned market analyst with 12 years of experience.
    You specialize in {industry} across the Indian market.
    You are known for data-driven, contrarian insights backed by
    primary sources like RBI circulars and SEBI filings.
  llm: gpt-4o
  max_iter: 15
  allow_delegation: false

writer:
  role: "Content Strategist"
  goal: "Transform research into compelling {output_format} for {audience}"
  backstory: >
    You are a financial content strategist who has written for
    Bloomberg Quint and The Ken. You distill complex data into
    crisp, actionable narratives.
  llm: gpt-4o-mini
  allow_delegation: false

reviewer:
  role: "Quality Assurance Editor"
  goal: "Ensure factual accuracy, logical consistency, and professional tone"
  backstory: >
    You are a meticulous editor with a background in fact-checking
    for financial publications. You catch errors that others miss.
  llm: gpt-4o
  allow_delegation: true

# config/tasks.yaml
research_task:
  description: >
    Research the top {num_trends} emerging trends in {industry}.
    Focus on regulatory changes, market share shifts, and
    technology innovations. Use recent data from official sources.
  expected_output: "Detailed research brief with {num_trends} trends, each with data points."
  agent: researcher

writing_task:
  description: >
    Using the research brief, write a {word_count}-word {output_format}
    for {audience}. Include a competitive landscape table.
  expected_output: "Polished {output_format} in markdown format."
  agent: writer
  context:
    - research_task

review_task:
  description: >
    Review the {output_format} for factual accuracy, logical flow,
    and professional tone. Flag any unsubstantiated claims.
    Provide a final score out of 10.
  expected_output: "Reviewed document with inline corrections and quality score."
  agent: reviewer
  context:
    - writing_task

from crewai import Agent, Task, Crew, Process

# Worker agents
data_collector = Agent(
    role="Data Collection Specialist",
    goal="Gather structured data from APIs and web sources",
    backstory="Expert at scraping and API integration with 8 years at a Bengaluru data analytics firm.",
    tools=[web_search_tool, scraper_tool],
    llm="gpt-4o-mini",  # Cheaper model for data gathering
)

analyst = Agent(
    role="Quantitative Analyst",
    goal="Perform statistical analysis and identify actionable patterns",
    backstory="IIT Bombay statistics graduate with 5 years at Zerodha's quant desk.",
    tools=[code_execution_tool],
    llm="gpt-4o",  # More capable model for analysis
)

report_writer = Agent(
    role="Report Author",
    goal="Produce clear, well-structured analytical reports",
    backstory="Former McKinsey consultant who writes reports that CXOs actually read.",
    llm="gpt-4o-mini",
)

# Custom manager agent (optional -- CrewAI creates one by default)
manager = Agent(
    role="Research Director",
    goal="Coordinate the research team to produce high-quality deliverables on time",
    backstory=(
        "You are a seasoned research director who has managed teams of 20+ analysts. "
        "You excel at task prioritization, quality control, and knowing when to "
        "re-delegate work that doesn't meet standards."
    ),
    llm="gpt-4o",
    allow_delegation=True,
)

# Tasks (agent assignment is optional in hierarchical -- manager decides)
collect_task = Task(
    description="Collect Q3 2025 revenue data for top 10 Indian SaaS companies.",
    expected_output="Structured dataset with company name, revenue, growth rate, and source URL.",
)

analyze_task = Task(
    description="Analyze revenue trends, compute CAGR, and identify top 3 outliers.",
    expected_output="Statistical analysis with visualizations and key findings.",
)

report_task = Task(
    description="Write a 2-page executive summary of the Indian SaaS market analysis.",
    expected_output="Executive summary in markdown with tables and key recommendations.",
)

# Hierarchical crew with custom manager
crew = Crew(
    agents=[data_collector, analyst, report_writer],
    tasks=[collect_task, analyze_task, report_task],
    process=Process.hierarchical,
    manager_agent=manager,
    memory=True,  # Enable short-term, long-term, and entity memory
    verbose=True,
)

result = crew.kickoff()
print(result.raw)

from crewai.flow.flow import Flow, listen, start, router
from crewai import Crew
from pydantic import BaseModel


class MarketReportState(BaseModel):
    industry: str = ""
    research_quality_score: float = 0.0
    research_output: str = ""
    final_report: str = ""


class MarketReportFlow(Flow[MarketReportState]):

    @start()
    def gather_research(self):
        """First step: run the research crew."""
        research_crew = Crew(
            agents=[researcher_agent],
            tasks=[research_task],
            process="sequential",
        )
        result = research_crew.kickoff(
            inputs={"industry": self.state.industry}
        )
        self.state.research_output = result.raw
        return result.raw

    @router(gather_research)
    def evaluate_quality(self):
        """Route based on research quality."""
        # Use a lightweight LLM call to score quality
        score = evaluate_research_quality(self.state.research_output)
        self.state.research_quality_score = score
        if score >= 7.0:
            return "proceed_to_writing"
        else:
            return "redo_research"

    @listen("redo_research")
    def refine_research(self):
        """Re-run research with enhanced instructions."""
        enhanced_crew = Crew(
            agents=[senior_researcher_agent],
            tasks=[enhanced_research_task],
            process="sequential",
        )
        result = enhanced_crew.kickoff(
            inputs={
                "industry": self.state.industry,
                "prior_research": self.state.research_output,
            }
        )
        self.state.research_output = result.raw
        return result.raw

    @listen("proceed_to_writing")
    def write_report(self):
        """Final step: produce the report."""
        writing_crew = Crew(
            agents=[writer_agent, editor_agent],
            tasks=[write_task, edit_task],
            process="sequential",
        )
        result = writing_crew.kickoff(
            inputs={"research": self.state.research_output}
        )
        self.state.final_report = result.raw
        return result.raw


# Execute the flow
flow = MarketReportFlow()
flow.kickoff(inputs={"industry": "Indian fintech"})
print(flow.state.final_report)

from crewai.tools import BaseTool
from pydantic import Field
import requests


class NSEStockPriceTool(BaseTool):
    """Fetch real-time stock price from NSE India."""
    name: str = "NSE Stock Price Lookup"
    description: str = (
        "Fetches the current stock price for an Indian company listed on NSE. "
        "Input should be the NSE stock symbol (e.g., 'RELIANCE', 'TCS', 'INFY')."
    )

    def _run(self, symbol: str) -> str:
        try:
            url = f"https://www.nseindia.com/api/quote-equity?symbol={symbol}"
            headers = {
                "User-Agent": "Mozilla/5.0",
                "Accept": "application/json",
            }
            session = requests.Session()
            session.get("https://www.nseindia.com", headers=headers)
            response = session.get(url, headers=headers)
            data = response.json()
            price_info = data.get("priceInfo", {})
            return (
                f"Symbol: {symbol}\n"
                f"Last Price: INR {price_info.get('lastPrice', 'N/A')}\n"
                f"Change: {price_info.get('change', 'N/A')} "
                f"({price_info.get('pChange', 'N/A')}%)\n"
                f"Day High: INR {price_info.get('intraDayHighLow', {}).get('max', 'N/A')}\n"
                f"Day Low: INR {price_info.get('intraDayHighLow', {}).get('min', 'N/A')}"
            )
        except Exception as e:
            return f"Error fetching stock data for {symbol}: {str(e)}"


# Use the custom tool with an agent
nse_tool = NSEStockPriceTool()

financial_analyst = Agent(
    role="Indian Equity Analyst",
    goal="Analyze stock performance of Indian listed companies",
    backstory="CFA charterholder with 10 years analyzing Indian equities at HDFC Securities.",
    tools=[nse_tool],
    llm="gpt-4o",
)

# docker-compose.yaml for production CrewAI deployment
version: '3.8'
services:
  crewai-app:
    build: .
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - CREWAI_MEMORY=true
      - CREWAI_VERBOSE=false
    volumes:
      - ./config:/app/config
      - crew-memory:/app/.crewai
    deploy:
      resources:
        limits:
          memory: 2G
          cpus: '1.0'
    restart: unless-stopped

  chromadb:
    image: chromadb/chroma:latest
    ports:
      - "8000:8000"
    volumes:
      - chroma-data:/chroma/chroma

volumes:
  crew-memory:
  chroma-data:

# Estimated monthly costs (India, ap-south-1):
# - EC2 t3.medium: ~INR 2,500/month ($30)
# - LLM API (10K executions/day, GPT-4o-mini): ~INR 12,500/month ($150)
# - ChromaDB storage: ~INR 500/month ($6)
# Total: ~INR 15,500/month ($186)