Как ИИ революционизирует управление программами Agile с Confluence & Streatlit

Problem Statement:

С помощью Copilot, интегрированного в приложения организации, поиск редко используемых данных из файлов, SharePoint и других доступных источников стало невероятно простым. Я в значительной степени полагался на эту возможность искусственного интеллекта. Однажды мне понадобился краткий обзор всех функций (результаты команды за четверть в Agile Framework) и их статусы, над которыми работает команда. К сожалению, Copilot отрицал данные чтения со страницы слияния, что является идеальным и ожидаемым. Многие организации, проекты и обновления программ хранятся на страницах слияния. Получение обзора командных целей, результатов, рисков проекта и статуса может быть трудоемким для лидера или человека, занимающегося несколькими программами. Я подумал, почему бы не иметь интеллектуального помощника, чтобы получить и суммировать эту информацию? Это будет эффективным фактором для менеджера программы и старшего руководства.

Solution: AI Agentic Assistant Powered by Streamlit + Semantic Kernel

Внедрение агентского ИИ было для меня спасителем, и я решил выбрать эту структуру в качестве решения. Тем не менее, были проблемы: какую структуру следует использовать, и есть ли доступный открытый исходный код? Насколько дорогой будет управляемая платформа? Наконец, со всеми исследованиями я решил пойти с открытым исходным кодом и использовать приведенный ниже технический стек, чтобы создать легкого помощника ИИ, использующего:

• Потокдля передней части,
• Семантическое ядродля быстрого управления и цепочки,
• Azure openaiДля обработки естественного языка,
• ДраматургДля безопасного и динамического сетевого соскоба страниц слияния.

Что он делает

Этот инструмент позволит менеджерам и лидерам программ:

• ВыберитеНазвание команды или программыИз выпадения,

• Автоматически извлекать связанный URL -адрес страницы слияния,

• Секция контента Crape Key с этой страницы (например, функции, эпохи, зависимости, риски, члены команды),

• Задайте такие вопросы, как «Каковы результаты команды Q4?» или «суммировать функции, основанные на статусе» и т. Д.,

• Отображать ответы в виде суммированного текста.

How it works - Pseudo Code

Шаг 1.Поиск страницы слияния.

Вместо вручную вручную URL -адреса, каждая команда названо на карту с URL -адресом слияния с использованием словаря. Когда пользователь выбирает «команду А» из правой панели, бэкэнд автоматически приносит связанный URL -адрес и триггеры.

team_to_url_map = { "Team A": "https://confluence.company.com/display/TEAM_A",

"Team B": "https://confluence.company.com/display/TEAM_B", ... }

Шаг 2. Скрепинг веб -сайта через драматург

Этот шаг занял некоторое время. Наконец, я закончил тем, что использовал Playwright для безголовного скребки на основе браузера, который помогает нам загрузить динамическое содержание и обрабатывать вход:

Неудачный подход:-

• []Используя библиотеку запросов Python, получите данные о слиянии с помощью API. Механизм аутентификации не был успешным. В противном случае это был бы отличный способ получить данные страницы слияния.
• []Использование библиотеки Python Beautifysoup. Это было исключено из -за динамического содержания.
• []Я закончил с Python Playwright. У слоя SSO были проблемы, но, наконец, он сработал после загрузки State State JSON HTML и повторного использования.

@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"]:

Шаг 3. Определите клиента, агента и быстрого разработки с помощью семантического ядра.

Инструмент предназначен для того, чтобы стать фактором управления программой. Инструкции агента составлены для использования скрещенного контента и получения результата, подходящего для вопросов PM. Инструкции помогут получить вывод в качестве текстовой резюме или в формате диаграммы. Это также пример низкого кода.

Кроме того, я определил клиента как агента искусственного интеллекта как плагин.

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 )

Шаг 4. Определите ввод пользователя, чтобы обработать вопрос с или без инструмента.

Я решил добавить дополнительный клиент LLM, чтобы проверить, имеет ли значение пользовательский ввод для управления программой или нет.

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.

Вот весь код. Я удалил свои детали, специфичные для проекта. Нам нужно хранить State.json в первую очередь, чтобы использовать его в коде

import json import os import asyncio import pandas as pd import streamlit as st from typing import Annotated from dotenv import load_dotenv from openai import AsyncAzureOpenAI from playwright.async_api import async_playwright from bs4 import BeautifulSoup from semantic_kernel.functions import kernel_function from typing import Annotated import re import matplotlib.pyplot as plt от semantic_kernel.agents import chatcompletionagent from semantic_kernel.connectors.ai.open_ai import openaichatcompletion из semantic_kernel.contents importcontent, функции, функциональный

