My personal “majordhomme” (butler) AI agent with CrewAI, Granite, DeepSeek and more…

An agent based on CrewAI with different skills using Granite and DeepSeek so far (more to come)

Introduction

This article describes a project I work on for my own learning and hands-on work on making agents and implementing different skills using divers tools. It is far from a professional class job, but I learn a lot while making my hands dirty and writing code on my own.

The main idea is to build an agent that during a chat, detects specific keywords, and based on the keywords uses set of tools implemented inside the application to propose and suggest additional services.

For example, if the user asks “Where is the capital of Russia”, the chat will obviously answer “Moscow” and then would suggest to the user if they want to know the weather in Moscow!

This is what I implemented so far and plan to expand to more set of tools to come (such as fetching airplane travel fees/timetables and maps directions and more…).

To do this I use the CrewAI framework. Services implemented so far are;

• DeepSeek for general chat purpose (because I never tried it before),

• granite-3.2–2b-instruct for using some tools (such as calling the weather API),

• granite-vision-3.2–2b for describing images uploaded to the app.

• I used Streamlit to build the GUI as by default it creates nice user interfaces.
• The LLMs are accessed through Hugging Face.

So meet the ‘team’!

And the available and soon to be tools/options…

Implementation logique

As I mentioned in the introduction, the LLMs used are accessed through the Hugging Face platform, and so far I use 3 different LLM.

# Initialize Hugging Face Inference
client = InferenceClient(
    model="ibm-granite/granite-3.2-2b-instruct", token=HUGGINGFACE_API_TOKEN
)

vision_client = InferenceClient(
    model="ibm-granite/granite-vision-3.2-2b", token=HUGGINGFACE_API_TOKEN
)

# Initialize DeepSeek 
def initialize_deepseek_llm():
    llm = HuggingFaceEndpoint(
        repo_id="deepseek-ai/DeepSeek-R1-Distill-Qwen-32B",
        huggingfacehub_api_token=HUGGINGFACE_API_TOKEN,
        temperature=0.4,  
        max_new_tokens=512,
        #max_length=512,  
        stop_sequences=["\nQuestion:", "\nUser:"],
    )
    return llm

Having implemented and tested each crew member individually, I wanted to establish a logic of collaboration between the members. So the idea became as the following schema describes it.

I had to try different types of attributes for DeepSeek in order to obtain what I wanted for my sample agent.

I have not bulletproofed the application (yet).

As shown below in the code which follows, I spent a lot of energy focusing on the use case of “city” name detection in the answer of the question “Where is the capital of XXX”, and did various tests on this use-case. The next step of my application would be to achieve a global solution. However so far the implementation is demonstrated in the code below.

For instance, I had to refine the prompt in order to have accurate answers for my ‘city’ use case.

def initialize_deepseek_prompt():
    template = """
    Question: What is the capital of France?
    Answer: Paris
    Question: What is the capital of Japan?
    Answer: Tokyo
    Question: What is the currency of Brazil?
    Answer: Brazilian Real
    Question: What is the tallest building in the world?
    Answer: Burj Khalifa
    Question: Provide only the name of the capital city of {question}
    Answer: """  
    prompt = PromptTemplate(template=template, input_variables=["question"])
    return prompt

First things first, I use a virtual environment.

python3.12 -m venv crewai_env
source crewai_env/bin/activate

Then goes all the requirements for the application.

streamlit
crewai
python-dotenv
requests
huggingface_hub
Pillow
spacy
langchain_community
langchain
langchain_huggingface

And the credentials and other parameters needed for Hugging Face and other tools (one might need) in an ".env" file.

OPENWEATHERMAP_API_KEY="YOUR_KEY"
HUGGINGFACE_API_TOKEN="YOUR_TOKEN"
OPENAI_API_KEY="dummy_key"

I have the habit now to isolate and test different functionalities in tiny code/apps to be able to prevent the upcoming probable problems. For instance here I test my Hugging Face connection.

from langchain.llms import HuggingFaceHub

try:
    llm = HuggingFaceHub(repo_id="google/flan-t5-small")
    print("HuggingFaceHub imported and initialized successfully.")
except NameError:
    print("NameError: HuggingFaceHub not found.")
