Problem Statement: Declaración do problema: Con Copilot integrado nas aplicacións da organización, atopar datos raramente utilizados a partir de ficheiros, SharePoint e outras fontes accesibles fíxose incriblemente doado. Confiei moito nesta capacidade de IA de Xénero. Un día, necesitei unha visión resumida de todas as características (as entregas do equipo para un cuarto en Agile Framework) e os seus estados nos que traballa o equipo. Desafortunadamente, Copilot negou ler os datos da páxina de Confluence, o que é ideal e esperado. Moitas organizacións, proxectos e actualizacións de programas almacénanse nas páxinas de Confluence. Obter unha visión xeral dos obxectivos do equipo, as entregas, os riscos do proxecto e o estado pode ser demorado para un líder ou unha persoa que manexa múltiples programas. Pensei, por que non ter un asistente Solution: AI Agentic Assistant Powered by Streamlit + Semantic Kernel Solución: Axente Axente de IA con Streamlit + Núcleo Semántico A introdución de Agentic AI foi un salvador para min, e decidín elixir este marco como unha solución. Con todo, houbo retos: que marco debe usarse, e hai un código aberto dispoñible? ¿Canto custaría a plataforma xestionada? Finalmente, con toda a investigación, decidín ir con código aberto e usar a pila de tecnoloxía de abaixo para construír un asistente de IA lixeiro usando: para a liña de chegada ao final, Núcleo semántico para a xestión rápida e cadeaxe, Azure OpenAI para procesamento de linguaxe natural Playwright para o rascado web seguro e dinámico de páxinas de Confluence. O que fai Esta ferramenta permitirá aos xestores e líderes do 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 Como funciona - PseudoCódigo Step 1. Confluence Page LookUp. Instead of manually pasting URLs, each team name is mapped to its Confluence URL using a dictionary. When a user selects “Team A” from the right pane, the backend automatically fetches the associated URL and triggers scraping. 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, acabei usando Playwright para o rascado baseado no navegador sen cabeza, o que nos axuda a cargar contido dinámico e manexar o login: Unha aproximación fracasada: [ ]Usando a biblioteca de solicitudes de Python, obtén os datos de Confluencia usando a API. O mecanismo de autenticación non foi exitoso. [ ]Usando a biblioteca de Python BeautifySoup. Foi excluído debido ao contido dinámico. A capa SSO tivo retos, pero finalmente, funcionou despois de descargar o estado HTML JSON e reutilizalo. @kernel_function(description="Scrape and return text content from a Confluence page.") async def get_confluence_page_content(self, team_name: Annotated[str, "Nome do equipo áxil"]) -> Annotated[str, "Retorna o contido de texto extraído da páxina"): @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. A ferramenta está destinada a ser un Enabler de xestión de programas. As instrucións do axente están redactadas para usar o contido rasgado e producir un resultado axeitado para as preguntas PM. As instrucións axudarán a obter a saída como un resumo de texto ou nun formato de gráfico. Ademais, definei o cliente como un axente de IA como un plugin. Axente_instruccións = “” “” cliente = OpenAI (<Local Open Source LLM>) chat_completion_service = OpenAIChatCompletion(ai_model_id="<>", async_client=client Páxina axente = ChatCompletionAgent( servizo=chat_completion_service, plugins=[ ConfluencePlugin() ], nome="ConfluenceAgent", instrucións=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ín engadir un cliente LLM adicional para comprobar se a entrada do usuario é relevante para a xestión de programas ou non. completion = await client.chat.completions.create( model="gpt-4o", messages=[ {"role": "sistema", "contido": "Vostede é un Xuíz do contido da entrada do usuario. Analize a entrada do usuario. Se lle pide raspar a páxina de COnfluence interna para un equipo, entón está relacionada coa xestión do programa. Se non está relacionada coa xestión do programa, proporcione a resposta pero engade 'Falseḳ' á resposta. Se está relacionada coa xestión do programa, engade 'Trueḳ' á resposta."}, {"role": "usuario", "contido": 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á o código completo. Eliminei os meus detalles específicos do proxecto. Precisamos almacenar o state.json primeiro para usalo no código import json import os import asyncio import pandas como pd import streamlit como st de tipificación import Anotado de dotenv import load_dotenv de openai import AsyncAzureOpenAI de playwright.async_api import async_playwright de bs4 import BeautifulSoup de semantic_kernel.funcións import kernel_función de tipificación importado importado anotado re import matplotlibot.pyplot como plt de semantic_kernel.agents import ChatCompletionAgent de semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion de semantic_kernel.contents importation FunCallContent, FunctionResultContent, StreamingTextContent de sem 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 ---- #diagrama de barras con tamaño fixo def plot_bar_chart(df): status_counts = df["status"].value_counts() fig, ax = plt.subplots(figsize=(1.5, 1)) # anchura, altura en polgadas ax.bar(status_counts.index, status_counts.values, color="#4CAF50") ax.set_title("Features by Status") ax.set_ylabel("Count") # Cambiar a cor de tick 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 atopar a primeira matriz JSON no texto match = re.search(r"(\[\s*{.*}\s*\)", texto, re.DOTALL) se match: return match.group(1) return None Páxina oficial: DEF (self): self.default_confluence_url = "<>" load_dotenv() init @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 # ---- Credenciais de API de carga ---- load_dotenv() client = AsyncAzureOpenAI( azure_endpoint="<>", api_key=os.getenv("AZURE_API_KEY"), api_version='<>' ) chat_completion_service = OpenAIChatCompletion( ai_model_id="<>", async_client=client ) AGENT_INSTRUCTIONS = """Vostede é un axente de IA útil de xestión de programas que pode axudar a extraer información clave como membro do equipo, características, Epics dunha páxina de confluencia. Importante: Cando os usuarios especifiquen unha páxina de equipo, só extraen as características e épicos dese equipo. Cando comece a conversa, preséntate con esta mensaxe: "¡Hola! son o teu asistente de PM. Sempre primeiro chame 'get_confluence_page_content' para raspar a páxina Confluence. Se a mensaxe do usuario comeza con "Team: {team_name}.", utilice ese {team_name} para o argumento 'team_name'. Por exemplo, se a entrada é "Team: Raptor. Cales son as últimas características?", o 'team_name' é "Raptor". 2. Se o usuario pide un resumo, proporcione unha lista de puntos de bala. 3. Se o usuario pide un array ou gráfico de JSON ou un plano. A continuación, chame inmediatamente a 'summarize_confluence_data' usando o contido rascado. 4. En base ao estilo de saída solicitado polo usuario, devolva unha matriz de JSON ou puntos de bala. 5. Se o usuario non especifica un estilo de saída, por defecto, a matriz de bala. 6. Se o usuario solicita unha matriz de JSON, só devol Instrucións: - Se output_style é 'bullet', devolve un resumo de puntos de bala. - Se output_style é 'json', devolve só a matriz JSON válida eliminando os caracteres e espazos impresos do comezo e o final. - NON escriba explicacións. - NON suxire fragmentos de código. - NON envolva JSON dentro de triples backticks ```json - Saia SÓ a lista de matriz JSON pura ou de puntos de bala. Que equipo está interesado en axudarche a planificar hoxe?" Sempre priorice as preferencias do usuario. Se mencionan un equipo específico, concentre os seus datos nese equipo en vez de suxerir alternativas. """ axente = ChatCompletionAgent( servizo=chat_completion_service, plugins=[ ConfluencePlugin() ], nome="ConfluenceAgent", instrucións=AGENT_INSTRUCTIONS ) # ---- A lóxica 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( modelo="gpt-4o", mensaxes=["role": "sistema", "contido": "Se vostede é un xuíz do contido da entrada do usuario. Analize a función de entrada do usuario. Se pide para raspar a páxina interna COnfluence para un equipo, entón está relacionado coa xestión do programa. Se non está relacionado coa xestión do programa, proporcione a resposta pero engade 'False TIT' á resposta. Se está relacionado coa xestión do programa, engade 'True TIT' 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, corpo, [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 do Programa Wiley.!") st.write("Podo axudarche a obter o estado de Funcións e 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 import json import os import asyncio import pandas como pd import streamlit como st de tipificación import Anotado de dotenv import load_dotenv de openai import AsyncAzureOpenAI de playwright.async_api import async_playwright de bs4 import BeautifulSoup de semantic_kernel.funcións import kernel_función de tipificación importado importado anotado re import matplotlibot.pyplot como plt de semantic_kernel.agents import ChatCompletionAgent de semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion de semantic_kernel.contents importation FunCallContent, FunctionResultContent, StreamingTextContent de sem 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 ---- #diagrama de barras con tamaño fixo def plot_bar_chart(df): status_counts = df["status"].value_counts() fig, ax = plt.subplots(figsize=(1.5, 1)) # anchura, altura en polgadas ax.bar(status_counts.index, status_counts.values, color="#4CAF50") ax.set_title("Features by Status") ax.set_ylabel("Count") # Cambiar a cor de tick 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 atopar a primeira matriz JSON no texto match = re.search(r"(\[\s*{.*}\s*\)", texto, re.DOTALL) se match: return match.group(1) return None Páxina oficial: DEF (self): self.default_confluence_url = "<>" load_dotenv() init @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 # ---- Credenciais de API de carga ---- load_dotenv() client = AsyncAzureOpenAI( azure_endpoint="<>", api_key=os.getenv("AZURE_API_KEY"), api_version='<>' ) chat_completion_service = OpenAIChatCompletion( ai_model_id="<>", async_client=client ) AGENT_INSTRUCTIONS = """Vostede é un axente de IA útil de xestión de programas que pode axudar a extraer información clave como membro do equipo, características, Epics dunha páxina de confluencia. Importante: Cando os usuarios especifiquen unha páxina de equipo, só extraen as características e épicos dese equipo. Cando comece a conversa, preséntate con esta mensaxe: "¡Hola! son o teu asistente de PM. Sempre primeiro chame 'get_confluence_page_content' para raspar a páxina Confluence. Se a mensaxe do usuario comeza con "Team: {team_name}.", utilice ese {team_name} para o argumento 'team_name'. Por exemplo, se a entrada é "Team: Raptor. Cales son as últimas características?", o 'team_name' é "Raptor". 2. Se o usuario pide un resumo, proporcione unha lista de puntos de bala. 3. Se o usuario pide un array ou gráfico de JSON ou un plano. A continuación, chame inmediatamente a 'summarize_confluence_data' usando o contido rascado. 4. En base ao estilo de saída solicitado polo usuario, devolva unha matriz de JSON ou puntos de bala. 5. Se o usuario non especifica un estilo de saída, por defecto, a matriz de bala. 6. Se o usuario solicita unha matriz de JSON, só devol Instrucións: - Se output_style é 'bullet', devolve un resumo de puntos de bala. - Se output_style é 'json', devolve só a matriz JSON válida eliminando os caracteres e espazos impresos do comezo e o final. - NON escriba explicacións. - NON suxire fragmentos de código. - NON envolva JSON dentro de triples backticks ```json - Saia SÓ a lista de matriz JSON pura ou de puntos de bala. Que equipo está interesado en axudarche a planificar hoxe?" Sempre priorice as preferencias do usuario. Se mencionan un equipo específico, concentre os seus datos nese equipo en vez de suxerir alternativas. """ axente = ChatCompletionAgent( servizo=chat_completion_service, plugins=[ ConfluencePlugin() ], nome="ConfluenceAgent", instrucións=AGENT_INSTRUCTIONS ) # ---- A lóxica 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( modelo="gpt-4o", mensaxes=["role": "sistema", "contido": "Se vostede é un xuíz do contido da entrada do usuario. Analize a función de entrada do usuario. Se pide para raspar a páxina interna COnfluence para un equipo, entón está relacionado coa xestión do programa. Se non está relacionado coa xestión do programa, proporcione a resposta pero engade 'False TIT' á resposta. Se está relacionado coa xestión do programa, engade 'True TIT' 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, corpo, [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 do Programa Wiley.!") st.write("Podo axudarche a obter o estado de Funcións e 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 Conclusión A súa axuda aos equipos a rastrexar as características do proxecto e as épicas das páxinas de Confluence. con OpenAI GPT-4o para raspar o contido da páxina Confluence específica do equipo usando O estado utilízase para a autenticación. A ferramenta permite a selección do Programa e do equipo relevante, e en función da selección, responderase á entrada do usuario. Coa función de AI de Axente, podemos permitir que o LLM sexa un verdadeiro asistente persoal. Pode ser poderoso en limitar o acceso do LLM aos datos, pero aínda aproveitar a característica de datos restrinxidos do LLM. É un exemplo para entender a característica de AI de Axente e o poder que pode ser. Streamlit-based Program Management AI chatbot Semantic Kernel agents Playwright A referencia: https://github.com/microsoft/ai-agents-for-beginners?tab=readme-ov-file Playwright documentation.
