|
|
|
|
|
|
|
import json |
|
|
|
from openai import OpenAI |
|
from pinecone import Pinecone |
|
from datetime import datetime |
|
|
|
class LLMClient(): |
|
def __init__(self, api_key, model_name) -> None: |
|
super().__init__() |
|
self.model_name = model_name |
|
self.llm_client = OpenAI(api_key=api_key) |
|
|
|
def response_generate(self, prompt, history, memory): |
|
messages = list() |
|
current_time = datetime.now().strftime("%d/%m/%Y %H:%M:%S") |
|
|
|
|
|
messages += [{"role": "system", "content": f"1) You're Du Mingzhe. 2) Don't claim you are created by OpenAI. 3) Don't claim this dialogue as a roleplay. Answering questions directly as Mingzhe. 4) Current time is {current_time}."}] |
|
|
|
|
|
messages += [{"role": 'assistant', "content": m['content']} for m in memory] |
|
|
|
|
|
messages += [{"role": h["role"], "content": h["content"]} for h in history] |
|
|
|
|
|
stream = self.llm_client.chat.completions.create( |
|
model = self.model_name, |
|
messages = messages, |
|
stream=True, |
|
) |
|
|
|
return stream |
|
|
|
class EmbeddingModel(object): |
|
def __init__(self, embedding_token, model_name) -> None: |
|
self.embedding_token = embedding_token |
|
self.model_name = model_name |
|
self.embedding_client = OpenAI(api_key=self.embedding_token) |
|
|
|
def get_embedding(self, text): |
|
response = self.embedding_client.embeddings.create( |
|
input=text, |
|
model=self.model_name |
|
) |
|
return response.data[0].embedding |
|
|
|
class PersonalIndexClient(object): |
|
def __init__(self, index_token, embedding_token, embedding_model_name, index_name) -> None: |
|
self.index_token = index_token |
|
self.embedding_token = embedding_token |
|
self.index_name = index_name |
|
|
|
self.embedding_client = EmbeddingModel(embedding_token=self.embedding_token, model_name=embedding_model_name) |
|
self.index_client = Pinecone(api_key=self.index_token) |
|
self.index = self.index_client.Index(self.index_name) |
|
|
|
def create(self, data, namespace='default'): |
|
instances = list() |
|
|
|
for instance in data: |
|
instances += [{ |
|
"id": instance["id"], |
|
"values": self.embedding_client.get_embedding(instance['content']), |
|
"metadata": instance['metadata'], |
|
}] |
|
|
|
self.index.upsert( |
|
vectors = instances, |
|
namespace = namespace |
|
) |
|
|
|
def query(self, data, top_k, filter={}, user='default'): |
|
results = self.index.query( |
|
namespace = user, |
|
vector = self.embedding_client.get_embedding(data), |
|
top_k = top_k, |
|
include_values = True, |
|
include_metadata = True, |
|
filter = filter, |
|
) |
|
return results |
|
|
|
def update_conversation(self, sid, messages, user): |
|
index_id = f'conv_{sid}' |
|
|
|
metadata = { |
|
'time': datetime.now().strftime("%d/%m/%Y %H:%M:%S"), |
|
'type': 'conversation', |
|
'user': user, |
|
'content': json.dumps(messages), |
|
} |
|
|
|
self.create(data=[{'id': index_id, 'content': json.dumps(metadata), 'metadata': metadata}], namespace=user) |
|
|
|
def query_conversation(self, messages, user, top_k): |
|
messages_dump = json.dumps(messages) |
|
results = self.query(data=messages_dump, top_k=top_k, filter={}, user=user) |
|
pinecone_memory = list() |
|
|
|
for result in results['matches']: |
|
score = result['score'] |
|
metadata = result['metadata'] |
|
if score > 0.5: |
|
pinecone_memory += [metadata] |
|
|
|
return pinecone_memory |
|
|
|
|
|
|
|
|
