feat: chain service

Author: ramamimu

config.py

@@ -14,7 +14,7 @@
 PGPORT = os.getenv("PGPORT")
 
 EMBED_MODEL_PATH = os.getenv("EMBED_MODEL_PATH")
-TEXT_GENERATION_MODEL_PATH = os.getenv("TEXT_GENERATION_MODEL_PATH")
+TEXT_GENERATION_MODEL = os.getenv("TEXT_GENERATION_MODEL")
 DOCUMENT_PATH = os.getenv("DOCUMENT_PATH")
 BASE_KNOWLEDGE_DOCUMENT_PATH = os.getenv("BASE_KNOWLEDGE_DOCUMENT_PATH")
 BASE_KNOWLEDGE_DOCUMENT_NAME = os.getenv("BASE_KNOWLEDGE_DOCUMENT_NAME")

example.env

@@ -1,8 +1,16 @@
 PORT=5000
 OPENAI_API_KEY=
 HUGGINGFACEHUB_API_TOKEN=
+TOKENIZERS_PARALLELISM=true
 
-EMBED_MODEL_PATH=./src/commons/models/indo-sentence-bert-base
-BASE_KNOWLEDGE_DOCUMENT_PATH=./src/commons/documents/ITS-profile
+PGUSER=
+PGHOST=
+PGPASSWORD=
+PGDATABASE=
+PGPORT=5435
+
+EMBED_MODEL_PATH=src/commons/models/indo-sentence-bert-base
+DOCUMENT_PATH=src/commons/documents
+BASE_KNOWLEDGE_DOCUMENT_PATH=src/commons/documents/ITS-profile
 BASE_KNOWLEDGE_DOCUMENT_NAME=ITS-Profile.pdf
-TEXT_GENERATION_MODEL_PATH=
+TEXT_GENERATION_MODEL=gpt-3.5-turbo

installer.txt

@@ -12,7 +12,9 @@ pip3 install pytest==8.2.1
 pip3 install pytest_cov==5.0.0
 pip3 install aiofiles==23.2.1
 pip3 install asyncpg==0.29.0
-pip3 install langchain_community==0.2.0
+pip3 install langchain_community==0.2.1
+pip3 install openai==1.30.2
+pip3 install langchain-openai==0.1.7
 
 INSTALLATION WITH IF
 

request.py

@@ -1,29 +1,64 @@
 import requests
+from config import PORT
 
-url = f"http://localhost:5000"
-
-# ping = requests.get(f'{url}/ping')
-# print(ping.text)
-
-payload = {
-    'id': "1",
-    'question': 'hello world'
-}
-
-headers = {
-    'Content-Type': 'application/json'
-}
-
-def get_stream(url):
-    s = requests.Session()
-    with s.post(url, headers=headers, json=payload, stream=True) as resp:
-    # with s.post(url, headers={'Content-Type': 'application/json'}, stream=True) as resp:
-        for line in resp.iter_lines():
-            if line:
-              print("triggered => ")
-              yield line.decode('utf-8')
-
-url = f'{url}/questions/stream-generator'
-# url = 'https://jsonplaceholder.typicode.com/posts/1'
-for line in get_stream(url):
-    print(line)
+URL = f"http://localhost:{PORT}"
+
+def get_ping():
+    ping = requests.get(f'{URL}/ping')
+    print(ping.text)
+
+def post_stream_generator():
+    payload = {
+        'id': "1",
+        'question': 'hello world'
+    }
+
+    headers = {
+        'Content-Type': 'application/json'
+    }
+
+    def get_stream(url):
+        s = requests.Session()
+        with s.post(url, headers=headers, json=payload, stream=True) as resp:
+        # with s.post(url, headers={'Content-Type': 'application/json'}, stream=True) as resp:
+            for line in resp.iter_lines():
+                if line:
+                    print("triggered => ")
+                    yield line.decode('utf-8')
+
+    url = f'{URL}/questions/stream-generator'
+    # url = 'https://jsonplaceholder.typicode.com/posts/1'
+    data_rcv = ''
+    for line in get_stream(url):
+        data_rcv += line[6:]
+        print(f"{data_rcv}\n")
+
+def post_questions(question):
+    payload = {
+        'id': "1",
+        'question': question
+    }
+
+    headers = {
+        'Content-Type': 'application/json'
+    }
+
+    def get_stream(url):
+        s = requests.Session()
+        with s.post(url, headers=headers, json=payload, stream=True) as resp:
+        # with s.post(url, headers={'Content-Type': 'application/json'}, stream=True) as resp:
+            for line in resp.iter_lines():
+                if line:
+                    print("triggered => ")
+                    yield line.decode('utf-8')
+
+    url = f'{URL}/questions'
+    # url = 'https://jsonplaceholder.typicode.com/posts/1'
+    data_rcv = ''
+    for line in get_stream(url):
+        data_rcv += line[6:]
+        print(f"{data_rcv}\n")
+
+# get_ping()
+# post_stream_generator()
+post_questions("bagaimana Pengambilan MK non-Konversi di semester yang sama dengan pengambilan MK Konversi? jawab bahasa indonesia")

