Build a containerized AI Agent with watsonx.ai & CrewAI (and Streamlit) and Podman

import sys

import time

import streamlit as st

from crewai import Agent, Task, Crew, Process

from langchain_community.tools import DuckDuckGoSearchRun

from langchain.agents import Tool

from ibm_watsonx_ai.foundation_models import Model

from ibm_watsonx_ai.credentials import Credentials

from langchain.llms.base import LLM

from typing import Any, List, Mapping, Optional



import os

import re

#from crewai import LLM

from dotenv import load_dotenv

load_dotenv()





WATSONX_URL="https://us-south.ml.cloud.ibm.com/"

WATSONX_PROJECT_ID= "YOUR-PROJECT-ID"

WATSONX_APIKEY = "APIKEY"

WATSONX_LLAMA3_MODEL_ID="meta-llama/llama-3-2-3b-instruct"





model_id = 'meta-llama/llama-3-2-3b-instruct'

parameters = {

    "decoding_method": "sample",

    "max_new_tokens": 500,

    "temperature": 0.7,

    "top_k": 50,

    "top_p": 1,

    "repetition_penalty": 1

}

credentials = Credentials(url=WATSONX_URL, api_key=WATSONX_APIKEY)



ibm_model = Model(

    model_id=model_id,

    params=parameters,

    credentials=credentials,

    project_id=WATSONX_PROJECT_ID

)



class IBMWatsonLLM(LLM):

    def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str:

        return ibm_model.generate_text(prompt=prompt, guardrails=False)



    @property

    def _llm_type(self) -> str:

        return "ibm_watson"



    def generate_text(self, prompt):

        return self._call(prompt)

llm = IBMWatsonLLM()





duckduckgo_search = DuckDuckGoSearchRun()



#to keep track of tasks performed by agents

task_values = []



def create_crewai_setup(product_name):

    # Define Agents

    market_research_analyst = Agent(

        role="Market Research Analyst",

        goal=f"""Analyze the market demand for {product_name} and 

                 suggest marketing strategies""",

        backstory=f"""Expert at understanding market demand, target audience, 

                      and competition for products like {product_name}. 

                      Skilled in developing marketing strategies 

                      to reach a wide audience.""",

        verbose=True,

        allow_delegation=True,

        tools=[duckduckgo_search],

        llm=llm,

    )



    technology_expert = Agent(

        role="Technology Expert",

        goal=f"Assess technological feasibilities and requirements for producing high-quality {product_name}",

        backstory=f"""Visionary in current and emerging technological trends, 

                      especially in products like {product_name}. 

                      Identifies which technologies are best suited 

                      for different business models.""",

        verbose=True,

        allow_delegation=True,

        llm=llm,

    )



    business_consultant = Agent(

        role="Business Development Consultant",

        goal=f"""Evaluate the business model for {product_name}, 

               focusing on scalability and revenue streams""",

        backstory=f"""Seasoned in shaping business strategies for products like {product_name}. 

                      Understands scalability and potential 

                      revenue streams to ensure long-term sustainability.""",

        verbose=True,

        allow_delegation=True,

        llm=llm,

    )





    # Define Tasks

    task1 = Task(

        description=f"""Analyze the market demand for {product_name}. Current month is November 2024.

                        Write a report on the ideal customer profile and marketing 

                        strategies to reach the widest possible audience. 

                        Include at least 10 bullet points addressing key marketing areas.""",

        expected_output="Report on market demand analysis and marketing strategies.",

        agent=market_research_analyst,

    )

    # Define Task 2

    task2 = Task(

        description=f"""Assess the technological aspects of manufacturing 

                    high-quality {product_name}. Write a report detailing necessary 

                    technologies and manufacturing approaches. 

                    Include at least 10 bullet points on key technological areas.""",

        expected_output="Report on technological aspects of manufacturing.",

        agent=technology_expert,

    )

    # Define Task 3

    task3 = Task(

        description=f"""Summarize the market and technological reports 

                    and evaluate the business model for {product_name}. 

                    Write a report on the scalability and revenue streams 

                    for the product. Include at least 10 bullet points 

                    on key business areas. Give Business Plan, 

                    Goals and Timeline for the product launch. Current month is Jan 2024.""",

        expected_output="Report on business model evaluation and product launch plan.",

        agent=business_consultant,

    )



    # Create and Run the Crew

    product_crew = Crew(

        agents=[market_research_analyst, technology_expert, business_consultant],

        tasks=[task1, task2, task3],

        verbose=True,

        process=Process.sequential,

    )







    crew_result = product_crew.kickoff()

    return crew_result



#display the console processing on streamlit UI

