added NVDA User guide content and added filter QA

app.py

@@ -11,14 +11,21 @@ from config import DB_CONFIG
 PERSIST_DIR_NAME = "nvdajp-book"
 
 
-def get_retrieval_qa() -> RetrievalQA:
+def get_retrieval_qa(temperature: int, option: str) -> RetrievalQA:
     embeddings = OpenAIEmbeddings()
     db_url, db_api_key, db_collection_name = DB_CONFIG
     client = QdrantClient(url=db_url, api_key=db_api_key)
     db = Qdrant(client=client, collection_name=db_collection_name, embeddings=embeddings)
-    retriever = db.as_retriever()
+    if option is None or option == "All":
+        retriever = db.as_retriever()
+    else:
+        retriever = db.as_retriever(
+            search_kwargs={
+                "filter": {"category": option},
+            }
+        )
     return RetrievalQA.from_chain_type(
-        llm=OpenAI(temperature=0), chain_type="stuff", retriever=retriever, return_source_documents=True,
+        llm=OpenAI(temperature=temperature), chain_type="stuff", retriever=retriever, return_source_documents=True,
     )
 
 
@@ -35,8 +42,8 @@ def get_related_url(metadata):
         yield f'<p>URL: <a href="{url}">{url}</a> (category: {category})</p>'
 
 
-def main(query: str):
-    qa = get_retrieval_qa()
+def main(query: str, option: str, temperature: int):
+    qa = get_retrieval_qa(temperature, option)
     try:
         result = qa(query)
     except InvalidRequestError as e:
@@ -50,7 +57,11 @@ def main(query: str):
 
 nvdajp_book_qa = gr.Interface(
     fn=main,
-    inputs=[gr.Textbox(label="query")],
+    inputs=[
+        gr.Textbox(label="query"),
+        gr.Radio(["All", "ja-book", "ja-nvda-user-guide", "en-nvda-user-guide"], label="絞り込み", info="ドキュメント制限する?"),
+        gr.Slider(0, 2)
+    ],
     outputs=[gr.Textbox(label="answer"), gr.outputs.HTML()],
 )
 

config.py

@@ -1,7 +1,7 @@
 import os
 
 
-SAAS = True
+SAAS = False
 
 
 def get_db_config():

nvda_ug_loader.py

@@ -0,0 +1,107 @@
+from dataclasses import dataclass
+import re
+from typing import Iterator, List
+from langchain.docstore.document import Document
+from langchain.document_loaders.base import BaseLoader
+
+from bs4 import BeautifulSoup, Tag, ResultSet
+import requests
+
+
+RE_HEADERS = re.compile(r"h[23]")
+
+
+@dataclass
+class Content:
+    name: str
+    title: str
+    text: str
+    body: list[Tag]
+
+
+def _get_anchor_name(header: Tag) -> str:
+    for tag in header.previous_elements:
+        if tag.name == "a":
+            return tag.attrs.get("name", "")
+    return ""
+
+
+def _reversed_remove_last_anchor(body: list[Tag]) -> Iterator[Tag]:
+    has_anchor = False
+    for tag in reversed(body):
+        if not has_anchor:
+            if tag.name == "a":
+                has_anchor = True
+            continue
+        else:
+            yield tag
+
+
+def _remove_last_anchor(body: list[Tag]) -> Iterator[Tag]:
+    return reversed(list(_reversed_remove_last_anchor(body)))
+
+
+def _get_bodys_text(body: list[Tag]) -> str:
+    text = ""
+    for tag in body:
+        text += tag.get_text()
+    return text
+
+
+def _get_child_content(header: Tag) -> Content:
+    title = header.get_text()
+    name = _get_anchor_name(header)
+    body = [header]
+    for i, child in enumerate(header.next_elements):
+        if i == 0:
+            continue
+        if child.name == "h2" or child.name == "h3":
+            break
+        body.append(child)
+    removed_next_anchor_body = list(_remove_last_anchor(body))
+    text = _get_bodys_text(removed_next_anchor_body)
+    return Content(name,
+                   title,
+                   text,
+                   removed_next_anchor_body
+                   )
+
+
+def get_contents(headers: ResultSet[Tag]) -> Iterator[Content]:
+    for header in headers:
+        yield _get_child_content(header)
+
+
+class NVDAUserGuideLoader(BaseLoader):
+    """
+    """
+    def __init__(self, url: str, category: str) -> None:
+        self.url = url
+        self.category = category
+
+    def fetch(self) -> BeautifulSoup:
+        res = requests.get(self.url)
+        soup = BeautifulSoup(res.content, 'lxml')
+        return soup
+
+    def lazy_load(self) -> Iterator[Document]:
+        soup = self.fetch()
+        # body = soup.body
+        headers = soup.find_all(RE_HEADERS)
+        for content in get_contents(headers):
+            name = content.name
+            title = content.title
+            text = content.text
+            metadata = {"category": self.category, "source": name, "url": f"{self.url}#{name}", "title": title}
+            yield Document(page_content=text, metadata=metadata)
+
+    def load(self) -> List[Document]:
+        return list(self.lazy_load())
+
+
+if __name__ == "__main__":
+    url = "https://www.nvaccess.org/files/nvda/documentation/userGuide.html"
+    loader = NVDAUserGuideLoader(url, "en-nvda-user-guide")
+    data = loader.load()
+    print(data)
+    # breakpoint()

store.py

@@ -3,6 +3,7 @@ from langchain.text_splitter import RecursiveCharacterTextSplitter
 from langchain.embeddings import OpenAIEmbeddings
 from langchain.vectorstores import Qdrant
 # from qdrant_client import QdrantClient
+from nvda_ug_loader import NVDAUserGuideLoader
 from config import DB_CONFIG
 
 
@@ -46,24 +47,39 @@ def store(texts):
     )
 
 
-def main(path: str):
+def rtd_main(path: str):
     docs = get_documents(path)
     texts = get_text_chunk(docs)
     store(texts)
 
 
+def nul_main(url: str):
+    if "www.nvda.jp" in url:
+        category = "ja-nvda-user-guide"
+    else:
+        category = "en-nvda-user-guide"
+    loader = NVDAUserGuideLoader(url, category)
+    docs = loader.load()
+    texts = get_text_chunk(docs)
+    store(texts)
+
+
 if __name__ == "__main__":
     """
-    $ python store.py "data/rtdocs/nvdajp-book.readthedocs.io/ja/latest"
+    $ python store.py rtd "data/rtdocs/nvdajp-book.readthedocs.io/ja/latest"
+    $ python store.py nul "https://www.nvaccess.org/files/nvda/documentation/userGuide.html"
+    $ python store.py nul "https://www.nvda.jp/nvda2023.1jp/ja/userGuide.html"
     """
     import sys
     args = sys.argv
-    if len(args) != 2:
-        print("No args, you need two args for html_path")
-        docs = get_documents("data/rtdocs/nvdajp-book.readthedocs.io/ja/latest")
-        print(type(docs))
-        breakpoint()
+    if len(args) != 3:
+        print("No args, you need two args for type, html_path")
     else:
-        path = args[1]
-        # dir_name = args[2]
-        main(path)
+        type_ = args[1]
+        path = args[2]
+        if type_ == "rtd":
+            rtd_main(path)
+        elif type_ == "nul":
+            nul_main(path)
+        else:
+            print("No type for store")