Log inGet Started

Building a Full-Stack AI Shopping Assistant with CrewAI and Tavily

AI Summary10 min read

TL;DR

This guide shows how to build a full-stack AI shopping assistant using CrewAI Flows, Tavily API for product searches, and CopilotKit for the frontend. It covers backend setup with Python and frontend integration with Next.js.

Key Takeaways

  • •Use CrewAI Flows to create flexible, event-driven workflows for AI agents in shopping applications.
  • •Integrate Tavily API for reliable, multi-retailer product searches and data extraction.
  • •Leverage CopilotKit for state management and building interactive frontends in React or Next.js.

Tags

webdevprogrammingaiproductivity

TL;DR

In this guide, you will learn how to build a full-stack AI shopping assistant using CrewAI Flows, paired with the Tavily Search API in the backend. We'll then walk through adding a frontend to the shopping agent using CopilotKit to interact with it.

The AI shopping assistant will be able to search products on commerce platforms such as Amazon, Target, or eBay- extract structured product data and provide tailored product recommendations.

Before we jump in, here is what we will cover:

  • What are CrewAI Flows?

  • Building the AI shopping assistant backend using CrewAI, Tavily, and CopilotKit

  • Building the AI shopping assistant frontend using CopilotKit.

Here is a preview of what we will be building:

What are CrewAI Flows?

Flows are a flexible building block in CrewAI that lets you control things at a detailed level while keeping it simple overall. You can use them to automate tasks ranging from a basic call to an AI model all the way to a team of smart AI agents that work on their own.

With Flows, it's straightforward to create and set up step-by-step processes that make the most of what CrewAI can do.

Here are some key features:

  • Easy Workflow Building: Quickly link up different teams of AI agents (called Crews) and individual tasks to make advanced AI systems.

  • Handling Shared Information: Flows make it really simple to keep track of and share data between various steps in your process.

  • Event-Based Design: It's built around reacting to events, which helps create workflows that adapt and respond quickly.

  • Customizable Paths: Add rules like "if this happens, then do that," repeat steps in loops, or split into different branches in your workflows.

You can learn more about CrewAI Flows here on CrewAI docs.

Image from Notion

Now that we have learned what the CrewAI Flows are, let us see how to build the CrewAI shopping assistant from the backend to the frontend.

Let’s get started!

Prerequisites

To fully understand this tutorial, you need to have a basic understanding of React or Next.js.

We'll also make use of the following:

  • Python - a popular programming language for building AI agents with LangGraph; make sure it is installed on your computer.

  • OpenAI API - to enable us to perform various tasks using the GPT models; for this tutorial, ensure you have access to the GPT-4 model.

  • CopilotKit - an open-source copilot framework for building custom AI chatbots, in-app AI agents, and text areas.

  • CrewAI - a Python framework that enables developers to create autonomous AI agents with high-level simplicity and precise low-level control.

  • Tavily API - a search engine built specifically for AI agents (LLMs), delivering real-time, accurate, and factual results at speed.

Setting up the project

To get started, clone the CrewAI-Shopping-Assistant repo that consists of a Python-based backend (agent) and a Next.js frontend (frontend).

Next, navigate to the backend directory:

cd agent
Enter fullscreen mode Exit fullscreen mode

Then install the dependencies using Poetry:

poetry install
Enter fullscreen mode Exit fullscreen mode

After that, create a .env file with OpenAI API Key and Tavily API Key:

OPENAI_API_KEY=<<your-OpenAI-key-here>>
TAVILY_API_KEY=<<your-tavily-key-here>>
Enter fullscreen mode Exit fullscreen mode

Then run the agent using the command below:

poetry run python main.py
Enter fullscreen mode Exit fullscreen mode

After that, navigate to the frontend directory:

cd frontend
Enter fullscreen mode Exit fullscreen mode

Next, create a .env file with OpenAI API Key API key:

OPENAI_API_KEY=<<your-OpenAI-key-here>>
Enter fullscreen mode Exit fullscreen mode

