Problem Statement: Declaración de problemas: Con Copilot integrado en las aplicaciones de la organización, encontrar datos raramente utilizados de archivos, SharePoint y otras fuentes accesibles se ha vuelto increíblemente fácil. He estado confiando mucho en esta capacidad Gen AI. Un día, necesitaba una visión resumida de todas las características (los entregables del equipo para un cuarto en el marco ágil) y sus estados en los que el equipo está trabajando. Desafortunadamente, Copilot negó leer los datos de la página Confluence, que es ideal y esperado. Muchas organizaciones, proyectos y actualizaciones de programas se almacenan en las páginas Confluence. Obtener una visión general de los objetivos del equipo, los entregables, los riesgos del proyecto y el estado puede ser demorado para un líder o una persona que maneja múltiples programas. Pensé, ¿por qué no tener un asistente Solution: AI Agentic Assistant Powered by Streamlit + Semantic Kernel Solución: AI Agentic Assistant Powered by Streamlit + Semantic Kernel La introducción de Agentic AI fue un salvador para mí, y decidí elegir este marco como una solución. Sin embargo, hubo desafíos: ¿qué marco debe usarse, y ¿hay un código abierto disponible? ¿Cuánto costaría la plataforma gestionada? Finalmente, con toda la investigación, decidí ir con código abierto y usar la pila de tecnología a continuación para construir un asistente de IA ligero usando: Limpieza para el extremo frontal, Núcleo semántico para la gestión rápida y la cadena, Azure OpenAI para procesamiento de idiomas naturales Playwright para rascado web seguro y dinámico de páginas de Confluence. Lo que hace Esta herramienta permitirá a los gerentes y líderes del programa: Select a from a dropdown, team or program name Automatically fetch the associated Confluence page URL, Scrape key content sections from that page (like Features, Epics, Dependencies, Risks, Team Members), Ask questions like “What are the team Q4 deliverables?” or “Summarize the features based on status,” etc., Display answers as summarized text. How it works - Pseudo Code Cómo funciona - PseudoCódigo Step 1. Confluence Page LookUp. En lugar de pegar URL manualmente, cada nombre de equipo se mapea a su URL Confluence usando un diccionario. Cuando un usuario selecciona "Team A" en el panel derecho, el backend recoge automáticamente la URL asociada y desencadena el rascado. team_to_url_map = { "Team A": "https://confluence.company.com/display/TEAM_A", "Team B": "https://confluence.company.com/display/TEAM_B", ... } team_to_url_map = { "Team A": "https://confluence.company.com/display/TEAM_A", "Team B": "https://confluence.company.com/display/TEAM_B", ... } Step 2. Web Scraping via Playwright Finalmente, terminé usando Playwright para el rascado basado en navegador sin cabeza, que nos ayuda a cargar contenido dinámico y manejar el inicio de sesión: Fracaso del enfoque: [ ]Utilizando la biblioteca de solicitudes de Python, obtenga los datos de Confluencia utilizando la API. El mecanismo de autenticación no fue exitoso. De lo contrario, sería una excelente manera de obtener datos de página de Confluencia. [ ]Utilizando la biblioteca BeautifySoup de Python. Fue excluido debido al contenido dinámico. I ended up with Python Playwright. The SSO layer had challenges, but finally, it worked after downloading the HTML state JSON and reusing it. [ ] @kernel_function(description="Scrape and return text content from a Confluence page.") async def get_confluence_page_content(self, team_name: Annotated[str, "Nombre del equipo Ágil"]) -> Annotated[str, "Retorna contenido de texto extraído de la página"): @kernel_function(description="Scrape and return text content from a Confluence page.") async def get_confluence_page_content(self, team_name: Annotated[str, "Name of the Agile team"]) -> Annotated[str, "Returns extracted text content from the page"]: Step 3. Define Client, Agent, and Prompt Engineering with Semantic Kernel. La herramienta está destinada a ser un Program Management Enabler. Las Instrucciones de Agente están redactadas para utilizar el contenido rasgado y producir un resultado adecuado para las preguntas PM. Las instrucciones ayudarán a obtener la salida como un resumen de texto o en un formato de gráfico. Además, he definido el cliente como un agente de IA como un plugin. AGENT_INSTRUCCIONES = “” “” cliente = OpenAI (<Local Open Source LLM>) chat_completion_service = OpenAIChatCompletion(ai_model_id="<>", async_client es el cliente agente = ChatCompletionAgent( servicio=chat_completion_service, plugins=[ ConfluencePlugin() ], nombre="ConfluenceAgent", instrucciones=AGENT_INSTRUCTIONS ) AGENT_INSTRUCTIONS = “““ “““ client = OpenAI(<Local open source LLM>) chat_completion_service = OpenAIChatCompletion(ai_model_id="<>", async_client=client ) agent = ChatCompletionAgent( service=chat_completion_service, plugins=[ ConfluencePlugin() ], name="ConfluenceAgent", instructions=AGENT_INSTRUCTIONS ) Step 4. Decide on User Input to process the question with or without the tool. Decidí añadir un cliente LLM adicional para comprobar si la entrada del usuario es relevante para la Gestión de Programas o no. completion = await client.chat.completions.create( model="gpt-4o", messages=[ {"role": "sistema", "contenido": "Usted es un Juez del contenido de la entrada del usuario. Analizando la entrada del usuario. Si se le pide raspar la Página de COnfluence interna para un equipo, entonces está relacionada con la Gestión de Programas. Si no está relacionada con la Gestión de Programas, proporcione la respuesta pero añade 'Falseḳ' a la respuesta. Si está relacionada con la Gestión de Programas, añade 'Trueḳ' a la respuesta."}, {"role": "usuario", "contenido": user_input} ], temperatura=0.5 ) completion = await client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": "You are a Judge of the content of user input. Anlyze the user's input. If it asking to scrap internal COnfluence Page for a team then it is related to Program Management. If it is not related to Program Management, provide the reply but add 'False|' to the response. If it is related to Program Management, add 'True|' to the response."}, {"role": "user", "content": user_input} ], temperature=0.5 ) Step 5. The final step is to produce the result. Here is the entire code. Aquí está el código completo. He borrado mis detalles específicos del proyecto. Necesitamos almacenar el state.json primero para usarlo en el código importación jsen importar os importar asyncio importar pandas como pd importar streamlit como st de escribir importar Anotado de dotenv importar load_dotenv de openai importar AsyncAzureOpenAI de playwright.async_api importar async_playwright de bs4 importar BeautifulSoup de semantic_kernel.funciones importar kernel_función de importar Importar anotado re importar matplotlibot.pyplot como plt de semantic_kernel.agentes importar ChatCompletionAgent de semantic_kernel.connectors.ai.open_ai importar OpenAIChatCompletion de semantic_kernel.contents importar FunCallContent, FunctionResultContent TEAM_URL_MAPPING = { "Team 1": "Confluence URL for Team 1", "Team 2": "Confluence URL for Team 2", "Team 3": "Confluence URL for Team 3", "Team 4": "Confluence URL for Team 4", "Team 5": "Confluence URL for Team 5", "Team 6": "Confluence URL for Team 6" } # ---- Definición de plugins ---- #Chart de barras con tamaño fijo def plot_bar_chart(df): status_counts = df["status"].value_counts() fig, ax = plt.subplots(figsize=(1.5, 1)) # ancho, altura en pulgadas ax.bar(status_counts.index, status_counts.values, color="#4CAF50") ax.set_title("Features by Status") ax.set_ylabel("Count") # Cambiar tick color ax.tick_params(axis='x', colors='blue', labelrotation=90) # x-ticks en azul, rotated axtick_params(axis='y', colors='green') # y-ticks en verde st.pyplot(fig) def extract_json_from_response(text): # Use regex para encontrar el primer array JSON en el texto match = re.search(r"(\[\s*{.*\s*\])", texto, re.DOTALL) si match: return match.group(1) return No clase ConfluenciaPlugin: def init(self): self.default_confluence_url = "<>" load_dotenv() @kernel_function(description="Scrape and return text content from a Confluence page.") async def get_confluence_page_content( self, team_name: Annotated\[str, "Name of the Agile team"\] ) -> Annotated\[str, "Returns extracted text content from the page"\]: print(f"Attempting to scrape Confluence page for team: '{team_name}'") # Added for debugging target_url = TEAM_URL_MAPPING.get(team_name) if not target_url: print(f"Failed to find URL for team: '{team_name}' in TEAM_URL_MAPPING.") # Added for debugging return f"❌ No Confluence URL mapped for team '{team_name}'" async with async_playwright() as p: browser = await p.chromium.launch() context = await browser.new_context(storage_state="state.json") page = await context.new_page() pages_to_scrape = \[target_url\] # Loop through each page URL and scrape the content for page_url in pages_to_scrape: await page.goto(page_url) await asyncio.sleep(30) # Wait for the page to load await page.wait_for_selector('div.refresh-module-id, table.some-jira-table') html = await page.content() soup = BeautifulSoup(html, "html.parser") body_div = soup.find("div", class_="wiki-content") or soup.body if not body_div: return "❌ Could not find content on the Confluence page." # Process the scraped content (example: extract headings) headings = soup.find_all('h2') text = body_div.get_text(separator="\\n", strip=True) return text\[:4000\] # Truncate if needed to stay within token limits await browser.close() @kernel_function(description="Summarize and structure scraped Confluence content into JSON.") async def summarize_confluence_data( self, raw_text: Annotated\[str, "Raw text scraped from the Confluence page"\], output_style: Annotated\[str, "Output style, either 'bullet' or 'json'"\] = "json" # Default to 'json' ) -> Annotated\[str, "Returns structured summary in JSON format"\]: prompt = f""" You are a Program Management Data Extractor. Your job is to analyze the following Confluence content and produce structured machine-readable output. Confluence Content: {raw_text} Instructions: - If output_style is 'bullet', return bullet points summary. - If output_style is 'json', return only valid JSON array by removing un printable characters and spaces from beginning and end. - DO NOT write explanations. - DO NOT suggest code snippets. - DO NOT wrap JSON inside triple backticks \`\`\`json - Output ONLY the pure JSON array or bullet points list. Output_style: {output_style} """ # Call OpenAI again completion = await client.chat.completions.create( model="gpt-4o", messages=\[ {"role": "system", "content": "You are a helpful Program Management Data Extractor."}, {"role": "user", "content": prompt} \], temperature=0.1 ) structured_json = completion.choices\[0\].message.content.strip() return structured_json # ---- Credenciales de API de carga ---- cliente load_dotenv() = AsyncAzureOpenAI( azure_endpoint="<>", api_key=os.getenv("AZURE_API_KEY"), api_version='<>' ) chat_completion_service = OpenAIChatCompletion( ai_model_id="<>", async_client=client ) AGENT_INSTRUCTIONS = """Usted es un útil Agente de Gestión de Programas AI que puede ayudar a extraer información clave como miembro del equipo, características, Epics de una página de confluencia. Importante: Cuando los usuarios especifiquen una página de equipo, solo extraen las características y episodios de ese equipo. Cuando empiece la conversación, presentate con este mensaje: "¡Hola! soy tu asistente de PM. puedo ayudarte a obtener el estado de Funciones y Epics. Siempre llame primero 'get_confluence_page_content' para raspar la página Confluence. Si el mensaje del usuario comienza con "Team: {team_name}.", utilice ese {team_name} para el argumento 'team_name'. Por ejemplo, si la entrada es "Team: Raptor. ¿Cuáles son las últimas características?", el 'team_name' es "Raptor". 2. Si el usuario solicita un resumen, proporcione una lista de puntos de bala. 3. Si el usuario solicita un array JSON o un gráfico o un plano. Luego llame inmediatamente 'summarize_confluence_data' utilizando el contenido rasgado. 4. Basándose en el estilo de salida solicitado por el usuario, devuelva una matriz JSON o puntos de bala. 5. Si el usuario no especifica un estilo de salida, devuelva por defecto un punto de bala. 6. Si el usuario solicita un array JSON, Instrucciones: - Si output_style es 'bullet', devuelve el resumen de los puntos de bala. - Si output_style es 'json', devuelve solo la matriz JSON válida eliminando los caracteres y espacios no imprimibles del comienzo y el final. - NO escriba explicaciones. - NO sugiera fragmentos de código. - NO envuelva JSON dentro de triples backticks ``'json - Salga SÓLO la lista de matriz JSON pura o puntos de bala. ¿Qué equipo está interesado en ayudarle a planificar hoy?" Si mencionan un equipo específico, enfocen sus datos en ese equipo en lugar de sugerir alternativas. """ agente = ChatCompletionAgent( service=chat_completion_service, plugins=[ ConfluencePlugin() ], name="ConfluenceAgent", instrucciones=AGENT_INSTRUCTIONS ) # ---- es la lógica async principal ---- async def stream_response(user_input, thread=None): html_blocks = [] full_response = [] function_calls = [] parsed_json_result = None completion = wait client.chat.completions.create( model="gpt-4o", messages=["role": "sistema", "contenido": "Si usted es un juez del contenido de la entrada del usuario. Analize la función de la entrada del usuario. Si se pide raspar la página interna COnfluence para un equipo, entonces está relacionado con la Gestión del Programa. Si no está relacionado con la Gestión del Programa, proporcione la respuesta, pero añade 'False T' a la respuesta. Si está relacionado con la Gestión del Programa, añ async for response in agent.invoke_stream(messages=user_input, thread=thread): print("Response:", response) thread = response.thread agent_name = response.name for item in list(response.items): if isinstance(item, FunctionCallContent): pass # You can ignore this now elif isinstance(item, FunctionResultContent): if item.name == "summarize_confluence_data": raw_content = item.result extracted_json = extract_json_from_response(raw_content) if extracted_json: try: parsed_json = json.loads(extracted_json) yield parsed_json, thread, function_calls except Exception as e: st.error(f"Failed to parse extracted JSON: {e}") else: full_response.append(raw_content) else: full_response.append(item.result) elif isinstance(item, StreamingTextContent) and item.text: full_response.append(item.text) #print("Full Response:", full_response) # After loop ends, yield final result if parsed_json_result: yield parsed_json_result, thread, function_calls else: yield ''.join(full_response), thread, function_calls # ---- Streamlit UI Setup ---- st.set_page_config(layout="wide") left_col, right_col = st.columns([1, 1]) st.markdown("" <style> html, cuerpo, [class*="css"] { font-size: 12px !important; } </style> """, unsafe_allow_html=True) # ---- Main Streamlit app ---- con left_col: st.title(" Program Management Enabler AI") st.write("Pregúntame sobre diferentes elementos comprometidos del Programa Wiley.!") st.write("Puedo ayudarle a obtener el estado de Funciones y Epics.") if "history" not in st.session_state: st.session_state.history = \[\] if "thread" not in st.session_state: st.session_state.thread = None if "charts" not in st.session_state: st.session_state.charts = \[\] # Each entry: {"df": ..., "title": ..., "question": ...} if "chart_dataframes" not in st.session_state: st.session_state.chart_dataframes = \[\] if st.button("🧹 Clear Chat"): st.session_state.history = \[\] st.session_state.thread = None st.rerun() # Input box at the top user_input = st.chat_input("Ask me about your team's features...") # Example: team_selected = st.session_state.get("selected_team") if st.session_state.get("selected_team") and user_input: user_input = f"Team: {st.session_state.get('selected_team')}. {user_input}" # Preserve chat history when program or team is selected if user_input and not st.session_state.get("selected_team_changed", False): st.session_state.selected_team_changed = False if user_input: df = pd.DataFrame() full_response_holder = {"text": "","df": None} with st.chat_message("assistant"): response_container = st.empty() assistant_text = "" try: chat_index = len(st.session_state.history) response_gen = stream_response(user_input, st.session_state.thread) print("Response generator started",response_gen) async def process_stream(): async for update in response_gen: nonlocal_thread = st.session_state.thread if len(update) == 3: content, nonlocal_thread, function_calls = update full_response_holder\["text"\] = content if isinstance(content, list): data = json.loads(re.sub(r'\[\\x00-\\x1F\\x7F\]', '', content.replace("\`\`\`json", "").replace("\`\`\`",""))) df = pd.DataFrame(data) df.columns = df.columns.str.lower() print("\\n📊 Features Status Chart") st.subheader("📊 Features Status Chart") plot_bar_chart(df) st.subheader("📋 Detailed Features Table") st.dataframe(df) chart_df.columns = chart_df.columns.str.lower() full_response_holder\["df"\] = chart_df elif (re.sub(r'\[\\x00-\\x1F\\x7F\]', '', content.replace("\`\`\`json", "").replace("\`\`\`","").replace(" ",""))\[0\] =="\[" and re.sub(r'\[\\x00-\\x1F\\x7F\]', '', content.replace("\`\`\`json", "").replace("\`\`\`","").replace(" ",""))\[-1\] == "\]"): data = json.loads(re.sub(r'\[\\x00-\\x1F\\x7F\]', '', content.replace("\`\`\`json", "").replace("\`\`\`",""))) df = pd.DataFrame(data) df.columns = df.columns.str.lower() chart_df = pd.DataFrame(data) chart_df.columns = chart_df.columns.str.lower() full_response_holder\["df"\] = chart_df else: if function_calls: st.markdown("\\n".join(function_calls)) flagtext = 'text' st.session_state.thread = nonlocal_thread try: with st.spinner("🤖 AI is thinking..."): flagtext = None # Run the async function to process the stream asyncio.run(process_stream()) # Update history with the assistant's response if full_response_holder\["df"\] is not None and flagtext is None: st.session_state.chart_dataframes.append({ "question": user_input, "data": full_response_holder\["df"\], "type": "chart" }) elif full_response_holder\["text"\].strip(): # Text-type response st.session_state.history.append({ "user": user_input, "assistant": full_response_holder\["text"\], "type": "text" }) flagtext = None except Exception as e: error_msg = f"⚠️ Error: {e}" response_container.markdown(error_msg) if chat_index > 0 and "Error" in full_response_holder\["text"\]: # Remove the last message only if it was an error st.session_state.history.pop(chat_index) # Handle any exceptions that occur during the async call except Exception as e: full_response_holder\["text"\] = f"⚠️ Error: {e}" response_container.markdown(full_response_holder\["text"\]) chat_index = len(st.session_state.history) #for item in st.session_state.history\[:-1\]: for item in reversed(st.session_state.history): if item\["type"\] == "text": with st.chat_message("user"): st.markdown(item\["user"\]) with st.chat_message("assistant"): st.markdown(item\["assistant"\]) con right_col:st.title("Select Wiley Program") team_list = { "Program 1": \["Team 1", "Team 2", "Team 3"\], "Program 2": \["Team 4", "Team 5", "Team 6"\] } selected_program = st.selectbox("Select the Program:", \["No selection"\] + list(team_list.keys()), key="program_selectbox") selected_team = st.selectbox("Select the Agile Team:", \["No selection"\] + team_list.get(selected_program, \[\]), key="team_selectbox") st.session_state\["selected_team"\] = selected_team if selected_team != "No selection" else None if st.button("🧹 Clear All Charts"): st.session_state.chart_dataframes = \[\] chart_idx = 1 #if len(st.session_state.chart_dataframes) == 1: for idx, item in enumerate(st.session_state.chart_dataframes): #for idx, item in enumerate(st.session_state.chart_dataframes): st.markdown(f"\*\*Chart {idx + 1}: {item\['question'\]}\*\*") st.subheader("📊 Features Status Chart") plot_bar_chart(item\["data"\]) st.subheader("📋 Detailed Features Table") st.dataframe(item\["data"\]) chart_idx += 1 importación jsen importar os importar asyncio importar pandas como pd importar streamlit como st de escribir importar Anotado de dotenv importar load_dotenv de openai importar AsyncAzureOpenAI de playwright.async_api importar async_playwright de bs4 importar BeautifulSoup de semantic_kernel.funciones importar kernel_función de importar Importar anotado re importar matplotlibot.pyplot como plt de semantic_kernel.agentes importar ChatCompletionAgent de semantic_kernel.connectors.ai.open_ai importar OpenAIChatCompletion de semantic_kernel.contents importar FunCallContent, FunctionResultContent TEAM_URL_MAPPING = { "Team 1": "Confluence URL for Team 1", "Team 2": "Confluence URL for Team 2", "Team 3": "Confluence URL for Team 3", "Team 4": "Confluence URL for Team 4", "Team 5": "Confluence URL for Team 5", "Team 6": "Confluence URL for Team 6" } # ---- Definición de plugins ---- #Chart de barras con tamaño fijo def plot_bar_chart(df): status_counts = df["status"].value_counts() fig, ax = plt.subplots(figsize=(1.5, 1)) # ancho, altura en pulgadas ax.bar(status_counts.index, status_counts.values, color="#4CAF50") ax.set_title("Features by Status") ax.set_ylabel("Count") # Cambiar tick color ax.tick_params(axis='x', colors='blue', labelrotation=90) # x-ticks en azul, rotated axtick_params(axis='y', colors='green') # y-ticks en verde st.pyplot(fig) def extract_json_from_response(text): # Use regex para encontrar el primer array JSON en el texto match = re.search(r"(\[\s*{.*\s*\])", texto, re.DOTALL) si match: return match.group(1) return No clase ConfluenciaPlugin: def init(self): self.default_confluence_url = "<>" load_dotenv() @kernel_function(description="Scrape and return text content from a Confluence page.") async def get_confluence_page_content( self, team_name: Annotated\[str, "Name of the Agile team"\] ) -> Annotated\[str, "Returns extracted text content from the page"\]: print(f"Attempting to scrape Confluence page for team: '{team_name}'") # Added for debugging target_url = TEAM_URL_MAPPING.get(team_name) if not target_url: print(f"Failed to find URL for team: '{team_name}' in TEAM_URL_MAPPING.") # Added for debugging return f"❌ No Confluence URL mapped for team '{team_name}'" async with async_playwright() as p: browser = await p.chromium.launch() context = await browser.new_context(storage_state="state.json") page = await context.new_page() pages_to_scrape = \[target_url\] # Loop through each page URL and scrape the content for page_url in pages_to_scrape: await page.goto(page_url) await asyncio.sleep(30) # Wait for the page to load await page.wait_for_selector('div.refresh-module-id, table.some-jira-table') html = await page.content() soup = BeautifulSoup(html, "html.parser") body_div = soup.find("div", class_="wiki-content") or soup.body if not body_div: return "❌ Could not find content on the Confluence page." # Process the scraped content (example: extract headings) headings = soup.find_all('h2') text = body_div.get_text(separator="\\n", strip=True) return text\[:4000\] # Truncate if needed to stay within token limits await browser.close() @kernel_function(description="Summarize and structure scraped Confluence content into JSON.") async def summarize_confluence_data( self, raw_text: Annotated\[str, "Raw text scraped from the Confluence page"\], output_style: Annotated\[str, "Output style, either 'bullet' or 'json'"\] = "json" # Default to 'json' ) -> Annotated\[str, "Returns structured summary in JSON format"\]: prompt = f""" You are a Program Management Data Extractor. Your job is to analyze the following Confluence content and produce structured machine-readable output. Confluence Content: {raw_text} Instructions: - If output_style is 'bullet', return bullet points summary. - If output_style is 'json', return only valid JSON array by removing un printable characters and spaces from beginning and end. - DO NOT write explanations. - DO NOT suggest code snippets. - DO NOT wrap JSON inside triple backticks \`\`\`json - Output ONLY the pure JSON array or bullet points list. Output_style: {output_style} """ # Call OpenAI again completion = await client.chat.completions.create( model="gpt-4o", messages=\[ {"role": "system", "content": "You are a helpful Program Management Data Extractor."}, {"role": "user", "content": prompt} \], temperature=0.1 ) structured_json = completion.choices\[0\].message.content.strip() return structured_json # ---- Credenciales de API de carga ---- cliente load_dotenv() = AsyncAzureOpenAI( azure_endpoint="<>", api_key=os.getenv("AZURE_API_KEY"), api_version='<>' ) chat_completion_service = OpenAIChatCompletion( ai_model_id="<>", async_client=client ) AGENT_INSTRUCTIONS = """Usted es un útil Agente de Gestión de Programas AI que puede ayudar a extraer información clave como miembro del equipo, características, Epics de una página de confluencia. Importante: Cuando los usuarios especifiquen una página de equipo, solo extraen las características y episodios de ese equipo. Cuando empiece la conversación, presentate con este mensaje: "¡Hola! soy tu asistente de PM. puedo ayudarte a obtener el estado de Funciones y Epics. Siempre llame primero 'get_confluence_page_content' para raspar la página Confluence. Si el mensaje del usuario comienza con "Team: {team_name}.", utilice ese {team_name} para el argumento 'team_name'. Por ejemplo, si la entrada es "Team: Raptor. ¿Cuáles son las últimas características?", el 'team_name' es "Raptor". 2. Si el usuario solicita un resumen, proporcione una lista de puntos de bala. 3. Si el usuario solicita un array JSON o un gráfico o un plano. Luego llame inmediatamente 'summarize_confluence_data' utilizando el contenido rasgado. 4. Basándose en el estilo de salida solicitado por el usuario, devuelva una matriz JSON o puntos de bala. 5. Si el usuario no especifica un estilo de salida, devuelva por defecto un punto de bala. 6. Si el usuario solicita un array JSON, Instrucciones: - Si output_style es 'bullet', devuelve el resumen de los puntos de bala. - Si output_style es 'json', devuelve solo la matriz JSON válida eliminando los caracteres y espacios no imprimibles del comienzo y el final. - NO escriba explicaciones. - NO sugiera fragmentos de código. - NO envuelva JSON dentro de triples backticks ``'json - Salga SÓLO la lista de matriz JSON pura o puntos de bala. What team are you interested to help you plan today?" Si mencionan un equipo específico, enfocen sus datos en ese equipo en lugar de sugerir alternativas. """ agente = ChatCompletionAgent( service=chat_completion_service, plugins=[ ConfluencePlugin() ], name="ConfluenceAgent", instrucciones=AGENT_INSTRUCTIONS ) # ---- es la lógica async principal ---- async def stream_response(user_input, thread=None): html_blocks = [] full_response = [] function_calls = [] parsed_json_result = None completion = wait client.chat.completions.create( model="gpt-4o", messages=["role": "sistema", "contenido": "Si usted es un juez del contenido de la entrada del usuario. Analize la función de la entrada del usuario. Si se pide raspar la página interna COnfluence para un equipo, entonces está relacionado con la Gestión del Programa. Si no está relacionado con la Gestión del Programa, proporcione la respuesta, pero añade 'False T' a la respuesta. Si está relacionado con la Gestión del Programa, añ async for response in agent.invoke_stream(messages=user_input, thread=thread): print("Response:", response) thread = response.thread agent_name = response.name for item in list(response.items): if isinstance(item, FunctionCallContent): pass # You can ignore this now elif isinstance(item, FunctionResultContent): if item.name == "summarize_confluence_data": raw_content = item.result extracted_json = extract_json_from_response(raw_content) if extracted_json: try: parsed_json = json.loads(extracted_json) yield parsed_json, thread, function_calls except Exception as e: st.error(f"Failed to parse extracted JSON: {e}") else: full_response.append(raw_content) else: full_response.append(item.result) elif isinstance(item, StreamingTextContent) and item.text: full_response.append(item.text) #print("Full Response:", full_response) # After loop ends, yield final result if parsed_json_result: yield parsed_json_result, thread, function_calls else: yield ''.join(full_response), thread, function_calls # ---- Streamlit UI Setup ---- st.set_page_config(layout="wide") left_col, right_col = st.columns([1, 1]) st.markdown("" <style> html, cuerpo, [class*="css"] { font-size: 12px !important; } </style> """, unsafe_allow_html=True) # ---- Main Streamlit app ---- con left_col: st.title(" Program Management Enabler AI") st.write("Pregúntame sobre diferentes elementos comprometidos del Programa Wiley.!") st.write("Puedo ayudarle a obtener el estado de Funciones y Epics.") if "history" not in st.session_state: st.session_state.history = \[\] if "thread" not in st.session_state: st.session_state.thread = None if "charts" not in st.session_state: st.session_state.charts = \[\] # Each entry: {"df": ..., "title": ..., "question": ...} if "chart_dataframes" not in st.session_state: st.session_state.chart_dataframes = \[\] if st.button("🧹 Clear Chat"): st.session_state.history = \[\] st.session_state.thread = None st.rerun() # Input box at the top user_input = st.chat_input("Ask me about your team's features...") # Example: team_selected = st.session_state.get("selected_team") if st.session_state.get("selected_team") and user_input: user_input = f"Team: {st.session_state.get('selected_team')}. {user_input}" # Preserve chat history when program or team is selected if user_input and not st.session_state.get("selected_team_changed", False): st.session_state.selected_team_changed = False if user_input: df = pd.DataFrame() full_response_holder = {"text": "","df": None} with st.chat_message("assistant"): response_container = st.empty() assistant_text = "" try: chat_index = len(st.session_state.history) response_gen = stream_response(user_input, st.session_state.thread) print("Response generator started",response_gen) async def process_stream(): async for update in response_gen: nonlocal_thread = st.session_state.thread if len(update) == 3: content, nonlocal_thread, function_calls = update full_response_holder\["text"\] = content if isinstance(content, list): data = json.loads(re.sub(r'\[\\x00-\\x1F\\x7F\]', '', content.replace("\`\`\`json", "").replace("\`\`\`",""))) df = pd.DataFrame(data) df.columns = df.columns.str.lower() print("\\n📊 Features Status Chart") st.subheader("📊 Features Status Chart") plot_bar_chart(df) st.subheader("📋 Detailed Features Table") st.dataframe(df) chart_df.columns = chart_df.columns.str.lower() full_response_holder\["df"\] = chart_df elif (re.sub(r'\[\\x00-\\x1F\\x7F\]', '', content.replace("\`\`\`json", "").replace("\`\`\`","").replace(" ",""))\[0\] =="\[" and re.sub(r'\[\\x00-\\x1F\\x7F\]', '', content.replace("\`\`\`json", "").replace("\`\`\`","").replace(" ",""))\[-1\] == "\]"): data = json.loads(re.sub(r'\[\\x00-\\x1F\\x7F\]', '', content.replace("\`\`\`json", "").replace("\`\`\`",""))) df = pd.DataFrame(data) df.columns = df.columns.str.lower() chart_df = pd.DataFrame(data) chart_df.columns = chart_df.columns.str.lower() full_response_holder\["df"\] = chart_df else: if function_calls: st.markdown("\\n".join(function_calls)) flagtext = 'text' st.session_state.thread = nonlocal_thread try: with st.spinner("🤖 AI is thinking..."): flagtext = None # Run the async function to process the stream asyncio.run(process_stream()) # Update history with the assistant's response if full_response_holder\["df"\] is not None and flagtext is None: st.session_state.chart_dataframes.append({ "question": user_input, "data": full_response_holder\["df"\], "type": "chart" }) elif full_response_holder\["text"\].strip(): # Text-type response st.session_state.history.append({ "user": user_input, "assistant": full_response_holder\["text"\], "type": "text" }) flagtext = None except Exception as e: error_msg = f"⚠️ Error: {e}" response_container.markdown(error_msg) if chat_index > 0 and "Error" in full_response_holder\["text"\]: # Remove the last message only if it was an error st.session_state.history.pop(chat_index) # Handle any exceptions that occur during the async call except Exception as e: full_response_holder\["text"\] = f"⚠️ Error: {e}" response_container.markdown(full_response_holder\["text"\]) chat_index = len(st.session_state.history) #for item in st.session_state.history\[:-1\]: for item in reversed(st.session_state.history): if item\["type"\] == "text": with st.chat_message("user"): st.markdown(item\["user"\]) with st.chat_message("assistant"): st.markdown(item\["assistant"\]) with right_col:st.title("Select Wiley Program") team_list = { "Program 1": \["Team 1", "Team 2", "Team 3"\], "Program 2": \["Team 4", "Team 5", "Team 6"\] } selected_program = st.selectbox("Select the Program:", \["No selection"\] + list(team_list.keys()), key="program_selectbox") selected_team = st.selectbox("Select the Agile Team:", \["No selection"\] + team_list.get(selected_program, \[\]), key="team_selectbox") st.session_state\["selected_team"\] = selected_team if selected_team != "No selection" else None if st.button("🧹 Clear All Charts"): st.session_state.chart_dataframes = \[\] chart_idx = 1 #if len(st.session_state.chart_dataframes) == 1: for idx, item in enumerate(st.session_state.chart_dataframes): #for idx, item in enumerate(st.session_state.chart_dataframes): st.markdown(f"\*\*Chart {idx + 1}: {item\['question'\]}\*\*") st.subheader("📊 Features Status Chart") plot_bar_chart(item\["data"\]) st.subheader("📋 Detailed Features Table") st.dataframe(item\["data"\]) chart_idx += 1 Conclusión El Ayuda a los equipos a rastrear las características del proyecto y los episodios de las páginas de Confluence. con OpenAI GPT-4o para raspar el contenido de la página Confluence específica del equipo utilizando El estado se utiliza para la autenticación. La herramienta permite la selección del Programa y del equipo relevante, y en función de la selección, se responderá a la entrada del usuario. Con la característica Agentic AI, podemos permitir que el LLM sea un verdadero asistente personal. Puede ser poderoso en limitar el acceso del LLM a los datos, pero aún aprovecha la característica LLM de datos restringidos. Es un ejemplo para comprender la característica Agentic AI y cuán poderoso puede ser. Streamlit-based Program Management AI chatbot Semantic Kernel agents Playwright La referencia: https://github.com/microsoft/ai-agents-for-beginners?tab=readme-ov-file Playwright documentation.