except Exception as e:
    print(f"An error occurred: {e}")

A simple test with “python app.py” lets me know if my token is correctly ‘sourced’ and if I can connect to HF platform.

Now, here is the looooong listing of the application.

import streamlit as st
import os
from crewai import Agent, Task, Crew, Process
from dotenv import load_dotenv
import requests
import json
import io
import base64
import time
import sys
import re
import spacy

nlp = spacy.load("en_core_web_sm")

from datetime import datetime
from huggingface_hub import InferenceClient
from PIL import Image
from langchain_huggingface import HuggingFaceEndpoint
from langchain.prompts import PromptTemplate
from langchain_core.runnables import RunnableSequence
from langchain_community.llms import HuggingFaceHub # Corrected import
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate


# Initializations
if "stop_app" not in st.session_state:
    st.session_state.stop_app = False

if st.button("Stop App"):
    st.session_state.stop_app = True

if st.session_state.stop_app:
    st.write("Stopping the app...")
    sys.exit()


if "nlp" not in st.session_state:
    st.session_state.nlp = spacy.load("en_core_web_sm")


# Load environment variables
load_dotenv()
OPENWEATHERMAP_API_KEY = os.getenv("OPENWEATHERMAP_API_KEY")
HUGGINGFACE_API_TOKEN = os.getenv("HUGGINGFACE_API_TOKEN")
GLOBAL_CITY = ""

#print(f"OpenWeatherMap API Key: {OPENWEATHERMAP_API_KEY}")
#print(f"Hugging Face API Token: {HUGGINGFACE_API_TOKEN}")


client = InferenceClient(
    model="ibm-granite/granite-3.2-2b-instruct", token=HUGGINGFACE_API_TOKEN

)

vision_client = InferenceClient(
    model="ibm-granite/granite-vision-3.2-2b", token=HUGGINGFACE_API_TOKEN

)


def initialize_deepseek_llm():
    llm = HuggingFaceEndpoint(
        repo_id="deepseek-ai/DeepSeek-R1-Distill-Qwen-32B",
        huggingfacehub_api_token=HUGGINGFACE_API_TOKEN,
        temperature=0.4,  
        max_new_tokens=512,
        #max_length=512,  
        stop_sequences=["\nQuestion:", "\nUser:"],
    )
    return llm

def initialize_deepseek_prompt():
    template = """
    Question: What is the capital of France?
    Answer: Paris
    Question: What is the capital of Japan?
    Answer: Tokyo
    Question: What is the currency of Brazil?
    Answer: Brazilian Real
    Question: What is the tallest building in the world?
    Answer: Burj Khalifa
    Question: Provide only the name of the capital city of {question}
    Answer: """  # Refined prompt
    prompt = PromptTemplate(template=template, input_variables=["question"])
    return prompt

def initialize_deepseek_chain(llm, prompt):
    llm_chain = RunnableSequence(prompt, llm)
    return llm_chain

if "deepseek_llm" not in st.session_state:
    st.session_state.deepseek_llm = initialize_deepseek_llm()
if "deepseek_prompt" not in st.session_state:
    st.session_state.deepseek_prompt = initialize_deepseek_prompt()
if "deepseek_chain" not in st.session_state:
    st.session_state.deepseek_chain = initialize_deepseek_chain(st.session_state.deepseek_llm, st.session_state.deepseek_prompt)


# --- Weather Forecast ---

def get_weather_forecast(city):

    url = f"http://api.openweathermap.org/data/2.5/forecast?q={city}&appid={OPENWEATHERMAP_API_KEY}&units=metric"
    try:
        response = requests.get(url)
        time.sleep(1) 
        response.raise_for_status()
        data = response.json()
        return data
    except requests.exceptions.RequestException as e:
        return {"error": f"Error fetching weather data: {e}"}

