from fastapi import FastAPI,HTTPException, Request from models.incident_connection import IncidentConnection from schema.incident_schema import IncidenSchema from cachetools import cached, TTLCache import matplotlib.pyplot as plt import io from fastapi.responses import StreamingResponse import uuid from langchain.chains import LLMChain from langchain_openai import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain.prompts import PromptTemplate from langchain.chains import create_qa_with_sources_chain from langchain.chains.combine_documents.stuff import StuffDocumentsChain from langchain.chains import ConversationalRetrievalChain from fastapi import FastAPI, HTTPException # from langchain_community.vectorstores import Milvus from langchain_core.embeddings import Embeddings from models.MessageRequest import MessageRequest from models.TicketRequest import TicketRequest import requests app = FastAPI() conn = IncidentConnection() cache = TTLCache(maxsize=128, ttl=300) # Declaración de las variables url y token url = "https://hawa.proactivanet.com/proactivanet/api/table/data?url=https%3A%2F%2Fhawa.proactivanet.com%2Fproactivanet%2Fservicedesk%2Fincidents%2FallIncidents%2FallIncidents.paw%3Fsource%3DviewAllGlobalIncidents%26pawNodeExtra%3D%26pawParentNode%3Dnode%26pawParentData%3D%26pawNode%3DtreeAllTickets%26pawData%3DshowArchivedButton%253D1" token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJqYW5hdmFycm9AaGF3YXNvbHV0aW9ucy5jb20iLCJvdnIiOiJmYWxzZSIsImF1dCI6IjAiLCJuYmYiOjE3MDk2NzY5NDYsImV4cCI6MTc0MTIxMjk0NiwiaWF0IjoxNzA5Njc2OTQ2LCJpc3MiOiJwcm9hY3RpdmFuZXQiLCJhdWQiOiJhcGkifQ.UVvUmjo0yO3R-PjBimNsvHeIHjbbdO8v6Ou6_TjlcGw" cache = TTLCache(maxsize=128, ttl=300) # class MessageRequest(BaseModel): # message: str # endpint de ruta base para obtener la data sin definir /api/nombre de endpoint @app.get("/") @cached(cache) def root(): items = [] for data in conn.read_all(): dictionary = {} dictionary["Ver Registro"] = data[0] dictionary["codigo"] = data[1] dictionary["fecha de registro"] = data[2] items.append(dictionary) return items # endpoint para devolver los tickets caducados en base al sla @app.get("/api/promed_R_by_tecnico/{technician_name}") @cached(cache) def getAll(technician_name: str): items = [] for data in conn.read_by_technician(technician_name): dictionary = { "Categoria": data[0], "Tipo": data[1], "servicio": data[2], "cliente": data[3], "tecnico": data[4], "Promedio tiempo resolucion": data[5], } items.append(dictionary) return items # endpoint para devolver los tickets en base a la categoria @app.get("/api/categoria/{categoria}") @cached(cache) def getAll(categoria: str): return conn.read_by_category(categoria) # endpoint clasificador 1 Hawa @app.get("/api/Url_provider_categoria") async def getAll(): data = conn.get_data_in_url(url,token) if "dataTable" in data: items = [] for item in data["dataTable"]: dictionary = { "Categoria": item["code"], # Ejemplo: "code" es la clave para "Categoria" "Tipo": item["incidentTitle"], # Ejemplo: "incidentTitle" es la clave para "Tipo" "servicio": item["status"], # Ejemplo: "status" es la clave para "servicio" "cliente": item["padCustomersName"], # Ejemplo: "padCustomersName" es la clave para "cliente" "tecnico": item["PawSvcAuthUsersResponsFullName"], # Ejemplo: "PawSvcAuthUsersResponsFullName" es la clave para "tecnico" "Promedio tiempo resolucion": item["formattedResponseTime"] # Ejemplo: "formattedResponseTime" es la clave para "Promedio tiempo resolucion" } items.append(dictionary) return items else: return [] # endpoint para devolver los tickets caducados en base al sla @app.get("/api/ticket_sla/{sla_prefix}") @cached(cache) def getAll(sla_prefix: str): items = [] for data in conn.read_all_tickets_by_sla(sla_prefix): dictionary = { "Sla": data[0], "Total tickets": data[1], "Tickets Caducados": data[2], "Tickets no caducados": data[3], "Porcentaje caducados": data[4], "Porcentaje no caducados": data[5], } items.append(dictionary) return items # endpoint para devolver los tickets caducados en base al sla @app.get("/api/ticket_sla_no_params") @cached(cache) def getAll(): items = [] for data in conn.read_all_tickets_noparams_sla(): dictionary = { "Sla": data[0], "Total tickets": data[1], "Tickets Caducados": data[2], "Tickets no caducados": data[3], "Porcentaje caducados": data[4], "Porcentaje no caducados": data[5], } items.append(dictionary) return items # endpoint para devolver los tickets caducados en base al sla @app.get("/api/ticket_sla_no_params_c_t_s") @cached(cache) def getAll(): items = [] for data in conn.read_all_tickets_no_params_sla_c_t_s(): dictionary = { "Sla": data[0], "Categoria": data[1], "Tipo": data[2], "Servicio": data[3], "Total tickets": data[4], "Tickets caducados": data[5], "Tickets no caducados": data[6], "Porcentaje caducados": data[7], "porcentaje caducados": data[8], } items.append(dictionary) return items # endpoint con tres parametros año1 año2 y mes @app.get("/api/tickets_repetitivos_Y_Comparativa_Anual/{anio1}/{anio2}/{mes}") @cached(cache) def getAll(anio1: str, anio2: str, mes: str): items = [] for data in conn.read_ticket_rep_and_comp_year(anio1, anio2, mes): dictionary = { "categoria": data[0], "tipo": data[1], "incidentes_2019": data[2], "incidentes_2020": data[3], } items.append(dictionary) return items # endpoint sin parametros por caducados y no caducados @app.get("/api/tickets_cad_y_nocad_by_years") @cached(cache) def getAll(): items = [] for data in conn.read_ticket_cad_y_nocad_year(): dictionary = { "Año": data[0], "Categoria": data[1], "Servicio": data[2], "Tipo": data[3], "Total incidentes caducados": data[4], "Total incidentes no caducados": data[5], "Total incidentes": data[6], } items.append(dictionary) return items # endpoint para devolver el grupo de soporte con mas resoluciones @cached(cache) @app.get("/api/total_incidents_by_group") def get_total_incidents_by_group(): items = [] for data in conn.read_total_incidents_by_group(): dictionary = { "grupo": data[0], "total_incidentes": data[1], "year_with_most_incidents": data[2], } items.append(dictionary) return items # endpoint para devolver la tendencia de tickets en base al tipo @app.get("/api/tendencia_tipo/{anio1}/{anio2}") @cached(cache) def getAll(anio1: str, anio2: str): items = [] for data in conn.read_tendencia_by_tipo(anio1, anio2): dictionary = {} dictionary["tipo"] = data[0] dictionary["titulo"] = data[1] dictionary["cantidad_tickets_2019"] = data[2] dictionary["cantidad_tickets_2020"] = data[3] items.append(dictionary) return items @app.get("/api/tendencia_tipo_imagen/{anio1}/{anio2}") def getAll(anio1: str, anio2: str): items = [] for data in conn.read_tendencia_by_tipo(anio1, anio2): dictionary = {} dictionary["tipo"] = data[0] dictionary["titulo"] = data[1] dictionary["cantidad_tickets_2019"] = data[2] dictionary["cantidad_tickets_2020"] = data[3] items.append(dictionary) # Crear gráfico tipos = [item['tipo'] for item in items] tickets_2019 = [item['cantidad_tickets_2019'] for item in items] tickets_2020 = [item['cantidad_tickets_2020'] for item in items] plt.figure(figsize=(10, 5)) plt.bar(tipos, tickets_2019, color='blue', alpha=0.7, label='2019') plt.bar(tipos, tickets_2020, color='red', alpha=0.7, label='2020', bottom=tickets_2019) plt.xlabel('Tipo') plt.ylabel('Cantidad de Tickets') plt.title('Tendencia por Tipo') plt.legend() # Guardar el gráfico en un objeto BytesIO buf = io.BytesIO() plt.savefig(buf, format='png') buf.seek(0) # Devolver el gráfico como una imagen return StreamingResponse(buf, media_type="image/png") # endpoint para devolver tickets y porcentajes por sla @app.get("/api/sla_cads_nocads_percents") @cached(cache) # Decorador para almacenamiento en caché def getAll(): items = [] for data in conn.read_sla_cads_nocads_percents(): dictionary = {} dictionary["SLA"] = data[0] dictionary["Total tickets"] = data[1] dictionary["Tickets Caducados"] = data[2] dictionary["Tickets no caducados"] = data[3] dictionary["Porcentaje caducados"] = data[4] dictionary["Porcentaje no caducados"] = data[5] items.append(dictionary) return items # endpoint para devolver la tendencia de tickets en base al tipo categoria y por todos los años @app.get("/api/tendencias_by_c_t_all_anios") @cached(cache) # Decorador para almacenamiento en caché def getAll(): items = [] for data in conn.read_tendencias_tickets_all_anios(): dictionary = {} dictionary["Categoria"] = data[0] dictionary["Tipo"] = data[1] dictionary["Cantidad tickets 2019"] = data[2] dictionary["Cantidad tickets 2020"] = data[3] dictionary["Cantidad tickets 2021"] = data[4] dictionary["Cantidad tickets 2022"] = data[5] dictionary["Cantidad tickets 2023"] = data[6] dictionary["Cantidad tickets 2024"] = data[7] items.append(dictionary) return items # endpoint para devolver la data de las calificaciones de las empresas @app.get("/api/califications_Emp") @cached(cache) def getall(): items = [] for data in conn.read_all_califications(): dictionary = {} dictionary["Empresa"] = data[0] dictionary["total_calificaciones"] = data[1] dictionary["Muy Satisfecho"] = data[2] dictionary["Satisfecho"] = data[3] dictionary["Neutral"] = data[4] dictionary["Insatisfecho"] = data[5] dictionary["Muy Insatisfecho"] = data[6] dictionary["porcentaje_muy_satisfecho"] = data[7] dictionary["porcentaje_satisfecho"] = data[8] dictionary["porcentaje_neutral"] = data[9] dictionary["porcentaje_insatisfecho"] = data[10] dictionary["porcentaje_muy_insatisfecho"] = data[11] items.append(dictionary) return items @app.post("/extraer_decision") async def extractdesition(request: TicketRequest): try: # Configuración del modelo GPT llm = ChatOpenAI(temperature=0.5, model="gpt-4", openai_api_key="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") # Obtener la respuesta del modelo directamente response = await llm.apredict( text=f"Del siguiente texto: {request.message}, extrae la decisión de si el usuario quiere crear un ticket. Luego, pregunta por el motivo del ticket y pide una descripción." ) if "crear un ticket" in response.lower(): return {"response": "Claro, ingresa el motivo de tu requerimiento."} # Aquí se espera que el usuario haya proporcionado el motivo elif "motivo:" in request.title.lower(): motivo = request.title.lower().replace("motivo:", "").strip() return {"response": "Describe tu caso.", "motivo": motivo} # Aquí se espera que el usuario haya proporcionado la descripción elif "descripcion:" in request.description.lower(): descripcion = request.description.lower().replace("descripcion:", "").strip() motivo = request.headers.get("motivo", "") # Obtener el motivo de las cabeceras ticket_data = TicketRequest( Title=motivo, Description=descripcion, PanUsers_idSource="0f2afbe5-7f66-4a47-ae46-20373d64038c" # Id de usuario fijo para el ejemplo ) # Enviar la solicitud para crear el ticket response = requests.post( "https://hawasolutions.proactivanet.com/panet/api/Incidents", json=ticket_data.dict(), headers={"Content-Type": "application/json"} ) if response.status_code == 201: return {"response": {"titulo": motivo, "descripcion": descripcion}} else: raise HTTPException(status_code=response.status_code, detail="Error al crear el ticket") else: return {"response": "No se ha reconocido una solicitud válida. Por favor, indica si deseas crear un ticket, proporciona un motivo o una descripción."} except Exception as e: print(f"Error: {str(e)}") raise HTTPException(status_code=500, detail="Internal Server Error") @app.post("/consulta_api") async def apiConsultaTWS(request: MessageRequest): try: # Configuración del modelo GPT llm = ChatOpenAI(temperature=0.5, model="gpt-4o-mini", openai_api_key="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") # Definición del prompt para condensar la pregunta condense_question_prompt_template = """Toma la última pregunta del historial de chat y únela a la pregunta de entrada de seguimiento y crea una pregunta independiente. Historial de Chat: {chat_history} Entrada de seguimiento: {question} Pregunta independiente:""" condense_question_prompt = PromptTemplate( template=condense_question_prompt_template, input_variables=["chat_history", "question"] ) condense_question_chain = LLMChain( llm=llm, prompt=condense_question_prompt, ) memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True,k=3) # Definición del prompt para interpretar la respuesta qa_prompt_template = """ Actúa como un asistente de soporte técnico con Conocimeintoes en ITIL, COBIT, ISO 27001, CISCO que permite al usuario interactuar y obtener una solución de un problema, debes ser preciso en la resolución del problema que el usuario plantea, debe enfocarse solo en lo que se refiere a soporte técnico de mesa de servicios, mantenimiento de equipos tecnológicos, si el usuario hace preguntas que no tengan nada que ver con la mesa de servicios simplemente no respondas.. Historial de Chat: {chat_history} Entrada de seguimiento: {question} """ qa_prompt = PromptTemplate( template=qa_prompt_template, input_variables=["chat_history", "question"] ) qa_chain = LLMChain(llm=llm, prompt=qa_prompt, memory=memory) # Condensar la pregunta condensed_question = await condense_question_chain.apredict( chat_history=memory.load_memory_variables({})["chat_history"], question=request.message ) # Obtener la respuesta del modelo response = await qa_chain.apredict( chat_history=memory.load_memory_variables({})["chat_history"], question=condensed_question ) sentiment_analysis = await replace_product_codes(request.message) # final_response = f'{response + sentiment_analysis}' # sentiment_analysis = await replace_product_codes(final_response) # Actualizar la memoria con la nueva interacción memory.save_context({"input": request.message}, {"output": response}) return {"answer": response, "sentiment": sentiment_analysis} except Exception as e: print(f"Error: {str(e)}") raise HTTPException(status_code=500, detail="Internal Server Error") # @app.post("/consulta_resumen") async def replace_product_codes(text:str): # prompt = f"De la siguiente conversación:{text} extrae en formato json el resumen y el sentimiento de la conversación." try: max_tokens = 300 llm = ChatOpenAI( model = "gpt-4o", max_tokens=max_tokens, temperature=0.0, api_key="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" ) messages = [ ( "system", "eres un generador json de resumen y sentimiento de conversaciones", ), ( "human", "De la siguiente conversación:"+text+" extrae en formato json el resumen y el sentimiento de la conversación. si no te llega nada en "+text+" no respondas", ), ] ai_msg = llm.invoke(messages) # response_json = json.loads(ai_msg.content) # return_sentiment_only=True # if return_sentiment_only: # sentimiento = response_json.get("sentimiento", {}) # sentimiento_str = json.dumps(sentimiento) # return sentimiento_str return ai_msg.content except Exception as e: print(f"Error: {str(e)}") raise HTTPException(status_code=500, detail="Internal Server Error") @app.post("/consulta_api_bgr") async def apiConsultaBGR(request: MessageRequest): try: # Configuración del modelo GPT llm = ChatOpenAI(temperature=0.5, model="gpt-4o-mini", openai_api_key="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") # Definición del prompt para condensar la pregunta condense_question_prompt_template = """Toma la última pregunta del historial de chat y únela a la pregunta de entrada de seguimiento y crea una pregunta independiente. Historial de Chat: {chat_history} Entrada de seguimiento: {question} Pregunta independiente:""" condense_question_prompt = PromptTemplate( template=condense_question_prompt_template, input_variables=["chat_history", "question"] ) print(f"Tipo de request: {type(request)}") print(f"Contenido del request: {request}") # Verificar el valor de request.message print(f"Mensaje recibido: {request.message}") condense_question_chain = LLMChain( llm=llm, prompt=condense_question_prompt, ) memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True,k=3) # Definición del prompt para interpretar la respuesta qa_prompt_template = """ Eres LHIA es un asistente de soporte técnico del Banco General Rumiñahui desarrollado por TWS . te enfocas exclusivamente en resolver problemas relacionados con la mesa de servicios, ITIL, COBIT, ISO 27001, y tecnologías CISCO, brindando soluciones precisas y eficientes en mantenimiento de equipos tecnológicos y que permite al usuario interactuar y obtener una solución de un problema. **Reglas y restricciones**: 1. LHIA solo responde a consultas directamente relacionadas con soporte técnico de mesa de servicios y mantenimiento de equipos tecnológicos. 2. No proporciona respuestas ni sugerencias sobre temas no relacionados con soporte técnico, ITIL, COBIT, ISO 27001, o tecnologías CISCO. 3. Mantén siempre la precisión en las respuestas, enfocándote en la resolución del problema técnico planteado. 4. Rechaza educadamente cualquier solicitud fuera del alcance del soporte técnico, aclarando que solo se manejan problemas relacionados con soporte técnico y mantenimiento de equipos. 5. Siempre mantén un tono profesional y cortés, siguiendo los estándares del Banco General Rumiñahui. **Historial de Chat**: {chat_history} **Entrada de seguimiento**: {question} """ qa_prompt = PromptTemplate( template=qa_prompt_template, input_variables=["chat_history", "question"] ) qa_chain = LLMChain(llm=llm, prompt=qa_prompt, memory=memory) memory_variables = memory.load_memory_variables({}) chat_history = memory_variables.get("chat_history", "") # Condensar la pregunta condensed_question = await condense_question_chain.apredict( chat_history=chat_history, question=request.message ) # Obtener la respuesta del modelo response = await qa_chain.apredict( chat_history=memory.load_memory_variables({})["chat_history"], question=condensed_question ) sentiment_analysis = await replace_product_codes(request.message) # final_response = f'{response + sentiment_analysis}' # sentiment_analysis = await replace_product_codes(final_response) # Actualizar la memoria con la nueva interacción memory.save_context({"input": request.message}, {"output": response}) return {"answer": response, "sentiment": sentiment_analysis} except Exception as e: print(f"Error: {str(e)}") raise HTTPException(status_code=500, detail="Internal Server Error")