Spaces:
Sleeping
Sleeping
| from subprocess import call | |
| import streamlit as st | |
| import random | |
| import time | |
| import os | |
| import json | |
| import pathlib | |
| import textwrap | |
| import google.generativeai as genai | |
| from IPython.display import display | |
| from IPython.display import Markdown | |
| import urllib | |
| import warnings | |
| from pathlib import Path as p | |
| from pprint import pprint | |
| from langchain_community.document_loaders import TextLoader | |
| import pandas as pd | |
| from langchain import PromptTemplate | |
| from langchain.chains.question_answering import load_qa_chain | |
| from langchain.document_loaders import PyPDFLoader | |
| from langchain.text_splitter import RecursiveCharacterTextSplitter | |
| from langchain.vectorstores import Chroma | |
| from langchain.chains import RetrievalQA | |
| from langchain_core.documents import Document | |
| from langchain_google_genai import ChatGoogleGenerativeAI | |
| from langchain_google_genai import GoogleGenerativeAIEmbeddings | |
| from langchain import PromptTemplate, LLMChain | |
| warnings.filterwarnings("ignore") | |
| # restart python kernal if issues with langchain import. | |
| GOOGLE_API_KEY='AIzaSyCKmpSRVR3J_uJPvdNIfON4NYNwTTRI3oc' | |
| def to_markdown(text): | |
| text = text.replace('•', ' *') | |
| return Markdown(textwrap.indent(text, '> ', predicate=lambda _: True)) | |
| import requests | |
| from urllib3.exceptions import InsecureRequestWarning | |
| requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning) | |
| model = ChatGoogleGenerativeAI(model="gemini-pro",google_api_key=GOOGLE_API_KEY,temperature=0.3,convert_system_message_to_human=True) | |
| embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001",google_api_key=GOOGLE_API_KEY) | |
| text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=10) | |
| location = os.getcwd() | |
| # persist_directory='/Users/vp317n/Documents/JUPYTER NOTEBOOK/db/' | |
| # file_path = "100_names.txt" | |
| # with open(file_path, 'r') as file: | |
| # file_content = file.read() | |
| # names_split = text_splitter.split_text(file_content) | |
| # service_names = Chroma.from_texts(names_split, embeddings,persist_directory=location) | |
| import requests | |
| from urllib3.exceptions import InsecureRequestWarning | |
| requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning) | |
| def fetch_data_from_api(url): | |
| try: | |
| response = requests.get(url,verify=False) | |
| # requests.get('https://github.com', verify='/path/to/certfile') | |
| # Check if the request was successful (status code 200) | |
| if response.status_code == 200: | |
| # Parse JSON data | |
| data = response.json() | |
| return data | |
| else: | |
| print(f"Failed to fetch data. Status code: {response.status_code}") | |
| return None | |
| except requests.RequestException as e: | |
| print(f"Error fetchinag data: {e}") | |
| return None | |
| print("\n\n\Actual answer:\n") | |
| service_names = Chroma(persist_directory=location,embedding_function=embeddings) | |
| service_names_chain = RetrievalQA.from_chain_type( | |
| model, | |
| retriever=service_names.as_retriever(search_kwargs={"k":5}), | |
| return_source_documents=True | |
| ) | |
| # question = "Get the names of the monitors present for the service : serveng-service-eng-order-access-gql-federation" | |
| # question = "Tell me when was the service : serveng-service-eng-order-access-gql-federation last refreshed" | |
| # question = "Tell me all the details about the service : serveng-service-eng-order-access-gql-federation" | |
| def res(p1): | |
| question = p1 | |
| question_internal_1 = "Just identify the name of the service from the question which starts at : "+question+" and ends here" | |
| result = service_names_chain({"query": question_internal_1}) | |
| name = result["result"] | |
| # print(name) | |
| #Backstage | |
| api_url_backstage = "https://kube-backstage-backend.service.intraiad1.consul.csnzoo.com/api/catalog/entities/by-name/component/default/"+name | |
| response_backstage = str(fetch_data_from_api(api_url_backstage)) | |
| page_content_backstage="This document contains basic information about the sevice "+name+", which is present in the form of json.\nThe json information starts below: \n" | |
| page_content_backstage=page_content_backstage+response_backstage | |
| #monitors | |
| api_url_monitors = "https://kube-backstage-backend.service.intraiad1.consul.csnzoo.com/api/proxy/datadog/api/v1/monitor/search?query=tag:service:"+name | |
| response_monitors = fetch_data_from_api(api_url_monitors) | |
| monitors = response_monitors['monitors'] | |
| extracted_data = [] | |
| for monitor in monitors: | |
| monitor_info = { | |
| 'id': monitor['id'], | |
| 'name': monitor['name'], | |
| 'type': monitor['type'], | |
| 'status': monitor['status'], | |
| 'creator_name': monitor['creator']['name'], | |
| 'notifications': [notification['name'] for notification in monitor['notifications']], | |
| 'last_triggered_ts': monitor['last_triggered_ts'], | |
| 'tags': monitor['tags'], | |
| 'metrics':monitor['metrics'] | |
| } | |
| extracted_data.append(monitor_info) | |
| page_content_monitors = "This document contains information about the monitors of a service named "+name+". The details of each monitor is mentioned below.\n" | |
| for monitor_info in extracted_data: | |
| monitor_string = ( | |
| f"ID of this monitor is {monitor_info['id']}, " | |
| f"Name of this monitor is {monitor_info['name']}, " | |
| f"and it's type is {monitor_info['type']}, " | |
| f"whose Status is {monitor_info['status']}, " | |
| f"It is created by {monitor_info['creator_name']}." | |
| f"Notifications for this monitor are {', '.join(monitor_info['notifications'])}, " | |
| f"The timestamp when this monitor was last triggered is {monitor_info['last_triggered_ts']}, " | |
| f"Metrics for this monitor are {monitor_info['metrics']}, " | |
| f"Tags for this monitor are {', '.join(monitor_info['tags'])}\n" | |
| ) | |
| page_content_monitors=page_content_monitors+monitor_string | |
| doc1 = Document(page_content=page_content_backstage, metadata={"Metadata: Contains general information about the service, such as name, description, tags, lifecycle, etc.":1, | |
| "Consumes APIs: Lists the APIs consumed by the service.":2,"Provides APIs: Lists the APIs provided by the service.":3}) | |
| doc2 = Document(page_content=page_content_monitors, metadata={"ID: Unique identifier for the monitor":1,"Name: Name or description of the monitor.":2,"Type: Type of monitor (e.g., query alert).":3,"Status: Current status of the monitor (e.g., OK).":4, | |
| "Creator: Name of the person who created the monitor.":5,"Notifications: Methods of notification configured for the monitor (e.g., Slack, PagerDuty).":6, | |
| "Last Triggered Timestamp: Timestamp indicating when the monitor was last triggered.":7,"Metrics: Metrics being monitored by the alert (e.g., 'trace.servlet.request.errors').":8, | |
| "Tags: Relevant tags associated with the monitor (e.g., env:prod, incident_metrics_service_sync:true, service:serveng-service-eng-order-access-gql-federation)":9}) | |
| template = """Use the following pieces of context to answer the question at the end. If you don't know the answer, just say that you don't know, don't try to make up an answer. Keep the answer as concise as possible. Always say "thanks for asking!" at the end of the answer. | |
| Answer the question in a human readable way. | |
| {context} | |
| Question: {question} | |
| Helpful Answer:""" | |
| QA_CHAIN_PROMPT = PromptTemplate.from_template(template)# Run chain | |
| db = Chroma(embedding_function=embeddings) | |
| db.add_documents([doc1,doc2]) | |
| info_chain = RetrievalQA.from_chain_type( | |
| model, | |
| retriever=db.as_retriever(search_kwargs={"k":5}), | |
| return_source_documents=True, | |
| chain_type_kwargs={"prompt": QA_CHAIN_PROMPT} | |
| ) | |
| result = info_chain({"query": question}) | |
| # print(result["result"]) | |
| return result["result"] | |