Then install the dependencies:

pnpm install
Enter fullscreen mode Exit fullscreen mode

After that, start the development server:

pnpm run dev
Enter fullscreen mode Exit fullscreen mode

Navigate to http://localhost:3000, and you should see the AI shopping assistant UI up and running.

Image from Notion

Let’s now see how to build the AI shopping assistant backend using CrewAI, Tavily, and CopilotKit.

Building the AI shopping assistant backend using CrewAI, Tavily, and CopilotKit

In this section, you will learn how to build the AI shopping assistant backed using CrewAI for agent flows, Tavily for reliable web search, and CopilotKit for state management + UI updates.

Let’s jump in.

Step 1: Define the agent state

First, define the AgentState class that extends CopilotKitState in order to track the product data, user preferences, logs for UI updates, and more, as shown in the agent/shopping_assistant.py file.

from copilotkit.crewai import CopilotKitState
from typing import List, Dict, Any

class AgentState(CopilotKitState):
    """
    Manages the complete state of the shopping workflow.
    """
    # List of products currently displayed on the canvas
    products: List = []

    # List of users' favorite/saved products
    favorites: List = []

    # Temporary buffer to store products before confirmation
    buffer_products: List = []

    # User's wishlist of products
    wishlist: List = []

    # Activity logs to show processing status to the user
    logs: List = []

    # Generated comparison report for products
    report: Any | None = None

    # Flag to control when to show results in UI
    show_results: bool = False

    # Canvas logging information with title and subtitle for UI updates
    canvas_logs: dict = { "title" : "", "subtitle" : "" }
Enter fullscreen mode Exit fullscreen mode

Step 2: Create the Shopping Agent Workflow

Once you have defined the agent state, define the ShoppingAgentFlow class that extends the Flow[AgentState]. Use the start() method to handle the entire flow, where it initializes state, validates environment variables, processes user queries, and emits state updates via CopilotKit, as shown in the ./agent/shopping_assistant.py file.

class ShoppingAgentFlow(Flow[AgentState]):

    @start()
    async def start(self):
        """
        Main entry point for the shopping assistant workflow.
        This method handles the entire flow from user request to product results.
        """
        try:
            # Step 1: Initialize the workflow
            print("Starting Shopping Agent Flow")

            # Update canvas with initial status
            self.state.canvas_logs = {
                "title" : f"Parsing your request",
                "subtitle" : "Deciding to run product search or not"
            }
            await copilotkit_emit_state(self.state)
            await asyncio.sleep(0)

            # Step 2: Validate required environment variables
            if not os.getenv("TAVILY_API_KEY"):
                raise RuntimeError("Missing TAVILY_API_KEY")
            if not os.getenv("OPENAI_API_KEY"):
                raise RuntimeError("Missing OPENAI_API_KEY")

            # Step 3: Check if this is a report generation request (assistant message)
            if self.state.messages[-1]['role'] == 'assistant':
                # Generate and return product comparison report
                result =await generate_report(self.state.products)
                print(result, "result")
                self.state.report = json.loads(result)
                await copilotkit_emit_state(self.state)
                return

            # Step 4: Add initial processing log
            self.state.logs.append({
                "message" : "Analyzing user query",
                "status" : "processing"
            })
            await copilotkit_emit_state(self.state)

            # Step 5: Mark analysis as completed
            self.state.logs[-1]["status"] = "completed"
            await copilotkit_emit_state(self.state)

            // ...
Enter fullscreen mode Exit fullscreen mode

Step 3: Multi-Retailer Product Search with Tavily API

After creating the shopping agent flow, initialize the Tavily client and prepare for multi-retailer search, as shown below.

