fmt

Author: framsouza

supporting-blog-content/github-assistant/evaluation.py

@@ -20,20 +20,29 @@
 logging.basicConfig(stream=sys.stdout, level=logging.INFO)
 logging.getLogger().addHandler(logging.StreamHandler(stream=sys.stdout))
 
-parser = argparse.ArgumentParser(description='Process documents and questions for evaluation.')
-parser.add_argument('--num_documents', type=int, default=None,
-                    help='Number of documents to process (default: all)')
-parser.add_argument('--skip_documents', type=int, default=0,
-                    help='Number of documents to skip at the beginning (default: 0)')
-parser.add_argument('--num_questions', type=int, default=None,
-                    help='Number of questions to process (default: all)')
-parser.add_argument('--skip_questions', type=int, default=0,
-                    help='Number of questions to skip at the beginning (default: 0)')
-parser.add_argument('--process_last_questions', action='store_true',
-                    help='Process last N questions instead of first N')
+parser = argparse.ArgumentParser(description="Process documents and questions for evaluation.")
+parser.add_argument("--num_documents",
+                    type=int, 
+                    default=None,
+                    help="Number of documents to process (default: all)")
+parser.add_argument("--skip_documents", 
+                    type=int, 
+                    default=0,
+                    help="Number of documents to skip at the beginning (default: 0)")
+parser.add_argument("--num_questions", 
+                    type=int, 
+                    default=None,
+                    help="Number of questions to process (default: all)")
+parser.add_argument("--skip_questions",
+                    type=int, 
+                    default=0,
+                    help="Number of questions to skip at the beginning (default: 0)")
+parser.add_argument("--process_last_questions",
+                    action="store_true",
+                    help="Process last N questions instead of first N")
 args = parser.parse_args()
 
-load_dotenv('.env')
+load_dotenv(".env")
 
 reader = SimpleDirectoryReader("/tmp/elastic/production-readiness-review")
 documents = reader.load_data()
@@ -43,10 +52,10 @@
 
 
 if args.skip_documents > 0:
-    documents = documents[args.skip_documents:]
+    documents = documents[args.skip_documents :]
 
 if args.num_documents is not None:
-    documents = documents[:args.num_documents]
+    documents = documents[: args.num_documents]
 
 print(f"Number of documents loaded: {len(documents)}")
 
@@ -57,7 +66,7 @@
 try:
     eval_questions = data_generator.generate_questions_from_nodes()
     if isinstance(eval_questions, str):
-        eval_questions_list = eval_questions.strip().split('\n')
+        eval_questions_list = eval_questions.strip().split("\n")
     else:
         eval_questions_list = eval_questions
     eval_questions_list = [q for q in eval_questions_list if q.strip()]
@@ -100,32 +109,32 @@ def display_eval_df(
     eval_result_relevancy: EvaluationResult,
     eval_result_faith: EvaluationResult,
 ) -> None:
-    relevancy_feedback = getattr(eval_result_relevancy, 'feedback', '')
-    relevancy_passing = getattr(eval_result_relevancy, 'passing', False)
-    relevancy_passing_str = 'Pass' if relevancy_passing else 'Fail'
+    relevancy_feedback = getattr(eval_result_relevancy, "feedback", "")
+    relevancy_passing = getattr(eval_result_relevancy, "passing", False)
+    relevancy_passing_str = "Pass" if relevancy_passing else "Fail"
 
     relevancy_score = 1.0 if relevancy_passing else 0.0
 
-    faithfulness_feedback = getattr(eval_result_faith, 'feedback', '')
-    faithfulness_passing_bool = getattr(eval_result_faith, 'passing', False)
-    faithfulness_passing = 'Pass' if faithfulness_passing_bool else 'Fail'
+    faithfulness_feedback = getattr(eval_result_faith, "feedback", "")
+    faithfulness_passing_bool = getattr(eval_result_faith, "passing", False)
+    faithfulness_passing = "Pass" if faithfulness_passing_bool else "Fail"
 
     def wrap_text(text, width=50):
         if text is None:
-            return ''
+            return ""
         text = str(text)