src/api/files/handler.py

@@ -25,7 +25,7 @@ async def put_embed_files_handler(self, name: str = Form(...), file: UploadFile
     
     7. update vectorstore
     
-    8. update chain
+    8. update chain (tested does chain still need to update)
     """
 
     full_path:str = await self._file_storage_service.save_file_to_folder(name, file)

src/api/questions/__init__.py

@@ -1,6 +1,6 @@
 from src.api.questions.handler import QuestionsHandler
 from src.api.questions.routes import routes
 
-def register(lorem_generator_service):
-  questions_handler = QuestionsHandler(lorem_generator_service)
+def register(lorem_generator_service, chain_service):
+  questions_handler = QuestionsHandler(lorem_generator_service, chain_service)
   return routes(questions_handler)

src/api/questions/handler.py

@@ -1,11 +1,19 @@
 from src.commons.types.questions_api_handler_type import PostQuestionStreamGeneratorType
+from sse_starlette.sse import EventSourceResponse
 
 class QuestionsHandler:
-  def __init__(self, lorem_generator_service):
+  def __init__(self, lorem_generator_service, chain_service):
     self._lorem_generator_service = lorem_generator_service
+    self._chain_service = chain_service
 
   async def post_question_stream_generator_handler(self, payload: PostQuestionStreamGeneratorType):
-    # ignore the payload and question
+    # ignore the id and question
     return await self._lorem_generator_service.generate_lorem()
-    
+  
+  async def post_question_stream_handler(self, payload: PostQuestionStreamGeneratorType):
+    return EventSourceResponse(self._chain_streamer(payload.question, payload.id), media_type='text/event-stream')
+
+  async def _chain_streamer(self, question, id):
+    async for chunk in self._chain_service.get_chain().astream(question):
+        yield chunk
 

src/api/questions/routes.py

@@ -8,5 +8,10 @@ def routes(handler) -> List[HandlerRequestType]:
         method=Method.POST.value,
         path="/questions/stream-generator",
         handler=handler.post_question_stream_generator_handler
+    ),
+    HandlerRequestType(
+      method=Method.POST.value,
+      path="/questions",
+      handler=handler.post_question_stream_handler
     )
   ]

src/commons/types/questions_api_handler_type.py

@@ -1,5 +1,6 @@
+from typing import Optional
 from pydantic import BaseModel
 
 class PostQuestionStreamGeneratorType(BaseModel):
-  id: str
+  id: Optional[str] = None
   question: str

src/server/__init__.py

@@ -13,6 +13,7 @@
 from src.services.postgres import PostgresDb
 from src.services.rag.embedding_service import EmbeddingService
 from src.services.rag.vectorstore_service import VectorstoreService
+from src.services.rag.chain_service import ChainService
 
 # endpoints
 import src.api.questions as questions_endpoint
@@ -49,14 +50,15 @@ def configure_endpoint(self):
     files_db_service = FilesDbService(db)
     embedding_service = EmbeddingService(embedding_model)
     vectorstore_service = VectorstoreService(embedding_model, files_db_service)
+    chain_service = ChainService(files_db_service, vectorstore_service)
 
     # service builder
     vectorstore_service.load_all_local_embedding()
 
     # routes initiation
     endpoint_factory = EndpointFactory(self._app)
     endpoint_factory.routes_creator(health_check_endpoint.register())
-    endpoint_factory.routes_creator(questions_endpoint.register(lorem_generator_service))
+    endpoint_factory.routes_creator(questions_endpoint.register(lorem_generator_service, chain_service))
     endpoint_factory.routes_creator(files_endpoint.register(file_storage_service, files_db_service, embedding_service, vectorstore_service))
 
   def run(self):

src/services/rag/chain_service.py

@@ -1,3 +1,46 @@
+from langchain_openai import ChatOpenAI
+from langchain_core.runnables import RunnablePassthrough
+from langchain_core.prompts.chat import ChatPromptTemplate
+from langchain_core.output_parsers import StrOutputParser
+from langchain_community.vectorstores import FAISS
+
+import os
+
+from config import TEXT_GENERATION_MODEL, BASE_KNOWLEDGE_DOCUMENT_PATH
+
 class ChainService:
-  def __init__(self) -> None:
-    pass
+  def __init__(self, files_db_service, vectorstore_service) -> None:
+    self._files_db_service = files_db_service
+    self._vectorstore_service = vectorstore_service
+    self._chain = self._init_chain()
+
+  def _init_chain(self):
+    return (
+      {
+        "context": self._vectorstore_service.get_retriever(), 
+        "question": RunnablePassthrough()
+      }
+      | self._init_prompt()
+      | self._init_llm()
+      | StrOutputParser()
+    )
+
+  def _init_prompt(self):
+    template = """Answer the question based only on the following context:
+    {context}
+
+    Question: {question}
+    """
+
+    prompt = ChatPromptTemplate.from_template(template)
+    return prompt
+  
+  def _init_llm(self):
+    return ChatOpenAI(model_name=TEXT_GENERATION_MODEL, temperature=0, streaming=True)
+
+  def get_chain(self):
+    '''
+    The chain will automatically update since vectorstore update even with no reinitialization
+    '''
+    return self._chain
+

src/services/rag/vectorstore_service.py

@@ -3,14 +3,24 @@
 import os
 
 from config import BASE_KNOWLEDGE_DOCUMENT_PATH
-from src.services.postgres.models.tables import Files
 
 class VectorstoreService:
   def __init__(self, embedding_model, files_db_service) -> None:
     self._embedding_model = embedding_model
     self._vectorstore = FAISS.load_local(folder_path=f"{BASE_KNOWLEDGE_DOCUMENT_PATH}/embedding", embeddings=self._embedding_model, allow_dangerous_deserialization=True)
     self._files_db_service = files_db_service
 
+  def get_retriever(self):
+    retriever = self._vectorstore.as_retriever(search_kwargs={"k": 1})
+    if retriever is None:
+        raise ValueError("Vectorstore as retriever returned None, expected a valid retriever.")
+    return retriever
+
+  def get_vectorstore(self):
+    if self._vectorstore is None:
+        raise ValueError("Vectorstore is None, expected a valid vectorstore instance.")
+    return self._vectorstore
+  
   def load_all_local_embedding(self):
     files = self._files_db_service.get_all_file()
     for file in files:
@@ -23,3 +33,7 @@ def add_vectostore(self, path):
       self._vectorstore.merge_from(local_vectorstore)
     else:
       self._files_db_service.delete_file_by_id(path)
+  
+  def similarity_search(self, question):
+    ss = self._vectorstore.similarity_search(question, k=1)
+    return ss

version_checker.py

@@ -67,5 +67,12 @@
 print(asyncpg.__version__)
 
 import langchain_community
-# 0.2.0
-print(langchain_community.__version__)
+# 0.2.1
+print(langchain_community.__version__)
+
+import openai
+# 1.30.2
+print(openai.__version__)
+
+# checking by pip3 show langchain_openai
+# langchain_openai 0.1.7