FAQ_RAG_Agent.py

# You may need to add your working directory to the Python path. To do so, uncomment the following lines of code
# import sys
# sys.path.append("/Path/to/directory/besser-agentic-framework") # Replace with your directory path

import json
import logging
import operator

from baf.core.agent import Agent
from baf.library.transition.events.base_events import *
from baf.nlp.llm.llm_huggingface import LLMHuggingFace
from baf.nlp.llm.llm_huggingface_api import LLMHuggingFaceAPI
from baf.nlp.llm.llm_openai_api import LLMOpenAI
from baf.nlp.llm.llm_replicate_api import LLMReplicate
from baf.core.session import Session
from baf.nlp.intent_classifier.intent_classifier_configuration import LLMIntentClassifierConfiguration, SimpleIntentClassifierConfiguration
from baf.nlp.speech2text.openai_speech2text import OpenAISpeech2Text
from baf.nlp.text2speech.openai_text2speech import OpenAIText2Speech
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from baf import nlp
from baf.nlp.rag.rag import RAG

# Configure the logging module
logging.basicConfig(level=logging.INFO, format='{levelname} - {asctime}: {message}', style='{')

def build_agent_configuration_prompt(agent_configuration: dict) -> str:
    base_prompt = """
You are a personalized AI assistant.

You will be provided with an object called `agent_configuration` as a JSON string.
This object defines how you should present yourself and formulate your responses.

It controls stylistic and presentation aspects such as:
- Response language
- Level of formality
- Language complexity
- Sentence length
- And more

You must strictly follow the preferences defined in `agent_configuration` for all responses.

Here is the current agent configuration (JSON):

{agent_configuration_json}
"""
    return base_prompt.format(agent_configuration_json=json.dumps(agent_configuration, indent=2))


def build_user_profile_prompt(user_profile: dict) -> str:
    profile_prompt = """
You have access to the current user's profile encoded as JSON.
Leverage these traits whenever the agent configuration instructs you to adapt content.

Here is the active user profile (JSON):

{user_profile_json}
"""
    return profile_prompt.format(user_profile_json=json.dumps(user_profile, indent=2))


# Create the bot
agent = Agent('FAQ_RAG_Agent', user_profiles_path='user_profiles.json', persist_sessions=True)
# Load bot properties stored in a dedicated file
agent.load_properties('config.yaml')

# Define the platform your chatbot will use

