[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

EdgeCraftRAG/docker_compose/intel/gpu/arc/README.md

@@ -129,7 +129,7 @@ export MILVUS_ENABLED=0
 # export CHAT_HISTORY_ROUND= # change to your preference
 
 # Launch EC-RAG service with compose
-# Below enviroments are exposed for vLLM config, you can change them to your preference:
+# Below environments are exposed for vLLM config, you can change them to your preference:
 # export VLLM_SERVICE_PORT_0=8100
 # export DTYPE=float16
 # export ZE_AFFINITY_MASK=0

EdgeCraftRAG/docs/API_Guide.md

@@ -219,4 +219,4 @@ curl -X POST http://${HOST_IP}:16010/v1/retrieval -H "Content-Type: application/
 
 ```bash
 curl -X POST http://${HOST_IP}:16011/v1/chatqna -H "Content-Type: application/json" -d '{"messages":"#REPLACE WITH YOUR QUESTION HERE#", "top_n":5, "max_tokens":512}' | jq '.'
-```
+```

EdgeCraftRAG/edgecraftrag/api/v1/chatqna.py

@@ -34,7 +34,7 @@ async def chatqna(request: ChatCompletionRequest):
         set_current_session(sessionid)
         kb = ctx.knowledgemgr.get_active_experience()
         if kb:
-            request.tool_choice = 'auto' if kb.experience_active else 'none'
+            request.tool_choice = "auto" if kb.experience_active else "none"
         generator = ctx.get_pipeline_mgr().get_active_pipeline().generator
         if generator:
             request.model = generator.model_id

EdgeCraftRAG/edgecraftrag/api/v1/knowledge_base.py

@@ -5,10 +5,10 @@
 import json
 import os
 import re
+from typing import Dict, List, Union
 
-from typing import List, Dict, Union
 from edgecraftrag.api.v1.data import add_data
-from edgecraftrag.api_schema import DataIn, KnowledgeBaseCreateIn, ExperienceIn
+from edgecraftrag.api_schema import DataIn, ExperienceIn, KnowledgeBaseCreateIn
 from edgecraftrag.base import IndexerType
 from edgecraftrag.context import ctx
 from edgecraftrag.utils import compare_mappings
@@ -50,7 +50,7 @@ async def create_knowledge_base(knowledge: KnowledgeBaseCreateIn):
                 detail="Knowledge base names must begin with a letter or underscore",
             )
         kb = ctx.knowledgemgr.create_knowledge_base(knowledge)
-        if kb.active and kb.comp_type =="knowledge":
+        if kb.active and kb.comp_type == "knowledge":
             active_pl.indexer.reinitialize_indexer(kb.name)
             active_pl.update_indexer_to_retriever()
         await save_knowledge_to_file()
@@ -68,9 +68,14 @@ async def delete_knowledge_base(knowledge_name: str):
         active_pl = ctx.get_pipeline_mgr().get_active_pipeline()
         if rm_kb.comp_type == "knowledge":
             if not active_kb:
-                raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Please activate a knowledge base before proceeding.")
+                raise HTTPException(
+                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                    detail="Please activate a knowledge base before proceeding.",
+                )
             if active_kb.name == knowledge_name or active_kb.idx == knowledge_name:
-                raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Cannot delete a running knowledge base.")
+                raise HTTPException(
+                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Cannot delete a running knowledge base."
+                )
             kb_file_path = rm_kb.get_file_paths()
             if kb_file_path:
                 if active_pl.indexer.comp_subtype == "milvus_vector":
@@ -80,7 +85,10 @@ async def delete_knowledge_base(knowledge_name: str):
                     active_pl.update_indexer_to_retriever()
         if rm_kb.comp_type == "experience":
             if rm_kb.experience_active:
-                raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Cannot delete a running experience knowledge base.")
+                raise HTTPException(
+                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                    detail="Cannot delete a running experience knowledge base.",
+                )
             else:
                 rm_kb.clear_experiences()
         result = ctx.knowledgemgr.delete_knowledge_base(knowledge_name)
@@ -130,8 +138,10 @@ async def add_file_to_knowledge_base(knowledge_name, file_path: DataIn):
     try:
         active_pl = ctx.get_pipeline_mgr().get_active_pipeline()
         kb = ctx.knowledgemgr.get_knowledge_base_by_name_or_id(knowledge_name)
-        if kb.comp_type =="experience":
-             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="The experience type cannot perform file operations.")
+        if kb.comp_type == "experience":
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND, detail="The experience type cannot perform file operations."
+            )
         # Validate and normalize the user-provided path
         user_path = file_path.local_path
         normalized_path = os.path.normpath(os.path.join(KNOWLEDGE_BASE_ROOT, user_path))
@@ -183,8 +193,10 @@ async def remove_file_from_knowledge_base(knowledge_name, file_path: DataIn):
     try:
         active_pl = ctx.get_pipeline_mgr().get_active_pipeline()
         kb = ctx.knowledgemgr.get_knowledge_base_by_name_or_id(knowledge_name)
-        if kb.comp_type =="experience":
-             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="The experience type cannot perform file operations.")
+        if kb.comp_type == "experience":
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND, detail="The experience type cannot perform file operations."
+            )
         active_kb = ctx.knowledgemgr.get_active_knowledge_base()
         if file_path.local_path in kb.get_file_paths():
             kb.remove_file_path(file_path.local_path)
@@ -212,14 +224,16 @@ async def remove_file_from_knowledge_base(knowledge_name, file_path: DataIn):
     except ValueError as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
 
+
 @kb_app.post("/v1/experience")
 def get_experience_by_question(req: ExperienceIn):
-    kb = ctx.knowledgemgr.get_experience_kb()   
+    kb = ctx.knowledgemgr.get_experience_kb()
     result = kb.get_experience_by_question(req.question)
     if not result:
         raise HTTPException(404, detail="Experience not found")
     return result
 
+
 @kb_app.get("/v1/experiences")
 def get_all_experience():
     kb = ctx.knowledgemgr.get_experience_kb()
@@ -228,48 +242,53 @@ def get_all_experience():
     else:
         return kb
 
+
 @kb_app.patch("/v1/experiences")
 def update_experience(experience: ExperienceIn):
     kb = ctx.knowledgemgr.get_experience_kb()
     result = kb.update_experience(experience.question, experience.content)
     if not result:
-        raise HTTPException(404, detail=f"Question not found")
+        raise HTTPException(404, detail="Question not found")
     return result
 
+
 @kb_app.delete("/v1/experiences")
-def delete_experience(req :ExperienceIn):
+def delete_experience(req: ExperienceIn):
     kb = ctx.knowledgemgr.get_experience_kb()
     success = kb.delete_experience(req.question)
     if not success:
         raise HTTPException(404, detail=f"Question {req.question} not found")
-    return {"message": f"Question deleted"}
+    return {"message": "Question deleted"}
+
 
 @kb_app.post("/v1/multiple_experiences/check")
 def check_duplicate_multiple_experiences(experiences: List[Dict[str, Union[str, List[str]]]]):
     kb = ctx.knowledgemgr.get_experience_kb()
     if not kb:
-       raise HTTPException(404, detail=f"No active experience type knowledge base")
+        raise HTTPException(404, detail="No active experience type knowledge base")
     all_existing = kb.get_all_experience()
     existing_questions = {item["question"] for item in all_existing}
     new_questions = [exp["question"] for exp in experiences if "question" in exp]
     duplicate_questions = [q for q in new_questions if q in existing_questions]
     if duplicate_questions:
         return {"code": 2001, "detail": "Duplicate experiences are appended OR overwritten!"}
     else:
-        kb.add_multiple_experiences(experiences=experiences, flag=True)  
-        return {"status": "success","detail": "No duplicate experiences, added successfully"}
+        kb.add_multiple_experiences(experiences=experiences, flag=True)
+        return {"status": "success", "detail": "No duplicate experiences, added successfully"}
+
 
 @kb_app.post("/v1/multiple_experiences/confirm")
-def confirm_multiple_experiences(experiences: List[Dict[str, Union[str, List[str]]]],flag: bool):
+def confirm_multiple_experiences(experiences: List[Dict[str, Union[str, List[str]]]], flag: bool):
     kb = ctx.knowledgemgr.get_experience_kb()
     try:
         if not kb:
-            raise HTTPException(404, detail=f"No active experience type knowledge base")
+            raise HTTPException(404, detail="No active experience type knowledge base")
         kb.add_multiple_experiences(experiences=experiences, flag=flag)
         return {"status": "success", "detail": "Experiences added successfully"}
     except Exception as e:
         raise HTTPException(status_code=500, detail=f"Add Failure：{str(e)}")
 
+
 @kb_app.post("/v1/experiences/files")
 def add_experiences_from_file(req: DataIn):
     kb = ctx.knowledgemgr.get_experience_kb()
@@ -279,6 +298,7 @@ def add_experiences_from_file(req: DataIn):
     except Exception as e:
         raise HTTPException(status_code=400, detail=str(e))
 
+
 # Update knowledge base data
 async def update_knowledge_base_handler(file_path=None, knowledge_name: str = "default_kb", add_file: bool = False):
     if ctx.get_pipeline_mgr().get_active_pipeline() is None:
@@ -334,7 +354,7 @@ async def load_knowledge_from_file():
             for Knowledgebase_data in all_data:
                 pipeline_req = KnowledgeBaseCreateIn(**Knowledgebase_data)
                 kb = ctx.knowledgemgr.create_knowledge_base(pipeline_req)
-                if  kb.comp_type =="knowledge":
+                if kb.comp_type == "knowledge":
                     if Knowledgebase_data["file_map"]:
                         if active_pl.indexer.comp_subtype != "milvus_vector" and Knowledgebase_data["active"]:
                             for file_path in Knowledgebase_data["file_map"].values():
@@ -364,7 +384,14 @@ async def save_knowledge_to_file():
         kb_base = ctx.knowledgemgr.get_all_knowledge_bases()
         knowledgebases_data = []
         for kb in kb_base:
-            kb_json = {"name": kb.name, "description": kb.description, "active": kb.active, "file_map": kb.file_map, "comp_type": kb.comp_type, "experience_active": kb.experience_active}
+            kb_json = {
+                "name": kb.name,
+                "description": kb.description,
+                "active": kb.active,
+                "file_map": kb.file_map,
+                "comp_type": kb.comp_type,
+                "experience_active": kb.experience_active,
+            }
             knowledgebases_data.append(kb_json)
         json_str = json.dumps(knowledgebases_data, indent=2, ensure_ascii=False)
         with open(KNOWLEDGEBASE_FILE, "w", encoding="utf-8") as f:
@@ -375,15 +402,18 @@ async def save_knowledge_to_file():
 
 all_pipeline_milvus_maps = {}
 current_pipeline_kb_map = {}
+
+
 async def refresh_milvus_map(milvus_name):
     current_pipeline_kb_map.clear()
     knowledge_bases_list = await get_all_knowledge_bases()
     for kb in knowledge_bases_list:
         if kb.comp_type == "experience":
-                    continue
+            continue
         current_pipeline_kb_map[kb.name] = kb.file_map
     all_pipeline_milvus_maps[milvus_name] = copy.deepcopy(current_pipeline_kb_map)
 
+
 async def Synchronizing_vector_data(old_active_pl, new_active_pl):
     try:
         active_kb = ctx.knowledgemgr.get_active_knowledge_base()

EdgeCraftRAG/edgecraftrag/api_schema.py

@@ -81,11 +81,13 @@ class KnowledgeBaseCreateIn(BaseModel):
     description: Optional[str] = None
     active: Optional[bool] = None
     comp_type: Optional[str] = "knowledge"
-    experience_active:  Optional[bool] = None
+    experience_active: Optional[bool] = None
+
 
 class ExperienceIn(BaseModel):
     question: str
     content: list[str] = None
 
+
 class MilvusConnectRequest(BaseModel):
     vector_uri: str

EdgeCraftRAG/edgecraftrag/components/knowledge_base.py

@@ -1,8 +1,9 @@
 # Copyright (C) 2024 Intel Corporation
 # SPDX-License-Identifier: Apache-2.0
 
-import os, json
-from typing import Any, List, Optional, Dict, Union
+import json
+import os
+from typing import Any, Dict, List, Optional, Union
 
 from edgecraftrag.base import BaseComponent
 from pydantic import model_serializer
@@ -40,7 +41,7 @@ def ensure_file_exists(self):
         dir_path = os.path.dirname(self.file_paths[0])
         os.makedirs(dir_path, exist_ok=True)
         if not os.path.exists(self.file_paths[0]):
-            with open(self.file_paths[0], 'w', encoding='utf-8') as f:
+            with open(self.file_paths[0], "w", encoding="utf-8") as f:
                 json.dump([], f, ensure_ascii=False, indent=4)
 
     def get_all_experience(self) -> List[Dict]:
@@ -49,80 +50,82 @@ def get_all_experience(self) -> List[Dict]:
             self.file_paths.append(experinence_file)
         if not os.path.isfile(self.file_paths[0]):
             self.ensure_file_exists()
-        with open(self.file_paths[0], 'r', encoding='utf-8') as f:
+        with open(self.file_paths[0], "r", encoding="utf-8") as f:
             return json.load(f)
 
     def get_experience_by_question(self, question: str) -> Optional[Dict]:
         for item in self.get_all_experience():
-            if item.get('question') == question:
+            if item.get("question") == question:
                 return item
         return None
 
-    def add_multiple_experiences(self, experiences: List[Dict[str, Union[str, List[str]]]], flag: bool = True) -> List[Dict]:
+    def add_multiple_experiences(
+        self, experiences: List[Dict[str, Union[str, List[str]]]], flag: bool = True
+    ) -> List[Dict]:
         all_experiences = self.get_all_experience()
         result = []
         for exp in experiences:
-            question = exp.get('question')
+            question = exp.get("question")
             if not question:
                 raise ValueError("Must exist when uploading question")
-            content = exp.get('content', [])
+            content = exp.get("content", [])
             found = False
             for item in all_experiences:
-                if item['question'] == question:
+                if item["question"] == question:
                     if flag:
-                        item['content'].extend([c for c in content if c not in item['content']])
+                        item["content"].extend([c for c in content if c not in item["content"]])
                     else:
-                        item['content'] = content
+                        item["content"] = content
                     result.append(item)
                     found = True
                     break
             if not found:
-                new_item = {'question': question, 'content': content}
+                new_item = {"question": question, "content": content}
                 all_experiences.append(new_item)
                 result.append(new_item)
-        with open(self.file_paths[0], 'w', encoding='utf-8') as f:
+        with open(self.file_paths[0], "w", encoding="utf-8") as f:
             json.dump(all_experiences, f, ensure_ascii=False, indent=4)
         return result
 
     def delete_experience(self, question: str) -> bool:
         items = self.get_all_experience()
-        remaining_items = [item for item in items if item.get('question') != question]
+        remaining_items = [item for item in items if item.get("question") != question]
         if len(remaining_items) == len(items):
             return False
-        with open(self.file_paths[0], 'w', encoding='utf-8') as f:
+        with open(self.file_paths[0], "w", encoding="utf-8") as f:
             json.dump(remaining_items, f, ensure_ascii=False, indent=4)
         return True
 
     def clear_experiences(self) -> bool:
         all_experiences = self.get_all_experience()
-        with open(self.file_paths[0], 'w', encoding='utf-8') as f:
+        with open(self.file_paths[0], "w", encoding="utf-8") as f:
             json.dump([], f, ensure_ascii=False, indent=4)
         return True
 
     def update_experience(self, question: str, content: List[str]) -> Optional[Dict]:
         items = self.get_all_experience()
         for i, item in enumerate(items):
-            if item.get('question') == question:
-                updated_item = {'question': question, 'content': content}
+            if item.get("question") == question:
+                updated_item = {"question": question, "content": content}
                 items[i] = updated_item
-                with open(self.file_paths[0], 'w', encoding='utf-8') as f:
+                with open(self.file_paths[0], "w", encoding="utf-8") as f:
                     json.dump(items, f, ensure_ascii=False, indent=4)
                 return updated_item
         return None
 
     def add_experiences_from_file(self, file_path: str, flag: bool = False) -> List[Dict]:
-        if not file_path.endswith('.json'):
+        if not file_path.endswith(".json"):
             raise ValueError("File upload type error")
         try:
-            with open(file_path, 'r', encoding='utf-8') as f:
+            with open(file_path, "r", encoding="utf-8") as f:
                 experiences = json.load(f)
             if not isinstance(experiences, list):
                 raise ValueError("The contents of the file must be a list")
             return self.add_multiple_experiences(experiences=experiences, flag=flag)
         except json.JSONDecodeError as e:
-            raise ValueError(f"File parsing failure")
+            raise ValueError("File parsing failure")
         except Exception as e:
-            raise RuntimeError(f"File Error")
+            raise RuntimeError("File Error")
 
     def calculate_totals(self):
         if self.comp_type == "knowledge":
@@ -146,6 +149,6 @@ def ser_model(self):
             "description": self.description,
             "active": self.active,
             "experience_active": self.experience_active,
-            "total": self.calculate_totals()
+            "total": self.calculate_totals(),
         }
         return set