def format_forecast(forecast_data):
    """Formats the raw forecast data into a human-readable string."""
    if "error" in forecast_data:
        return forecast_data["error"]

    forecast_list = forecast_data["list"]
    daily_forecasts = {}

    for forecast in forecast_list:
        date_time = datetime.fromisoformat(forecast["dt_txt"])
        date = date_time.date()

        if date not in daily_forecasts:
            daily_forecasts[date] = {
                "temperatures": [],
                "descriptions": [],
                "humidity": [],
                "wind_speed": [],
            }

        daily_forecasts[date]["temperatures"].append(forecast["main"]["temp"])
        daily_forecasts[date]["descriptions"].append(forecast["weather"][0]["description"])
        daily_forecasts[date]["humidity"].append(forecast["main"]["humidity"])
        daily_forecasts[date]["wind_speed"].append(forecast["wind"]["speed"])

    formatted_forecast = ""
    for date, data in daily_forecasts.items():
        avg_temp = sum(data["temperatures"]) / len(data["temperatures"])
        description = ", ".join(set(data["descriptions"]))
        avg_humidity = sum(data["humidity"]) / len(data["humidity"])
        avg_wind_speed = sum(data["wind_speed"]) / len(data["wind_speed"])

        formatted_forecast += f"**{date.strftime('%Y-%m-%d')}**:\n"
        formatted_forecast += f"- Average Temperature: {avg_temp:.1f}°C\n"
        formatted_forecast += f"- Description: {description}\n"
        formatted_forecast += f"- Average Humidity: {avg_humidity:.1f}%\n"
        formatted_forecast += f"- Average Wind Speed: {avg_wind_speed:.1f} m/s\n\n"

    return formatted_forecast

def generate_response(prompt, max_retries=3, retry_delay=5):
    retries = 0
    while retries < max_retries:
        try:

            response = client.text_generation(prompt, max_new_tokens=250)
            return response
        except requests.exceptions.RequestException as e:
            if "503 Server Error" in str(e):
                retries += 1
                print(f"Retry {retries}/{max_retries}. Error: {e}")
                time.sleep(retry_delay)
            else:
                return f"Error generating response: {e}"
        except Exception as e:
            return "Hugging Face is temporarily unavailable. Please try again later."
    return "Maximum retries exceeded. Hugging Face is not responding. Please try again later."


weather_fetcher = Agent( 
    role="Weather Data Fetcher",
    goal="Fetch weather forecast data from OpenWeatherMap API",
    backstory="You are an expert in retrieving weather data from online APIs.",
    allow_delegation=False,
    verbose=True,
    llm=None,
    tools=[],
)    


weather_formatter = Agent(
    role="Weather Forecast Formatter",
    goal="Format raw weather data into a readable forecast",
    backstory="You are skilled at presenting complex weather data in a clear and concise way.",
    allow_delegation=False,
    verbose=True,
    llm=None,
    tools=[],
)


fetch_weather_task = Task(
    description="Fetch the 7-day weather forecast for the specified city.",
    agent=weather_fetcher,
    process=Process.sequential,
    expected_output="JSON Weather Data",
)

format_weather_task = Task(
    description="Format the raw weather data into a detailed and easy-to-understand forecast.",
    agent=weather_formatter,
    process=Process.sequential,
    expected_output="Formatted Weather Forecast",
)


class WeatherCrew:
    def __init__(self):
        self.crew = Crew(
            agents=[weather_fetcher, weather_formatter],
            tasks=[fetch_weather_task, format_weather_task],
            verbose=True,
            process=Process.sequential,
        )

    def get_forecast(self, city):

        forecast_data = get_weather_forecast(city)
        if "error" in forecast_data:
            return forecast_data["error"]

        formatted_data = format_forecast(forecast_data)
        prompt = f"Here is weather data: {formatted_data}. Please provide a summary."

        result = generate_response(prompt)
        return result

# --- Image Description ---

def describe_image(image_bytes):

    try:

        response = vision_client.image_to_text(image_bytes)
        return response
    except Exception as e:
        return f"Error describing image: {e}"

# --- Streamlit UI ---

def weather_app():
    st.title("Weather Forecast App")
    st.image("./icons/weather-news.png", width=100)
    city = st.text_input("Enter city name:")

    if st.button("Get Weather Forecast"):
        if city:
            with st.spinner("Fetching forecast..."):
                weather_crew = WeatherCrew()
                forecast = weather_crew.get_forecast(city)
                if "Hugging Face" in forecast: 
                    st.error(forecast) 
                else:
                    st.write(forecast)
        else:
            st.warning("Please enter a city name.")