# Collect profile names from provided personalization mappings
profile_names = []
agent_configurations = {}
user_profiles = {}
context_prompts = {}
profile_prompts = {}
profile_names.append('Layperson')
agent_configurations['Layperson'] = json.loads(r'''{
  "adaptContentToUserProfile": true,
  "agentLanguage": "original",
  "agentPlatform": "streamlit",
  "agentStyle": "original",
  "avatar": null,
  "inputModalities": [
    "text",
    "speech"
  ],
  "intentRecognitionTechnology": "llm-based",
  "interfaceStyle": {
    "alignment": "left",
    "color": "var(--apollon-primary-contrast)",
    "contrast": "medium",
    "font": "sans",
    "lineSpacing": 1.5,
    "size": 16
  },
  "languageComplexity": "original",
  "llm": {
    "model": "gpt-5",
    "provider": "openai"
  },
  "outputModalities": [
    "text",
    "speech"
  ],
  "responseTiming": "instant",
  "sentenceLength": "original",
  "useAbbreviations": false,
  "userProfileName": "Layperson",
  "voiceStyle": {
    "gender": "male",
    "speed": 1
  }
}''')
user_profiles['Layperson'] = json.loads(r'''{
  "model": {
    "class": "User",
    "id": "user_1"
  },
  "name": "UserProfile"
}''')
context_prompts['Layperson'] = build_agent_configuration_prompt(agent_configurations['Layperson'])
profile_prompts['Layperson'] = build_user_profile_prompt(user_profiles['Layperson'])
profile_names.append('Lawyer')
agent_configurations['Lawyer'] = json.loads(r'''{
  "adaptContentToUserProfile": true,
  "agentLanguage": "original",
  "agentPlatform": "streamlit",
  "agentStyle": "original",
  "avatar": null,
  "inputModalities": [
    "text",
    "speech"
  ],
  "intentRecognitionTechnology": "llm-based",
  "interfaceStyle": {
    "alignment": "left",
    "color": "var(--apollon-primary-contrast)",
    "contrast": "medium",
    "font": "sans",
    "lineSpacing": 1.5,
    "size": 16
  },
  "languageComplexity": "original",
  "llm": {
    "model": "gpt-5",
    "provider": "openai"
  },
  "outputModalities": [
    "text",
    "speech"
  ],
  "responseTiming": "instant",
  "sentenceLength": "original",
  "useAbbreviations": false,
  "userProfileName": "Layperson",
  "voiceStyle": {
    "gender": "male",
    "speed": 1
  }
}''')
user_profiles['Lawyer'] = json.loads(r'''{
  "model": {
    "Competence": {
      "Skill": {
        "name": "Legal_Expertise",
        "score": "100"
      }
    },
    "Personal_Information": {
      "age": 25
    },
    "class": "User",
    "id": "Lawyer"
  },
  "name": "UserProfile"
}''')
context_prompts['Lawyer'] = build_agent_configuration_prompt(agent_configurations['Lawyer'])
profile_names.append('User 2')
agent_configurations['User_2'] = json.loads(r'''{
  "adaptContentToUserProfile": true,
  "agentLanguage": "original",
  "agentPlatform": "streamlit",
  "agentStyle": "original",
  "avatar": null,
  "inputModalities": [
    "text",
    "speech"
  ],
  "intentRecognitionTechnology": "llm-based",
  "interfaceStyle": {
    "alignment": "left",
    "color": "var(--apollon-primary-contrast)",
    "contrast": "medium",
    "font": "sans",
    "lineSpacing": 1.5,
    "size": 16
  },
  "languageComplexity": "original",
  "llm": {
    "model": "gpt-5",
    "provider": "openai"
  },
  "outputModalities": [
    "text",
    "speech"
  ],
  "responseTiming": "instant",
  "sentenceLength": "original",
  "useAbbreviations": false,
  "userProfileName": "Layperson",
  "voiceStyle": {
    "gender": "male",
    "speed": 1
  }
}''')
user_profiles['User_2'] = json.loads(r'''{
  "model": {
    "Accessibility": {
      "Disability": {
        "affects": "Mobility",
        "description": "can\u0027t use lower body",
        "name": "Paraplegic"
      }
    },
    "class": "User",
    "id": "user_1"
  },
  "name": "UserProfile"
}''')
context_prompts['User_2'] = build_agent_configuration_prompt(agent_configurations['User_2'])
profile_names.append('User Diagram')
agent_configurations['User_Diagram'] = json.loads(r'''{
  "adaptContentToUserProfile": true,
  "agentLanguage": "original",
  "agentPlatform": "streamlit",
  "agentStyle": "original",
  "avatar": null,
  "inputModalities": [
    "text",
    "speech"
  ],
  "intentRecognitionTechnology": "llm-based",
  "interfaceStyle": {
    "alignment": "left",
    "color": "var(--apollon-primary-contrast)",
    "contrast": "medium",
    "font": "sans",
    "lineSpacing": 1.5,
    "size": 16
  },
  "languageComplexity": "original",
  "llm": {
    "model": "gpt-5",
    "provider": "openai"
  },
  "outputModalities": [
    "text",
    "speech"
  ],
  "responseTiming": "instant",
  "sentenceLength": "original",
  "useAbbreviations": false,
  "userProfileName": "Layperson",
  "voiceStyle": {
    "gender": "male",
    "speed": 1
  }
}''')
user_profiles['User_Diagram'] = json.loads(r'''{
  "model": {
    "Personal_Information": {
      "age": 65
    },
    "class": "User",
    "id": "user_1"
  },
  "name": "UserProfile"
}''')
context_prompts['User_Diagram'] = build_agent_configuration_prompt(agent_configurations['User_Diagram'])
agent.set_agent_configurations(agent_configurations)




platform = agent.use_websocket_platform(use_ui=True, authenticate_users=True)
# LLM instantiation based on config['llm']
reply_llm = LLMOpenAI(
    agent=agent,
    name='gpt-5',
    parameters={}
)