Team_url_mapping = {«Team 1»: «URL Confluence для команды 1», «Команда 2»: «URL Confluence для команды 2», «Команда 3»: «URL Confluence для команды 3», «Команда 4»: «Confluence URL для команды 4», «Команда 5»: «Contruence Url для команды 5», «Команда 6»: «Confluence URL для команды 6« Команда 6 «Команда 6« Команда 6 «Команда 6« Команда 6 «Команда 6« Команда 6 «Команда 6« Команда 6 «6».

# ---- Определение плагина ----

#BAR Диаграмма с фиксированным размером def plot_bar_chart (df): status_counts = df ["status"]. Value_counts () Fig, ax = plt.subplots (figsize = (1,5, 1)) #Ширина, высота в дюймах Ax.bar (status_counts.index, status_counts.values, #4caf50 "ax.setex. ax.set_ylabel ("count") # Изменить цветовой ax.tick_params (axis = 'x', colors = 'blue', labelrotation = 90) # x-тик в синем, вращающийся Ax.tick_params (Axis = 'y', Colors = 'green') # Y-Ticks в зеленой St.pryplot (Fig)

DEF EXTRACT_JSON_FROM_RESPONSE (TEXT): # Используйте REGEX, чтобы найти первый массив JSON в текстовом совпадении = re.search (r "(\ [\ s*{.*} \ s*\])", text, re.dotall), если совпадает: return match.group (1) note none

Class ConfluencePlugin: defинициатор(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

# ---- Загрузить учетные данные API ---- load_dotenv () client = asyncazureopenai (azure_endpoint = "<>", api_key = os.getenv ("azure_api_key"), api_version = '<>') cath_completion_service = openaichatcom async_client = client)

Agent_instructions = "" "Вы - полезный агент по управлению программами, который может помочь извлечь ключевую информацию, такую ​​как член команды, функции, Epics со страницы слияния.

ВАЖНО: Когда пользователи указывают страницу команды, только извлекают функции и эпосы этой команды.

Когда начнется разговор, представитесь этим сообщением: «Здравствуйте! Я ваш помощник по премьере. Я могу помочь вам получить статус функций и эпосов.

Шаги, которые вы должны следовать: 1. Всегда сначала вызовите `get_confluence_page_content`, чтобы скрепить страницу слияния.

• Если сообщение пользователя начинается с "team: {team_name}.", Используйте это {team_name} для аргумента `team_name`. Например, если ввод - «Команда: Raptor. Каковы последние функции?», «Team_Name» - это «Raptor». 2. Если пользователь просит краткую информацию, предоставьте список пулевых баллов. 3. Если пользователь запрашивает массив JSON, диаграмму или сюжет. Затем немедленно вызовите `Summarize_confluence_data` с помощью сохраненного контента. 4. На основании стиля вывода, запрашиваемого пользователем, верните либо массив JSON, либо точки пуль. 5. Если пользователь не указывает стиль вывода, по умолчанию сводку с пулевой точкой. 6. Если пользователь запрашивает массив JSON, верните только действительный JSON и сюжетный график/график.

Инструкции: - Если output_style - это «пуля», вернуть пуль -баллы. - Если output_style - это «json», верните только действительный массив JSON, удалив неватные символы и пространства с начала и конца. - Не пишите объяснения. - Не предлагайте фрагменты кода. - Не оберните JSON в Triple Backticks `` `json - выводит только список чистого массива JSON или пуль.

Какую команду вы заинтересованы, чтобы помочь вам планировать сегодня? "

Всегда расставляйте приоритеты пользовательских предпочтений. Если они упоминают конкретную команду, сосредоточьте свои данные на этой команде, а не на предложении альтернатив. "" "Agent = ChatCompletionAGENT (service = CHAT_COMPLETION_SERVICE, PLUGINS = [confluencePlugin ()], name =" confluenceagent ", инструкции = Agent_instructions)

# ---- Main Async Logic ---- Async def Stream_response (user_input, thread = none): html_blocks = [] full_response = [] function_calls = [] parsed_json_result = none opplion = await client.chat.completions.create (model = "gpt-4o", a await.chat.completions.create (model = "gpt-4o", ",", ",", ",", ",", ",", ",", ",", ",", ",", ",", ",", ",", ",", ",", ",", ",", ",", ",", ". «Вы являетесь судьей содержания пользовательского ввода. opplion.choices [0] .message.content.strip () print («Текст ответа:», response_text) if response_text.startswith ("false |"): response_text = response_text [6:] witch response_text, thread, [] # return response без какого -либо дальнейшего обработки response_text. re.findall (r "(\ w+\ (.*? \))", response_text) # Удалить вызовы функции из текста ответа для вызова в функции_каллов: response_text = response_text.replace (call, "") # Если ответ связан с управлением программой, продолжить обработку.

    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

# ---- STREADLIT UI SETUP ---- st.set_page_config (layout = "wide") Leat_col, right_col = st.columns ([1, 1]) St.markdown ("" <style> html, тело, [класс*= "css"] {font-size: 12px! ---- Основное приложение для потокового часа ---- с левой_кол: St.title («💬 💬 управление программами Ai Ai») St.Write («Спросите меня о различных коммерческих предметах Wiley.!») St.Write («Я могу помочь вам получить статус функций и эпосов».

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"\])

с right_col: St.title («Выберите программу Wiley»)

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

Заключение

ААйбот управления программами на основе потоковПомогает командам отслеживать функции проекта и эпосии со страниц слияния. Приложение используетСемантические агенты ядрас OpenAI GPT-4O, чтобы соскребить контент страницы слияния команды, используя страницу с использованиемДраматургПолем Государство используется для аутентификации. Инструмент позволяет выбирать программу и соответствующую команду, и, основываясь на выборе, вход пользователя будет отвечать. С помощью агентской функции ИИ мы можем позволить LLM быть настоящим личным помощником. Это может быть мощным в ограничении доступа LLM к данным, но все же использовать функцию LLM ограниченных данных. Это пример, чтобы понять агентскую функцию ИИ и насколько мощной она может быть.

Ссылка:https://github.com/microsoft/ai-agents-for-beginners?tab=readme-ov-file

              Playwright documentation.