File size: 4,625 Bytes
8c92d3f b4cb4dd f63de68 6a8f373 57ac245 6a8f373 b4cb4dd 8c92d3f f83ef13 4a7b77c 6a8f373 4a7b77c 6a8f373 8c92d3f 1950d0f 8c92d3f 989b6d4 8c92d3f 8675f62 1950d0f 8c92d3f 6a8f373 1950d0f 8c92d3f 9f0143f 8c92d3f |
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 |
from typing import Any, List, Mapping, Optional
from langchain_core.callbacks.manager import CallbackManagerForLLMRun
from langchain_core.language_models.llms import LLM
from typing import Literal
import requests
from langchain.prompts import PromptTemplate, ChatPromptTemplate
from operator import itemgetter
from langchain.memory import ChatMessageHistory, ConversationBufferMemory
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models import ChatOpenAI
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.messages import AIMessage, HumanMessage
from langchain_community.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyMuPDFLoader
import os
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS
from datasets import load_dataset
def create_vectorstore():
data = load_dataset("ruslanmv/ai-medical-chatbot", split='train')
emb_model = HuggingFaceEmbeddings(model_name='sentence-transformers/paraphrase-multilingual-mpnet-base-v2', encode_kwargs={'normalize_embeddings': True})
faiss = FAISS.from_texts(data['Doctor'], emb_model)
return faiss
def custom_chain_with_history(llm, memory):
prompt = PromptTemplate.from_template("""You are a helpful, respectful, and honest assistant. Always answer as helpfully as possible while being safe. Your answers should not include any harmful, unethical, racist, sexist, toxic dangerous, or illegal content. Please ensure that your responses are socially unbiased and positive in nature.
You have the access to the following context information:
{context}
If a question does not make any sense, or is not factually coherent, explain why instead of answering something not correct. If you don't know the answer to a question, please don't share false information.
{chat_history}
### User: {question}
### Assistant: """)
def prompt_memory(memory):
t = ""
for x in memory.chat_memory.messages:
# for x in memory.messages:
t += f"### Assistant: {x.content}\n" if type(x) is AIMessage else f"### User: {x.content}\n"
return "" if len(t) == 0 else t
def format_docs(docs):
print(len(docs))
return "\n".join([f"{i+1}. {d.page_content}" for i,d in enumerate(docs)])
return {
"chat_history":lambda x:prompt_memory(x['memory']),
"question":lambda x:x['question'],
"context": itemgetter("question") | create_vectorstore().as_retriever(search_type="similarity", search_kwargs={"k": 6}) | format_docs
} | prompt | llm
class CustomLLM(LLM):
repo_id : str
api_token : str
model_type: Literal["text2text-generation", "text-generation"]
max_new_tokens: int = None
temperature: float = 0.001
timeout: float = None
top_p: float = None
top_k : int = None
repetition_penalty : float = None
stop : List[str] = []
@property
def _llm_type(self) -> str:
return "custom"
def _call(
self,
prompt: str,
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> str:
headers = {"Authorization": f"Bearer {self.api_token}"}
API_URL = f"https://api-inference.huggingface.co/models/{self.repo_id}"
parameters_dict = {
'max_new_tokens': self.max_new_tokens,
'temperature': self.temperature,
'timeout': self.timeout,
'top_p': self.top_p,
'top_k': self.top_k,
'repetition_penalty': self.repetition_penalty,
'stop':self.stop
}
if self.model_type == 'text-generation':
parameters_dict["return_full_text"]=False
data = {"inputs": prompt, "parameters":parameters_dict, "options":{"wait_for_model":True}}
data = requests.post(API_URL, headers=headers, json=data).json()
return data[0]['generated_text']
@property
def _identifying_params(self) -> Mapping[str, Any]:
"""Get the identifying parameters."""
return {
'repo_id': self.repo_id,
'model_type':self.model_type,
'stop_sequences':self.stop,
'max_new_tokens': self.max_new_tokens,
'temperature': self.temperature,
'timeout': self.timeout,
'top_p': self.top_p,
'top_k': self.top_k,
'repetition_penalty': self.repetition_penalty
}
|