stt = OpenAISpeech2Text(agent=agent, model_name="whisper-1")
tts = OpenAIText2Speech(agent=agent, model_name="gpt-4o-mini-tts")

rag_llm = LLMOpenAI(
    agent=agent,
    name='gpt-4o-mini',
    parameters={},
    num_previous_messages=10
)

##############################
# RAG CONFIGURATIONS
##############################

esc_rag_vector_store = Chroma(
    embedding_function=OpenAIEmbeddings(openai_api_key=agent.get_property(nlp.OPENAI_API_KEY)),
    persist_directory='vector_store/esc_rag'
)
esc_rag_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
esc_rag_rag = RAG(
    agent=agent,
    vector_store=esc_rag_vector_store,
    splitter=esc_rag_splitter,
    llm_name='gpt-4o-mini',
    k=4,
    num_previous_messages=0
)
esc_rag_rag.load_pdfs('./esc_rag')



ic_config = LLMIntentClassifierConfiguration(
    llm_name='gpt-5',
    parameters={},
    use_intent_descriptions=True,
    use_training_sentences=True,
    use_entity_descriptions=False,
    use_entity_synonyms=False
)

agent.set_default_ic_config(ic_config)



##############################
# INTENTS
##############################

##############################
# PERSONALIZED INTENTS
##############################
# Intents for profile Layperson

# Intents for profile Lawyer

# Intents for profile User_2

# Intents for profile User_Diagram



##############################
# CUSTOM CONDITIONS
##############################






##############################
# STATES
##############################

# Dummy entry state to fan out to profile-specific initial states
router_initial_state = agent.new_state('router_initial_state', initial=True)

Greeting = agent.new_state('Greeting')
Idle = agent.new_state('Idle')
Response = agent.new_state('Response')

##############################
# PROFILE STATES
##############################
# States for profile Layperson
Greeting_Layperson = agent.new_state('Greeting_Layperson')
Idle_Layperson = agent.new_state('Idle_Layperson')
Response_Layperson = agent.new_state('Response_Layperson')
# States for profile Lawyer
Greeting_Lawyer = agent.new_state('Greeting_Lawyer')
Idle_Lawyer = agent.new_state('Idle_Lawyer')
Response_Lawyer = agent.new_state('Response_Lawyer')
# States for profile User_2
Greeting_User_2 = agent.new_state('Greeting_User_2')
Idle_User_2 = agent.new_state('Idle_User_2')
Response_User_2 = agent.new_state('Response_User_2')
# States for profile User_Diagram
Greeting_User_Diagram = agent.new_state('Greeting_User_Diagram')
Idle_User_Diagram = agent.new_state('Idle_User_Diagram')
Response_User_Diagram = agent.new_state('Response_User_Diagram')

##############################
# ROUTER TRANSITIONS TO PROFILE INITIAL STATES
##############################
router_initial_state.when_variable_matches_operation('user_profile', operator.eq, 'Layperson').go_to(Greeting_Layperson)
router_initial_state.when_variable_matches_operation('user_profile', operator.eq, 'Lawyer').go_to(Greeting_Lawyer)
router_initial_state.when_variable_matches_operation('user_profile', operator.eq, 'User 2').go_to(Greeting_User_2)
router_initial_state.when_variable_matches_operation('user_profile', operator.eq, 'User Diagram').go_to(Greeting_User_Diagram)


# Greeting
def Greeting_body(session: Session):
    speech_messages = []
    reply_text = 'Hi, I am your leagl assistant, I can answer questions related to the ESC.'
    session.reply(reply_text)
    speech_messages.append(reply_text)
    if speech_messages:
        platform.reply_speech(session, ' '.join(speech_messages))
Greeting.set_body(Greeting_body)
Greeting.go_to(Idle)
# Idle
def Idle_body(session: Session):
    speech_messages = []
    reply_text = 'What would you like to know about the ESC?'
    session.reply(reply_text)
    speech_messages.append(reply_text)
    if speech_messages:
        platform.reply_speech(session, ' '.join(speech_messages))
