crew-ai-codecollection/chat-panel.py

# start with the following comando: panel serve chat-panel.py

import os
import platform


# Environment Variablen importieren
from dotenv import load_dotenv
load_dotenv()


from crewai import Crew, Process, Agent, Task
from langchain_openai import ChatOpenAI

from langchain_core.callbacks import BaseCallbackHandler
from typing import TYPE_CHECKING, Any, Dict, Optional

from crewai_tools import Tool



# pip install panel
import panel as pn 
pn.extension(design="material")


import threading

from crewai.agents import CrewAgentExecutor
import time 

def user_input_callback(content):
    
    global human_chat_user_input

    #prompt = self._i18n.slice("getting_input").format(final_answer=final_answer)
    print(content);

    chat_interface.send("Bitte prüfen Sie, ob der Plan ihre Anforderungen erfüllt", user="User", respond=False)

    while human_chat_user_input == None:
        time.sleep(1)  

    human_comments = human_chat_user_input
    human_chat_user_input = None

    return human_comments


def custom_ask_human_input(self, final_answer: dict) -> str:
      
      global human_chat_user_input

      #prompt = self._i18n.slice("getting_input").format(final_answer=final_answer)

      chat_interface.send(final_answer, user="User", respond=False)

      while human_chat_user_input == None:
          time.sleep(1)  
          print("Waiting for human input...")

      human_comments = human_chat_user_input
      human_chat_user_input = None

      return human_comments


#CrewAgentExecutor._ask_human_input = custom_ask_human_input


human_chat_user_input = None
initiate_chat_task_created = False

def initiate_chat(message):

    global initiate_chat_task_created
    # Indicate that the task has been created
    initiate_chat_task_created = True

    message=message+' auf deutsch.'

    StartCrew(message)



def callback(contents: str, user: str, instance: pn.chat.ChatInterface):

    print(f"User: {user} - Message: {contents}")
    
    global initiate_chat_task_created
    global human_chat_user_input

    if not initiate_chat_task_created:
        thread = threading.Thread(target=initiate_chat, args=(contents,))
        thread.start()

    else:
        human_chat_user_input = contents

avators = {"theologe":"https://cdn-icons-png.flaticon.com/512/320/320336.png",
           "religionswissenschaftler":"https://cdn-icons-png.freepik.com/512/9408/9408201.png",
           "moderator":"https://cdn-icons-png.freepik.com/512/9408/9408201.png"}

class MyCustomHandler(BaseCallbackHandler):

    def __init__(self, agent_name: str) -> None:
        self.agent_name = agent_name

    def on_chain_start(
        self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
    ) -> None:
        """We start a chain."""

        chat_interface.send(inputs['input'], user="assistent", respond=False)

    def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
        """Print out that we finished a chain."""

        print(f"Finished chain with output: {outputs}")
    
        chat_interface.send(outputs['output'], user=self.agent_name, avatar=avators[self.agent_name], respond=False)




llm = ChatOpenAI(model="gpt-3.5-turbo-0125")
#llm = ChatOpenAI(model="gpt-4")

theologe = Agent(
    name="Theologe",
    role='Theologische Inhalte entwickeln',
    backstory='''Langjähriger Theologieprofessor mit Spezialisierung auf historisch-kritische Bibelinterpretation''',
    goal="Verständnis für biblische Texte und christliche Lehren fördern.",
    llm=llm,
    allow_delegation=False,
    verbose=True,
    callbacks=[MyCustomHandler("theologe")],
)
religionswissenschaftler = Agent(
    name="Religionswissenschaftler",
    role='Interreligiösen Dialog fördern',
    backstory='''Religionswissenschaftler mit internationaler Erfahrung im Dialog zwischen Weltreligionen.''',
    goal="Förderung von ethischer Argumentation und philosophischem Denken",
    llm=llm,
    callbacks=[MyCustomHandler("religionswissenschaftler")],
    allow_delegation=False,
    verbose=True,
)

