Notes

import os
from typing import List, Optional, Any, Dict, Generator
from pydantic import Field
from langchain.llms.base import LLM
from gpt4all import GPT4All

class MyGPT4ALL(LLM):
model_folder_path: str = Field(None, alias='model_folder_path')
model_name: str = Field(None, alias='model_name')
allow_download: bool = Field(None, alias='allow_download')
backend: Optional[str] = 'llama'
temp: Optional[float] = 0.7
top_p: Optional[float] = 0.1
top_k: Optional[int] = 40
n_batch: Optional[int] = 8
n_threads: Optional[int] = 4
n_predict: Optional[int] = 256
max_tokens: Optional[int] = 200
repeat_last_n: Optional[int] = 64
repeat_penalty: Optional[float] = 1.18

gpt4_model_instance: Any = None

def __init__(self, model_folder_path, model_name, allow_download, **kwargs):
super(MyGPT4ALL, self).__init__()
self.model_folder_path = model_folder_path
self.model_name = model_name
self.allow_download = allow_download
self.auto_download()

self.gpt4_model_instance = GPT4All(
model_name=self.model_name,
model_path=self.model_folder_path,
)

def auto_download(self):
# Same as your existing auto_download method
pass

@property
def _get_model_default_parameters(self):
return {
"max_tokens": self.max_tokens,
"n_predict": self.n_predict,
"top_k": self.top_k,
"top_p": self.top_p,
"temp": self.temp,
"n_batch": self.n_batch,
"repeat_penalty": self.repeat_penalty,
"repeat_last_n": self.repeat_last_n,
}

@property
def _identifying_params(self) -> Dict[str, Any]:
return {
'model_name': self.model_name,
'model_path': self.model_folder_path,
'model_parameters': self._get_model_default_parameters
}

@property
def _llm_type(self) -> str:
return 'llama'

def _call(self, prompt: str, stop: Optional[List[str]] = None, **kwargs) -> str:
params = {**self._get_model_default_parameters, **kwargs}
response = self.gpt4_model_instance.generate(prompt, **params)
return response

def stream_output(self, prompt: str, **kwargs) -> Generator[str, None, None]:
"""
Stream output from the model token by token.

Args:
prompt: Input prompt for the model.

Yields:
Tokens as they are generated.
"""
params = {**self._get_model_default_parameters, **kwargs}
for token in self.gpt4_model_instance.generate(prompt, stream=True, **params):
yield token

# Chat history management
class ChatHistory:
def __init__(self):
self.history = []

def add_message(self, role: str, message: str):
"""
Add a message to the chat history.

Args:
role: 'user' or 'assistant'
message: The content of the message.
"""
self.history.append({"role": role, "content": message})

def get_conversation(self) -> str:
"""
Return the conversation as a single formatted string.
"""
return "n".join(
f"{entry['role'].capitalize()}: {entry['content']}" for entry in self.history
)

# Example Usage
if __name__ == "__main__":
model_path = "./models"
model_name = "your_model_name.bin"
gpt4_model = MyGPT4ALL(model_path, model_name, allow_download=True)

chat_history = ChatHistory()

# Simulate a conversation
chat_history.add_message("user", "Hello!")
conversation = chat_history.get_conversation()

# Generate response
print("Assistant is typing...")
for token in gpt4_model.stream_output(conversation):
print(token, end="", flush=True)