def image_app():
    st.header("Image Description")
    st.image("./icons/camera.png", width=100)
    uploaded_file = st.file_uploader("Upload an image", type=["jpg", "jpeg", "png"])

    if uploaded_file is not None:
        image_bytes = uploaded_file.getvalue()
        st.image(image_bytes, caption="Uploaded Image.", use_container_width=True) # change use_column_width to use_container_width

        if st.button("Describe Image"):
            with st.spinner("Describing image..."):
                description = describe_image(image_bytes)
                if "Hugging Face" in description:
                    st.error(description)
                else:
                    st.write(description)

def chat_app():
    st.header("General Chat with DeepSeek")
    st.image("./icons/deepseek-logo-03.png", width=100)
    user_input = st.text_input("Enter your query: ")


    if "city_detected" not in st.session_state:
        st.session_state.city_detected = False
    if "last_mentioned_city" not in st.session_state:
        st.session_state.last_mentioned_city = None
    if "weather_prompt" not in st.session_state:
        st.session_state.weather_prompt = None
    if "weather_query" not in st.session_state:
        st.session_state.weather_query = False  

    if user_input:
        if "weather" in user_input.lower():

            if "last_mentioned_city" in st.session_state and st.session_state.last_mentioned_city:
                city = st.session_state.last_mentioned_city
                st.write(f"Using last_mentioned_city: {city}")  # Debug
            elif "that city" in user_input.lower() and "last_city" in st.session_state:
                city = st.session_state.last_city
                st.write(f"Using last_city: {city}")  # Debug
            else:
                match = re.search(r"weather\s+in\s+(.+?) (?:\s*\?)?$", user_input, re.IGNORECASE)
                if match:
                    city = match.group(1).strip()
                else:
                    city = None
            if city:
                with st.spinner("Fetching weather forecast..."):
                    weather_crew = WeatherCrew()
                    forecast = weather_crew.get_forecast(city)
                    if "Hugging Face" in forecast:
                        st.error(forecast)
                    else:
                        st.write(forecast)
            else:
                st.warning("Please specify a city after 'weather in'.")
            return

        with st.spinner("Generating response..."):
            response = st.session_state.deepseek_chain.invoke({"question": user_input})
        st.write(f"LLM Response: {response}")
        response = re.sub(r"<think>.*?</think>", "", response, flags=re.DOTALL)
        st.text_area("DeepSeek:", value=response, height=200)

        # Extract city name from DeepSeek response
        city = extract_city(response)
        st.write(f"Extracted City: {city}")  # Debug: 

        update_city_state(city)

        display_weather_option(city)  # Display weather option

        if st.session_state.weather_query and st.session_state.get("weather_prompt"):
            # Use the last mentioned city for the weather query
            city_to_use = st.session_state.last_mentioned_city
            st.write(f"Using city for weather: {city_to_use}") #Debug
            with st.spinner(f"Fetching weather for {city_to_use}..."):
                weather_crew = WeatherCrew()
                forecast = weather_crew.get_forecast(city_to_use)
                if "Hugging Face" in forecast:
                    st.error(forecast)
                else:
                    st.write(forecast)
            st.session_state.weather_query = False
            st.session_state.weather_prompt = None


if "switch_to_weather_app" in st.session_state and st.session_state.switch_to_weather_app:
    weather_app()
    st.session_state.switch_to_weather_app = False


if "switch_to_weather_app" in st.session_state and st.session_state.switch_to_weather_app:
    weather_app()
    st.session_state.switch_to_weather_app = False

def extract_city_ner(response):
    doc = nlp(response)
    for ent in doc.ents:
        if ent.label_ == "GPE":  # GPE = Geopolitical Entity 
            return ent.text
    return None


def extract_city(response):
    """
    Extracts a city name from the LLM's response
    """
    nlp = st.session_state.nlp  


    cities_found = re.findall(r"([A-Z][a-z]+(?: [A-Z][a-z]+)?)", response)
    if cities_found:
        answer_text = cities_found[-1]  # Get the last one
    else:
        answer_text = response  # Fallback


    doc = nlp(answer_text)
    for ent in doc.ents:
        if ent.label_ == "GPE":
            return ent.text
    return None