Idle.set_body(Idle_body)
Idle.when_no_intent_matched().go_to(Response)
Idle.when_variable_matches_operation('user_profile', operator.eq, 'Layperson').go_to(router_initial_state)
Idle.when_variable_matches_operation('user_profile', operator.eq, 'Lawyer').go_to(router_initial_state)
Idle.when_variable_matches_operation('user_profile', operator.eq, 'User 2').go_to(router_initial_state)
Idle.when_variable_matches_operation('user_profile', operator.eq, 'User Diagram').go_to(router_initial_state)
# Response
def Response_body(session: Session):
    rag_message = session.run_rag(session.event.message)
    platform.reply_rag(session, rag_message)
Response.set_body(Response_body)
Response.go_to(Idle)



##############################
# PROFILE STATE BODIES & TRANSITIONS
##############################
# Greeting (Layperson)
def Greeting_body_Layperson(session: Session):
    reply_llm.add_user_context(
        session=session,
        context=context_prompts.get('Layperson'),
        context_name='agent_configuration_Layperson'
    )
    reply_llm.add_user_context(
        session=session,
        context=profile_prompts.get('Layperson'),
        context_name='user_profile_Layperson'
    )
    speech_messages = []
    reply_text = 'Hi, I am your leagl assistant, I can answer questions related to the ESC.'
    session.reply(reply_text)
    speech_messages.append(reply_text)
    if speech_messages:
        platform.reply_speech(session, ' '.join(speech_messages))
Greeting_Layperson.set_body(Greeting_body_Layperson)
Greeting_Layperson.go_to(Idle_Layperson)
# Idle (Layperson)
def Idle_body_Layperson(session: Session):
    speech_messages = []
    reply_text = 'What would you like to know about the ESC?'
    session.reply(reply_text)
    speech_messages.append(reply_text)
    if speech_messages:
        platform.reply_speech(session, ' '.join(speech_messages))
Idle_Layperson.set_body(Idle_body_Layperson)
Idle_Layperson.when_no_intent_matched().go_to(Response_Layperson)
Idle_Layperson.when_variable_matches_operation('user_profile', operator.eq, 'Lawyer').go_to(router_initial_state)
Idle_Layperson.when_variable_matches_operation('user_profile', operator.eq, 'User 2').go_to(router_initial_state)
Idle_Layperson.when_variable_matches_operation('user_profile', operator.eq, 'User Diagram').go_to(router_initial_state)
# Response (Layperson)
def Response_body_Layperson(session: Session):
    rag_message = session.run_rag(session.event.message)
    platform.reply_rag(session, rag_message)
Response_Layperson.set_body(Response_body_Layperson)
Response_Layperson.go_to(Idle_Layperson)

# Greeting (Lawyer)
def Greeting_body_Lawyer(session: Session):
    reply_llm.add_user_context(
        session=session,
        context=context_prompts.get('Lawyer'),
        context_name='agent_configuration_Lawyer'
    )
    reply_llm.add_user_context(
        session=session,
        context=profile_prompts.get('Lawyer'),
        context_name='user_profile_Lawyer'
    )
    speech_messages = []
    reply_text = 'Hi, I am your leagl assistant, I can answer questions related to the ESC.'
    session.reply(reply_text)
    speech_messages.append(reply_text)
    if speech_messages:
        platform.reply_speech(session, ' '.join(speech_messages))
Greeting_Lawyer.set_body(Greeting_body_Lawyer)
Greeting_Lawyer.go_to(Idle_Lawyer)
# Idle (Lawyer)
def Idle_body_Lawyer(session: Session):
    speech_messages = []
    reply_text = 'What would you like to know about the ESC?'
    session.reply(reply_text)
    speech_messages.append(reply_text)
    if speech_messages:
        platform.reply_speech(session, ' '.join(speech_messages))
Idle_Lawyer.set_body(Idle_body_Lawyer)
Idle_Lawyer.when_no_intent_matched().go_to(Response_Lawyer)
Idle_Lawyer.when_variable_matches_operation('user_profile', operator.eq, 'Layperson').go_to(router_initial_state)
Idle_Lawyer.when_variable_matches_operation('user_profile', operator.eq, 'User 2').go_to(router_initial_state)
Idle_Lawyer.when_variable_matches_operation('user_profile', operator.eq, 'User Diagram').go_to(router_initial_state)
# Response (Lawyer)
def Response_body_Lawyer(session: Session):
    rag_message = session.run_rag(session.event.message)
    platform.reply_rag(session, rag_message)
