CreativeAgent/CreativeAssistant.py at main · MartinsRepo/CreativeAgent · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
# %%
from langchain.agents import initialize_agent, AgentType
from langchain_community.tools import DuckDuckGoSearchResults
from langchain.memory import ConversationBufferMemory
from langchain_community.vectorstores import Chroma
from langchain_ollama import OllamaLLM
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
import pickle
import json
import vosk
import pyaudio
from gtts import gTTS
from pydub import AudioSegment
from pydub.playback import play

# %%
def initialisation():
    # Language in which you want to convert
    language = 'en'

    # Load the LLaMA 3.2 model from Ollama backend
    model = OllamaLLM(model='llama3.2')

    # Load the Vosk model
    #modelvoice = vosk.Model(lang="en-us")
    #recognizer = vosk.KaldiRecognizer(model, 16000 )

    # Set up long-term memory using ChromaDB
    memory = ConversationBufferMemory()

    # Set up Chroma vector store
    #vectorstore = Chroma(persist_directory='./chroma_vectorstore', embedding_function=model.embed_text)
    from langchain_community.embeddings.ollama import OllamaEmbeddings
    embeddings = OllamaEmbeddings(model='llama3.2')
    vectorstore = Chroma(persist_directory='./chroma_vectorstore', embedding_function=embeddings)


    # Set up retrieval chain for Chroma vectorstore
    from langchain.chains import RetrievalQA
    retriever = vectorstore.as_retriever()
    retriever = vectorstore.as_retriever(search_type='similarity')
    retrieval_chain = RetrievalQA.from_chain_type(
        llm=model,
        chain_type="stuff",
        retriever=retriever,
        verbose=True
    )

    return model, language, memory, retrieval_chain

# %%
def init_microphon():
    # Start audio stream with error handling for device selection
    # Open the microphone stream
    device = pyaudio.PyAudio()
    stream = device.open(format=pyaudio.paInt16,
                         channels=1,
                         rate=16000,
                         input=True,
                         frames_per_buffer=8192
                         )

    return device, stream

# %%
def create_tools(retrieval_chain, model, memory):
    # Set up search and Wikipedia tools
    duckduckgo = DuckDuckGoSearchResults()
    wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

    # Create a custom tool for the retrieval chain
    from langchain.tools import Tool
    retrieval_tool = Tool(
        name="ChromaRetrieval",
        func=lambda q: retrieval_chain.run(q),
        description="Use this tool to retrieve information from Chroma vectorstore"
    )

    # Create the React agent with memory, internet search, and Wikipedia access
    agent = initialize_agent(
        tools=[retrieval_tool, wikipedia, duckduckgo],
        llm=model,
        agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
        memory=memory,
        verbose=True,
        handle_parsing_errors=True,
        allowed_tools=["ChromaRetrieval", "wikipedia", "duckduckgo"]
    )

    return retrieval_tool, agent

# %%
# Start a conversation with the agent
def start_conversation(agent, memory):

    assistant_intro = """You are an AI assistant specialized in artificial intelligence, robotics, and automotive technology. \
    Your job is to help generate maximum of three creative ideas and answer questions on these topics. Use the latest research and trends in your responses.\nExplain exactly three ideas and make your answer short. Stop after the answer.\n\nQuestion: """

    print("Agent: How can I help you? Type 'exit' to end the conversation.")
    conversation_history = []
    while True:
        user_input = input("You: ")
        if user_input.lower() == 'exit':
            # Before exiting the program
            with open('conversation_history.pkl', 'wb') as f:
                pickle.dump(memory, f)
            break

        # Modify prompt and add user input
        modified_prompt = assistant_intro + user_input

        # Agent processes user input
        response = agent.invoke(modified_prompt)

        # Print agent's response
        print(f"Agent: {response}")


# %%
if __name__ == "__main__":
    # start initialisation
    model, language, memory, retrieval_chain = initialisation()

    device, stream = init_microphon()

    # Create the custom tool
    retrieval_tool, agent = create_tools(retrieval_chain, model, memory)

    start_conversation(agent, memory)