def update_city_state(city):


    if "city_detected" not in st.session_state:
        st.session_state.city_detected = False
    if "last_mentioned_city" not in st.session_state:
        st.session_state.last_mentioned_city = None


    if city:
        st.session_state.last_mentioned_city = city
        st.write(f"Updated last_mentioned_city: {st.session_state.last_mentioned_city}")
    else:
        st.write(f"last_mentioned_city not updated, current value: {st.session_state.last_mentioned_city}")

def display_weather_option(city):

    if st.session_state.get("weather_prompt") != city:
        st.write(f"Did you want to know the weather in {city}?")
        st.session_state.city_detected = True

    if st.session_state.city_detected:
        st.session_state.weather_query = True
        st.session_state.city_detected = False
        st.session_state.weather_prompt = city  
    else:
        st.session_state.city_detected = True

    if st.session_state.weather_query and st.session_state.weather_prompt:  # Use weather_prompt directly

        city_to_use = st.session_state.last_mentioned_city
        with st.spinner (f"Fetching weather for {city_to_use}..."):
            weather_crew = WeatherCrew()
            forecast = weather_crew.get_forecast(city_to_use)
            if "Hugging Face" in forecast:
                st.error(forecast)
            else:
                st.write(forecast)
        st.session_state.weather_query = False
        st.session_state.weather_prompt = None      


def hotel_app():  
    st.header("--- Hotel Booking Agent to come --")  
    st.image("./icons/hotel.png", width=100)

def airplane_app():  
    st.header("--- Airplane Booking Helper agent --")    
    st.image("./icons/avion.png", width=100)  

def dummy_app():  
    st.header("--- application to come --")                

# --- Main App ---
st.markdown(
    '<p style="font-size: 30px;">Set of Agents to manage your day-to-day tasks, 24/7!<br>'
    'They will work together very shortly ;p)</p>',
    unsafe_allow_html=True,
)
st.sidebar.image("./icons/robot.png", width=80) 
st.sidebar.image("./icons/1_GlpCVWSHwX7U7wwBaUnZFA.webp", width=80) 
st.sidebar.image("./icons/1_nIobR8OHlk0UqHIRQtnsww.webp", width=80) 
with st.expander("About the Team:"):
        st.subheader("Diagram")
        left_co, cent_co,last_co = st.columns(3)
        with cent_co:
            st.image("./icons/robot.png", width=80)

        st.subheader("Le majordome ")
        st.text("""       
        Role = Answer general questions
        Goal = Answer general questions
        Backstory = He knows almost everything.
        Task = Answer your queries, whatever they are. """)

        st.subheader("Weather forecaster")
        st.text("""       
        Role = Weather forecaster
        Goal = Fetch the weather of the place you ask for!
        Backstory = Uses IBM Granite to fetch the weather.
        Task = Search the weather of every possible place on earth for you.""")

        st.subheader("Image analyzer")
        st.text("""       
        Role = Analyze images for you 
        Goal= Gives you explanations on images you upload!
        Backstory = Uses IBM Granite to describe an image.
        Task = Summarize and explain an image. """)

        st.subheader("--- TBN ---")
        st.text("""       
        Role = TBN 
        Goal= TBN
        Backstory = TBN.
        Task = TBN. """)

app_mode = st.sidebar.selectbox("Select App Mode", 
                                ["General Chat", 
                                 "Weather Forecast", 
                                 "Image Description", 
                                 "Hotel Reservation", 
                                 "Airline Travel Search"]
                                )

if app_mode == "Weather Forecast":
    weather_app()
elif app_mode == "Image Description":
    image_app()
elif app_mode == "General Chat":
    chat_app()    
elif app_mode == "Hotel Reservation":
    hotel_app()
elif app_mode == "Airline Travel Search":
    airplane_app()    

At the end, I was able to make the application work in the way I wanted as shown in the following screen captures.

And the weather forecast for the city.

Thanks for reading, appreciate feedbacks 🙏

Conclusion

This sample application implements a specific pre-determined keyword detection in a chat, and implements an intelligent agent which would provide in depth information regarding the detected keyword.