Response_Lawyer.set_body(Response_body_Lawyer)
Response_Lawyer.go_to(Idle_Lawyer)

# Greeting (User_2)
def Greeting_body_User_2(session: Session):
    reply_llm.add_user_context(
        session=session,
        context=context_prompts.get('User_2'),
        context_name='agent_configuration_User_2'
    )
    reply_llm.add_user_context(
        session=session,
        context=profile_prompts.get('User_2'),
        context_name='user_profile_User_2'
    )
    speech_messages = []
    reply_text = 'Hi, I am your leagl assistant, I can answer questions related to the ESC.'
    session.reply(reply_text)
    speech_messages.append(reply_text)
    if speech_messages:
        platform.reply_speech(session, ' '.join(speech_messages))
Greeting_User_2.set_body(Greeting_body_User_2)
Greeting_User_2.go_to(Idle_User_2)
# Idle (User_2)
def Idle_body_User_2(session: Session):
    speech_messages = []
    reply_text = 'What would you like to know about the ESC?'
    session.reply(reply_text)
    speech_messages.append(reply_text)
    if speech_messages:
        platform.reply_speech(session, ' '.join(speech_messages))
Idle_User_2.set_body(Idle_body_User_2)
Idle_User_2.when_no_intent_matched().go_to(Response_User_2)
Idle_User_2.when_variable_matches_operation('user_profile', operator.eq, 'Layperson').go_to(router_initial_state)
Idle_User_2.when_variable_matches_operation('user_profile', operator.eq, 'Lawyer').go_to(router_initial_state)
Idle_User_2.when_variable_matches_operation('user_profile', operator.eq, 'User Diagram').go_to(router_initial_state)
# Response (User_2)
def Response_body_User_2(session: Session):
    rag_message = session.run_rag(session.event.message)
    platform.reply_rag(session, rag_message)
Response_User_2.set_body(Response_body_User_2)
Response_User_2.go_to(Idle_User_2)

# Greeting (User_Diagram)
def Greeting_body_User_Diagram(session: Session):
    reply_llm.add_user_context(
        session=session,
        context=context_prompts.get('User_Diagram'),
        context_name='agent_configuration_User_Diagram'
    )
    reply_llm.add_user_context(
        session=session,
        context=profile_prompts.get('User_Diagram'),
        context_name='user_profile_User_Diagram'
    )
    speech_messages = []
    reply_text = 'Hi, I am your leagl assistant, I can answer questions related to the ESC.'
    session.reply(reply_text)
    speech_messages.append(reply_text)
    if speech_messages:
        platform.reply_speech(session, ' '.join(speech_messages))
Greeting_User_Diagram.set_body(Greeting_body_User_Diagram)
Greeting_User_Diagram.go_to(Idle_User_Diagram)
# Idle (User_Diagram)
def Idle_body_User_Diagram(session: Session):
    speech_messages = []
    reply_text = 'What would you like to know about the ESC?'
    session.reply(reply_text)
    speech_messages.append(reply_text)
    if speech_messages:
        platform.reply_speech(session, ' '.join(speech_messages))
Idle_User_Diagram.set_body(Idle_body_User_Diagram)
Idle_User_Diagram.when_no_intent_matched().go_to(Response_User_Diagram)
Idle_User_Diagram.when_variable_matches_operation('user_profile', operator.eq, 'Layperson').go_to(router_initial_state)
Idle_User_Diagram.when_variable_matches_operation('user_profile', operator.eq, 'Lawyer').go_to(router_initial_state)
Idle_User_Diagram.when_variable_matches_operation('user_profile', operator.eq, 'User 2').go_to(router_initial_state)
# Response (User_Diagram)
def Response_body_User_Diagram(session: Session):
    rag_message = session.run_rag(session.event.message)
    platform.reply_rag(session, rag_message)
Response_User_Diagram.set_body(Response_body_User_Diagram)
Response_User_Diagram.go_to(Idle_User_Diagram)


# RUN APPLICATION

if __name__ == '__main__':
    agent.run()