class StreamToExpander:

    def __init__(self, expander):

        self.expander = expander

        self.buffer = []

        self.colors = ['red', 'green', 'blue', 'orange']  # Define a list of colors

        self.color_index = 0  # Initialize color index



    def write(self, data):

        # Filter out ANSI escape codes using a regular expression

        cleaned_data = re.sub(r'\x1B\[[0-9;]*[mK]', '', data)



        # Check if the data contains 'task' information

        task_match_object = re.search(r'\"task\"\s*:\s*\"(.*?)\"', cleaned_data, re.IGNORECASE)

        task_match_input = re.search(r'task\s*:\s*([^\n]*)', cleaned_data, re.IGNORECASE)

        task_value = None

        if task_match_object:

            task_value = task_match_object.group(1)

        elif task_match_input:

            task_value = task_match_input.group(1).strip()



        if task_value:

            st.toast(":robot_face: " + task_value)



        # Check if the text contains the specified phrase and apply color

        if "Entering new CrewAgentExecutor chain" in cleaned_data:

            # Apply different color and switch color index

            self.color_index = (self.color_index + 1) % len(self.colors)  # Increment color index and wrap around if necessary



            cleaned_data = cleaned_data.replace("Entering new CrewAgentExecutor chain", f":{self.colors[self.color_index]}[Entering new CrewAgentExecutor chain]")



        if "Market Research Analyst" in cleaned_data:

            # Apply different color 

            cleaned_data = cleaned_data.replace("Market Research Analyst", f":{self.colors[self.color_index]}[Market Research Analyst]")

        if "Business Development Consultant" in cleaned_data:

            cleaned_data = cleaned_data.replace("Business Development Consultant", f":{self.colors[self.color_index]}[Business Development Consultant]")

        if "Technology Expert" in cleaned_data:

            cleaned_data = cleaned_data.replace("Technology Expert", f":{self.colors[self.color_index]}[Technology Expert]")

        if "Finished chain." in cleaned_data:

            cleaned_data = cleaned_data.replace("Finished chain.", f":{self.colors[self.color_index]}[Finished chain.]")



        self.buffer.append(cleaned_data)

        if "\n" in data:

            self.expander.markdown(''.join(self.buffer), unsafe_allow_html=True)

            self.buffer = []



# Streamlit interface

def run_crewai_app():

    st.title("Watsonx AI Agent for product potentials")

    with st.expander("About the Team:"):

        st.subheader("Diagram")

        left_co, cent_co,last_co = st.columns(3)

        with cent_co:

            st.image("my_img.png")



        st.subheader("Market Research Analyst")

        st.text("""       

        Role = Market Research Analyst

        Goal = Analyze the market demand for {product_name} and suggest marketing strategies

        Backstory = Expert at understanding market demand, target audience, 

                    and competition for products like {product_name}. 

                    Skilled in developing marketing strategies 

                    to reach a wide audience.

        Task = Analyze the market demand for {product_name}. Current month is Jan 2024.

               Write a report on the ideal customer profile and marketing 

               strategies to reach the widest possible audience. 

               Include at least 10 bullet points addressing key marketing areas. """)



        st.subheader("Technology Expert")

        st.text("""       

        Role = Technology Expert

        Goal = Assess technological feasibilities and requirements for producing high-quality {product_name}

        Backstory = Visionary in current and emerging technological trends, 

                    especially in products like {product_name}. 

                    Identifies which technologies are best suited 

                    for different business models. 

        Task = Assess the technological aspects of manufacturing 

               high-quality {product_name}. Write a report detailing necessary 

               technologies and manufacturing approaches. 

               Include at least 10 bullet points on key technological areas.""")



        st.subheader("Business Development Consultant")

        st.text("""       

        Role = Business Development Consultant 

        Goal= Evaluate the business model for {product_name}

              focusing on scalability and revenue streams

        Backstory = Seasoned in shaping business strategies for products like {product_name}. 

                    Understands scalability and potential 

                    revenue streams to ensure long-term sustainability.

        Task = Summarize the market and technological reports 

               and evaluate the business model for {product_name}. 

               Write a report on the scalability and revenue streams 

               for the product. Include at least 10 bullet points 

               on key business areas. Give Business Plan, 

               Goals and Timeline for the product launch. Current month is Jan 2024. """)



    product_name = st.text_input("Enter a product name to analyze the market and business strategy.")



    if st.button("Run Analysis"):

        # Placeholder for stopwatch

        stopwatch_placeholder = st.empty()



        # Start the stopwatch

        start_time = time.time()

        with st.expander("Processing!"):

            sys.stdout = StreamToExpander(st)

            with st.spinner("Generating Results"):

                crew_result = create_crewai_setup(product_name)



        # Stop the stopwatch

        end_time = time.time()

        total_time = end_time - start_time

        stopwatch_placeholder.text(f"Total Time Elapsed: {total_time:.2f} seconds")



        st.header("Tasks:")

        st.table({"Tasks" : task_values})



        st.header("Results:")

        st.markdown(crew_result)



if __name__ == "__main__":

    run_crewai_app()

#crewAI==0.51.0 

#or

crewAI

langchain_community

langchain_ibm

ibm-watsonx-ai

duckduckgo-search

crewai-tools

streamlit

# Use an official Python runtime as a parent image

FROM python:3.11-slim



# Set the working directory in the container

WORKDIR /app



# Copy the current directory contents into the container at /app

COPY . /app

RUN apt-get update && apt-get install -y \

    build-essential \

    curl \

    software-properties-common \

    git \

    && rm -rf /var/lib/apt/lists/*



RUN git clone https://github.com/streamlit/streamlit-example.git



# Install any needed dependencies specified in requirements.txt

#RUN pip install --no-cache-dir -r requirements.txt

RUN pip install --upgrade pip setuptools wheel \

    && pip install -r requirements.txt \

    && rm -rf /root/.cache/pip



EXPOSE 8051



HEALTHCHECK CMD curl --fail http://localhost:8501/_stcore/health





# Run app.py when the container launches

ENTRYPOINT ["streamlit", "run", "main.py", "--server.port=8501", "--server.address=0.0.0.0"]

podman build -t agent .



podman run -p 8501:8501 agent