-        text = text.replace('\r', '')
-        lines = text.split('\n')
+        text = text.replace("\r", "")
+        lines = text.split("\n")
         wrapped_lines = []
         for line in lines:
             wrapped_lines.extend(textwrap.wrap(line, width=width))
-            wrapped_lines.append('')
-        return '\n'.join(wrapped_lines)
+            wrapped_lines.append("")
+        return "\n".join(wrapped_lines)
 
     if response.source_nodes:
         source_content = wrap_text(response.source_nodes[0].node.get_content())
     else:
-        source_content = ''
+        source_content = ""
 
     eval_data = {
         "Query": wrap_text(query),
@@ -141,7 +150,7 @@ def wrap_text(text, width=50):
     eval_df = pd.DataFrame([eval_data])
 
     print("\nEvaluation Result:")
-    print(tabulate(eval_df, headers='keys', tablefmt='grid', showindex=False, stralign='left'))
+    print(tabulate(eval_df, headers="keys", tablefmt="grid", showindex=False, stralign="left"))
 
 query_engine = vector_index.as_query_engine(llm=llm)
 

supporting-blog-content/github-assistant/index.py

@@ -19,11 +19,11 @@
 
 #logging.basicConfig(stream=sys.stdout, level=logging.INFO)
 #logging.getLogger().addHandler(logging.StreamHandler(stream=sys.stdout))
-#logging.getLogger('elasticsearch').setLevel(logging.DEBUG)
+#logging.getLogger("elasticsearch").setLevel(logging.DEBUG)
 
 nest_asyncio.apply()
 
-load_dotenv('.env')
+load_dotenv(".env")
 
 Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-large")
 Settings.chunk_lines = 1024
@@ -35,7 +35,9 @@
 def clone_repository(owner, repo, branch, base_path="/tmp"):
     branch = branch or os.getenv("GITHUB_BRANCH")
     if not branch:
-        raise ValueError("Branch is not provided and GITHUB_BRANCH environment variable is not set.")
+        raise ValueError(
+            "Branch is not provided and GITHUB_BRANCH environment variable is not set."
+        )
 
     local_repo_path = os.path.join(base_path, owner, repo)
     clone_url = f"https://github.com/{owner}/{repo}.git"
@@ -50,7 +52,9 @@ def clone_repository(owner, repo, branch, base_path="/tmp"):
         try:
             os.makedirs(local_repo_path, exist_ok=True)
             print(f"Attempting to clone repository... Attempt {attempt + 1}")
-            subprocess.run(["git", "clone", "-b", branch, clone_url, local_repo_path], check=True)
+            subprocess.run(
+                ["git", "clone", "-b", branch, clone_url, local_repo_path], check=True
+            )
             print(f"Repository cloned into {local_repo_path}.")
             return local_repo_path
         except subprocess.CalledProcessError:
@@ -78,53 +82,63 @@ def collect_and_print_file_summary(file_summary):
         print(summary)
 
 def parse_documents():
-    owner = os.getenv('GITHUB_OWNER')
-    repo = os.getenv('GITHUB_REPO')
-    branch = os.getenv('GITHUB_BRANCH')
-    base_path = os.getenv('BASE_PATH', "/tmp")  
+    owner = os.getenv("GITHUB_OWNER")
+    repo = os.getenv("GITHUB_REPO")
+    branch = os.getenv("GITHUB_BRANCH")
+    base_path = os.getenv("BASE_PATH", "/tmp")  
 
     if not owner or not repo:
-        raise ValueError("GITHUB_OWNER and GITHUB_REPO environment variables must be set.")
+        raise ValueError(
+            "GITHUB_OWNER and GITHUB_REPO environment variables must be set."
+        )
 
     local_repo_path = clone_repository(owner, repo, branch, base_path)
 
     nodes = []
     file_summary = []
 
-    ts_parser = get_parser('typescript')
-    py_parser = get_parser('python')
-    go_parser = get_parser('go')
-    js_parser = get_parser('javascript')
-    bash_parser = get_parser('bash')
-    yaml_parser = get_parser('yaml')
+    ts_parser = get_parser("typescript")
+    py_parser = get_parser("python")
+    go_parser = get_parser("go")
+    js_parser = get_parser("javascript")
+    bash_parser = get_parser("bash")
+    yaml_parser = get_parser("yaml")
 
     parsers_and_extensions = [
         (SentenceSplitter(), [".md"]),
-        (CodeSplitter(language='python', parser=py_parser), [".py", ".ipynb"]),
-        (CodeSplitter(language='typescript', parser=ts_parser), [".ts"]),
-        (CodeSplitter(language='go', parser=go_parser), [".go"]),
-        (CodeSplitter(language='javascript', parser=js_parser), [".js"]),
-        (CodeSplitter(language='bash', parser=bash_parser), [".bash", ",sh"]),
-        (CodeSplitter(language='yaml', parser=yaml_parser), [".yaml", ".yml"]),
+        (CodeSplitter(language="python", parser=py_parser), [".py", ".ipynb"]),
+        (CodeSplitter(language="typescript", parser=ts_parser), [".ts"]),
+        (CodeSplitter(language="go", parser=go_parser), [".go"]),
+        (CodeSplitter(language="javascript", parser=js_parser), [".js"]),
+        (CodeSplitter(language="bash", parser=bash_parser), [".bash", ",sh"]),
+        (CodeSplitter(language="yaml", parser=yaml_parser), [".yaml", ".yml"]),
         (JSONNodeParser(), [".json"]),
     ]
 
     for parser, extensions in parsers_and_extensions:
         matching_files = []
         for ext in extensions:
-            matching_files.extend(glob.glob(f"{local_repo_path}/**/*{ext}", recursive=True))
+            matching_files.extend(
+                glob.glob(f"{local_repo_path}/**/*{ext}", recursive=True)
+            )
 
         if len(matching_files) > 0:
-            file_summary.append(f"Found {len(matching_files)} {', '.join(extensions)} files in the repository.")
-            loader = SimpleDirectoryReader(input_dir=local_repo_path, required_exts=extensions, recursive=True)
+            file_summary.append(
+                f"Found {len(matching_files)} {", ".join(extensions)} files in the repository."
+            )
+            loader = SimpleDirectoryReader(
+                input_dir=local_repo_path, required_exts=extensions, recursive=True
+            )
             docs = loader.load_data()
             parsed_nodes = parser.get_nodes_from_documents(docs)
 
             print_docs_and_nodes(docs, parsed_nodes)
 
             nodes.extend(parsed_nodes)
         else:
-            file_summary.append(f"No {', '.join(extensions)} files found in the repository.")
+            file_summary.append(
+                f"No {", ".join(extensions)} files found in the repository."
+            )
 
     collect_and_print_file_summary(file_summary)
     print("\n")
@@ -144,7 +158,7 @@ def get_es_vector_store():
                 es_cloud_id=es_cloud_id,
                 es_user=es_user,
                 es_password=es_password,
-                batch_size=100
+                batch_size=100,
             )
             print("Elasticsearch store initialized.")
             return es_vector_store

supporting-blog-content/github-assistant/query.py

@@ -21,7 +21,7 @@ def run_query_sync():
             llm=openai_llm,
             similarity_top_k=3,
             streaming=False, 
-            response_mode="tree_summarize"
+            response_mode="tree_summarize",
         )
 
         bundle = QueryBundle(query, embedding=embed_model.get_query_embedding(query))
@@ -31,9 +31,13 @@ def run_query_sync():
     except Exception as e:
         print(f"An error occurred while running the query: {e}")
     finally:
-        if hasattr(openai_llm, 'client') and isinstance(openai_llm.client, httpx.Client):
+        if hasattr(openai_llm, "client") and isinstance(
+            openai_llm.client, httpx.Client
+            ):
             openai_llm.client.close()
-        if hasattr(embed_model, 'client') and isinstance(embed_model.client, httpx.Client):
+        if hasattr(embed_model, "client") and isinstance(
+            embed_model.client, httpx.Client
+            ):
             embed_model.client.close()
         if hasattr(es_vector_store, "close"):
             es_vector_store.close()