Como a IA está revolucionando a xestión de programas áxiles con Confluence e Streamlit

`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 team or program name from a dropdown,
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: DEFinit(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
# ---- 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: DEFinit(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

# ---- 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úaStreamlit-based Program Management AI chatbotaxuda aos equipos a rastrexar as características do proxecto e as épicas das páxinas de Confluence.Semantic Kernel agentscon OpenAI GPT-4o para raspar o contido da páxina Confluence específica do equipo usandoPlaywrightO 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.

A referencia:https://github.com/microsoft/ai-agents-for-beginners?tab=readme-ov-file

              Playwright documentation.

Como a IA está revolucionando a xestión de programas áxiles con Confluence e Streamlit

Demasiado longo; Ler

`Problem Statement:`

`Solution: AI Agentic Assistant Powered by Streamlit + Semantic Kernel`

O que fai

`How it works - Pseudo Code`

Conclusión

About Author

Etiquetas colgantes

ESTE ARTIGO FOI PRESENTADO EN...

Trending Topics

Classic

Neon Noir

Minty

Newspaper

HN StartUps

Como a IA está revolucionando a xestión de programas áxiles con Confluence e Streamlit

Demasiado longo; Ler

Problem Statement:

Solution: AI Agentic Assistant Powered by Streamlit + Semantic Kernel

O que fai

How it works - Pseudo Code

Conclusión

About Author

Etiquetas colgantes

ESTE ARTIGO FOI PRESENTADO EN...

HISTORIAS RELACIONADAS

Trending Topics

Classic

Neon Noir

Minty

Newspaper

HN StartUps

`Problem Statement:`

`Solution: AI Agentic Assistant Powered by Streamlit + Semantic Kernel`

`How it works - Pseudo Code`