moderator = Agent(
    name="Moderator",
    role='Generiert einen minutiösen Verlaufsplan für eine Stunde Religionsunterricht',
    backstory='''Methoden und Didaktik Enthusiast mit besonderem Interesse am religionsbezogener Bildung.''',
    goal="Entwicklung von perfekten Unterrichtsstunden für den Religionsunterricht.",
    llm=llm,
    callbacks=[MyCustomHandler("moderator")],
    allow_delegation=False,
    verbose=True,
    
)

def StartCrew(prompt):

    task1 = Task(
      description=f"""Entwickeln Sie Unterrichtsideen auf deutsch zum Thema {prompt}. Berücksichtigen Sie bei ihren Überlegungen, dass eine Religionsstunde nur 45 Minuten dauert. """ ,
      agent=theologe,
      expected_output="Vorschlag für den inhaltlichen Fokus. Danach eine kurze Liste mit maximal 3 Unterrichtsideen und je einem methodischen Vorschlag mit relevante theologischen Perspektiven in deutscher Sprache.",
      
    )

    task2 = Task(
      description=("""Erweitern Sie den Fokus aus Ihrer Perspektive. Prüfen Sie die Unterrichtsideen und ergänzen Sie diese um maximal 2 weitere vor allem aus Ihrer religionswissenschaftlichen Perspektive.
                   Ergänzen Sie die Unterrichtsideen um Aspekte des interreligiösen Dialogs und angemessener Methoden.
                   Berücksichtigen Sie dabei, dass eine Religionsstunde nur 45 Minuten dauert.
                   Geben Sie Feedback und Verbesserungsvorschläge auf deutsch."""),
      agent=religionswissenschaftler,
      expected_output="Feedback und Ergänzungsvorschläge zu den Unterrichtsideen.",
      context=[task1],
    )

    task3 = Task(
      description=("""Erstellen Sie aus dem Gesprächverlauf einen finalen Ablaufplan für den Unterricht auf deutsch. 
                    Stellen Sie sicher, dass der Ablaufplan für eine 45-minütige Stunde geeignet ist.
                   Versichern Sie sich, dass sowohl theologische als auch religionswissenschaftliche, religionsübergreifende Aspekte berücksichtigt wurden.
                   Entwickeln sie aud sem Vroschlägen mit zeitgemäßen Methoden eine mitreißende Unterrichtstunde.
                   Schreiben Sie den Ablaufplan in tabellarischer Form auf. Geben Sie erklärrende Hinweise zu den einzelnen Schritten.
                   Wenn das Ergebniss Ihrer Meinung nach gut ist, legen Sie das Ergebnis einem Menschen zur Prüfung vor.
                   Make sure to check with a human if the draft is good before finalizing your answer.!
                   """),
      agent=moderator,
      expected_output="Ein perfekter Ablaufplan für den Unterricht in tabellarischer Form auf deutsch.",
      human_input=True,
      callback=user_input_callback,
      context=[task1, task2],
    )
    task4 = Task(
      description=("""Nehmen Sie Argumente aus dem menschlichen Feedback auf und ändern Sie den Ablaufplan so,dass das Feedback des Menschen berücksichtigt ."""),
      agent=moderator,
      expected_output="Finale Version des tabellarischen Ablaufplans für eine spannende Religionsstunde",
      context=[task3],
    )

    # Establishing the crew with a hierarchical process
    project_crew = Crew(
        tasks=[task1, task2, task3, task4],  # Tasks to be delegated and executed under the manager's supervision
        agents=[theologe, religionswissenschaftler, moderator],
        manager_llm=llm,
        process=Process.sequential,  # Specifies the hierarchical management approach
        memory=True,
        name="Religionsunterrichtsentwickelerrunde",
        verbose=True,
    )

    result = project_crew.kickoff()
    #print(result)

    chat_interface.send("## Gemeinsames Ergebnis\n"+result, user="assistant", respond=False)


chat_interface = pn.chat.ChatInterface(callback=callback)
chat_interface.send("Beschreibe dein Unterrichttsvorhaben!", user="System", respond=False)
chat_interface.servable()

#StartCrew('Teufel')