class ShoppingAgentFlow(Flow[AgentState]):

    @start()
    async def start(self):
        """
        Main entry point for the shopping assistant workflow.
        This method handles the entire flow from user request to product results.
        """
        try:
            // ...

            # Step 13: Set up Tavily search client and result containers
            tv = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
            results_all: List[Dict[str, Any]] = []
            total_mappings_list = []

            # Step 14: Update user with search progress
            self.state.logs.append({
                "message" : "Identifying the sites to search",
                "status" : "processing"
            })
            self.state.canvas_logs={
                "title" : "Identifying the sites to search",
                "subtitle" : "Tavily search in progress...."
            }
            await copilotkit_emit_state(self.state)
            await asyncio.sleep(1)
            self.state.logs[-1]["status"] = "completed"
            await copilotkit_emit_state(self.state)

            // ...
Enter fullscreen mode Exit fullscreen mode

Then search each major retailer using Tavily's domain-specific search capabilities to get relevant product URLs from each major retailer, as shown below.

class ShoppingAgentFlow(Flow[AgentState]):

    @start()
    async def start(self):
        """
        Main entry point for the shopping assistant workflow.
        This method handles the entire flow from user request to product results.
        """
        try:
            // ...

            # Step 15: Search across major retailers for product URLs
                        urls = {}
                        for retailer in RETAILERS:  # RETAILERS = ["target.com", "amazon.com", "ebay.com"]
                            # Search each retailer's domain for products matching the query
                            search = tv.search(
                                query=query,  # User's search query, e.g., "wireless headphones"
                                include_domains=[retailer],  # Limit search to specific retailer
                                include_answer=False,  # Don't include general answers
                                include_images=False,  # Don't include image results
                                include_raw_content=False,  # We'll extract content separately
                                search_depth="advanced",  # Use advanced search for better results
                                max_results=max_search_results,  # Typically 6 results per retailer
                            )

                            # Extract URLs from search results
                            urls[retailer] = [r["url"] for r in search.get("results", []) if r.get("url")]
                            if not urls[retailer]:
                                continue            
            // ...
Enter fullscreen mode Exit fullscreen mode

After gathering URLs, use Tavily's extraction capabilities to get detailed webpage content that includes images and metadata, as shown below.

class ShoppingAgentFlow(Flow[AgentState]):

    @start()
    async def start(self):
        """
        Main entry point for the shopping assistant workflow.
        This method handles the entire flow from user request to product results.
        """
        try:
            // ...

            # Step 17: Define a function for parallel URL content extraction
                        def extract_urls(urls: List[str], retailer: str) -> Dict[str, Any]:
                            """Extract content from URLs for a specific retailer"""
                            try:
                                print(f"Extracting urls for {retailer}. Started at {datetime.now()}")
                                ext1 = tv.extract(urls, extract_depth="advanced", include_images=True, timeout=120)
                                return [ext1, retailer]
                            except Exception as e:
                                print(f"Error extracting urls: {e}")
                                return None

                        # Step 18: Execute parallel extraction for all retailers
                        ext_results = {}
                        with ThreadPoolExecutor(max_workers=3) as executor:
                            # Submit extraction tasks for each retailer
                            futures = {executor.submit(extract_urls, urls[retailer], retailer) : retailer for retailer in RETAILERS}

                            # Collect results as they complete
                            for future in as_completed(futures):
                                result = future.result()
                                ext_results[result[1]] = result[0].get("results", [])
                                if result == None:
                                    print("Condition met! Cancelling remaining tasks...")
                                    # Cancel remaining futures if an error occurs
                                    for f in futures:
                                        f.cancel()
                                    break
            // ...
Enter fullscreen mode Exit fullscreen mode

Finally, process the extracted product data with LLMs to extract product information, as shown below.

class ShoppingAgentFlow(Flow[AgentState]):

    @start()
    async def start(self):
        """
        Main entry point for the shopping assistant workflow.
        This method handles the entire flow from user request to product results.
        """
        try:
            // ...

            # Step 20: Initialize data structures for product processing
            target_listing_pdps: List[str] = []
            done = False
            self.state.logs.append({
                "message" : "Processing the data",
                "status" : "processing"
            })
            await copilotkit_emit_state(self.state)

            # Initialize product containers for each retailer
            products_from_each_site= {
                "target.com" : [],
                "amazon.com" : [],

Visit Website